227 lines
6.8 KiB
Python
227 lines
6.8 KiB
Python
import os
|
|
from typing import List, Dict
|
|
import traceback
|
|
|
|
from flask import Flask, render_template
|
|
from flask_restful import Resource, Api, reqparse, abort
|
|
from flask_cors import CORS
|
|
|
|
from app.glob import get_config, get_manager
|
|
from app.manager import infobase_data_blank
|
|
from app.utils import validate_url
|
|
from brackets import get_infobases as br_get_infobases
|
|
|
|
apache_restart_flagfile = 'restart_apache'
|
|
|
|
pub_parser = reqparse.RequestParser()
|
|
pub_parser.add_argument('url', type=str)
|
|
pub_parser.add_argument('file', type=str, default='')
|
|
pub_parser.add_argument('force', type=bool, default=False)
|
|
|
|
pub_parser_with_name = pub_parser.copy()
|
|
pub_parser_with_name.add_argument('name', required=True, type=str, help='name required')
|
|
|
|
url_req_parser = reqparse.RequestParser()
|
|
url_req_parser.add_argument('url', type=str, required=True)
|
|
|
|
remove_parser = reqparse.RequestParser()
|
|
remove_parser.add_argument('force', type=bool, default=False)
|
|
|
|
|
|
def load_infobases(config) -> List[str]:
|
|
result = []
|
|
if 'infobases' in config and 'server_file' in config['infobases']:
|
|
result = br_get_infobases(config['infobases']['server_file'])
|
|
return result
|
|
|
|
|
|
class InfobasesAvailable(Resource):
|
|
def get(self) -> List[str]:
|
|
cfg = get_config()
|
|
bases = load_infobases(cfg)
|
|
return bases
|
|
|
|
|
|
class Config(Resource):
|
|
def get(self):
|
|
return get_config()
|
|
|
|
|
|
class ConfigTest(Resource):
|
|
def get(self):
|
|
manager = get_manager()
|
|
return manager.check()
|
|
|
|
|
|
class InfobasesAll(Resource):
|
|
def get(self) -> List[Dict[str, str]]:
|
|
cfg = get_config()
|
|
manager = get_manager()
|
|
infobase_names = load_infobases(cfg)
|
|
pubs = manager.publications()
|
|
result: List[Dict[str, str]] = []
|
|
for name in infobase_names:
|
|
publicated = False
|
|
for pub in filter(lambda x: x["name"] == name, pubs):
|
|
publicated = True
|
|
result.append(pub)
|
|
break
|
|
if not publicated:
|
|
result.append(infobase_data_blank(name))
|
|
return result
|
|
|
|
|
|
class Publications(Resource):
|
|
def get(self) -> List[Dict[str, str]]:
|
|
manager = get_manager()
|
|
return manager.publications()
|
|
|
|
def put(self):
|
|
args = pub_parser_with_name.parse_args()
|
|
validate_url(args.url)
|
|
manager = get_manager()
|
|
url = args.url
|
|
if not args.force and args.name in manager.list():
|
|
abort(409, message=f'publication exists: {args.name}')
|
|
try:
|
|
url = manager.add(args.name, url, args.file, args.force)
|
|
except Exception as e:
|
|
abort(422, message=f'publication failed: {args.name}', traceback=traceback.format_exc())
|
|
return {'message': 'created', 'name': args.name, 'url': url}, 201
|
|
|
|
|
|
class Publication(Resource):
|
|
def get(self, name: str):
|
|
manager = get_manager()
|
|
if name not in manager.list():
|
|
abort(404, message=f'publication not found: {name}')
|
|
return {
|
|
name: manager.get(name)
|
|
}
|
|
|
|
def put(self, name: str):
|
|
args = pub_parser.parse_args()
|
|
validate_url(args.url)
|
|
manager = get_manager()
|
|
url = args.url
|
|
if not args.force and name in manager.list():
|
|
abort(409, message=f'publication exists: {name}')
|
|
try:
|
|
url = manager.add(name, url, args.file, args.force)
|
|
except Exception as e:
|
|
abort(422, message=f'publication failed: {name}', traceback=traceback.format_exc())
|
|
return {'message': 'created', 'name': name, 'url': url}, 201
|
|
|
|
def delete(self, name: str):
|
|
args = remove_parser.parse_args()
|
|
manager = get_manager()
|
|
if name not in manager.list():
|
|
abort(404, message=f'publication not found: {name}')
|
|
try:
|
|
manager.remove(name, args.force)
|
|
except Exception as e:
|
|
abort(422, message='delete failed', traceback=traceback.format_exc())
|
|
return {'message': 'deleted', 'name': name}
|
|
|
|
|
|
class PublicationURL(Resource):
|
|
def get(self, name: str):
|
|
manager = get_manager()
|
|
if name not in manager.list():
|
|
abort(404, message=f'publication not found: {name}')
|
|
return {
|
|
'url': manager.get(name)['url']
|
|
}
|
|
|
|
def post(self, name: str):
|
|
args = url_req_parser.parse_args()
|
|
validate_url(args.url)
|
|
manager = get_manager()
|
|
url = args.url
|
|
if name not in manager.list():
|
|
abort(404, message=f'publication not found: {name}')
|
|
try:
|
|
manager.set_url(name, url)
|
|
except Exception as e:
|
|
abort(422, message=f'set url failed', traceback=traceback.format_exc())
|
|
return {'message': 'success', 'name': name, 'url': url}
|
|
|
|
|
|
class EnterpriseModule(Resource):
|
|
def get(self):
|
|
manager = get_manager()
|
|
if not manager.has_module():
|
|
abort(404, message='not found')
|
|
return {
|
|
'message': 'found'
|
|
}
|
|
|
|
def put(self):
|
|
manager = get_manager()
|
|
if manager.has_module():
|
|
return '', 304
|
|
manager.add_module()
|
|
return {
|
|
'message': 'success'
|
|
}
|
|
|
|
|
|
class ApacheRestartFlag(Resource):
|
|
def get(self):
|
|
cfg = get_config()
|
|
flagfile = cfg.get('apache_restart_flagfile', apache_restart_flagfile)
|
|
if not os.path.isfile(flagfile):
|
|
abort(404, message='not found')
|
|
return {
|
|
'message': 'found'
|
|
}
|
|
|
|
def put(self):
|
|
cfg = get_config()
|
|
flagfile = cfg.get('apache_restart_flagfile', apache_restart_flagfile)
|
|
if os.path.isfile(flagfile):
|
|
return '', 304
|
|
with open(flagfile, 'a'):
|
|
pass
|
|
return {
|
|
'message': 'success'
|
|
}
|
|
|
|
|
|
class APIIndex(Resource):
|
|
def get(self):
|
|
return ['infobases-available', 'infobases-all', 'publications',
|
|
'module', 'config', 'config-test', 'apache-restart']
|
|
|
|
|
|
frontend_dir = 'frontend/dist'
|
|
app = Flask(__name__, static_url_path='/', static_folder=frontend_dir, template_folder=frontend_dir)
|
|
api = Api(app, '/api/v1/')
|
|
cors = CORS(app, resources={r"/api/*": {"origins": "*"}})
|
|
|
|
api.add_resource(InfobasesAvailable, '/infobases-available')
|
|
api.add_resource(InfobasesAll, '/infobases-all')
|
|
api.add_resource(Publications, '/publications', '/publications/')
|
|
api.add_resource(Publication, '/publications/<string:name>')
|
|
api.add_resource(PublicationURL, '/publications/<string:name>/url')
|
|
api.add_resource(EnterpriseModule, '/module')
|
|
api.add_resource(Config, '/config')
|
|
api.add_resource(ConfigTest, '/config-test')
|
|
api.add_resource(ApacheRestartFlag, '/apache-restart')
|
|
api.add_resource(APIIndex, '/')
|
|
|
|
|
|
@app.route('/')
|
|
def index():
|
|
return index_content
|
|
|
|
|
|
with app.app_context():
|
|
get_config()
|
|
get_manager()
|
|
index_content = render_template("index.html")
|
|
|
|
|
|
if __name__ == '__main__':
|
|
app.run(debug=True)
|