pub1c-web/pub1c-rest.py

372 lines
11 KiB
Python
Raw Normal View History

2021-05-21 12:19:11 +00:00
import logging
import os
2021-05-31 13:08:10 +00:00
from typing import List, Optional, Dict
2021-05-21 12:19:11 +00:00
import traceback
import yaml
2021-06-15 14:04:36 +00:00
from flask import Flask, g, render_template
2021-05-21 12:19:11 +00:00
from flask_restful import Resource, Api, reqparse, abort
2021-05-31 13:08:10 +00:00
from flask_cors import CORS
2021-05-21 12:19:11 +00:00
from pathvalidate import is_valid_filepath
from brackets import get_infobases as br_get_infobases
2021-05-31 13:08:10 +00:00
from webpub1c.webpub1c import VRDConfig, ApacheConfig, urlpath_join, WebPublication
2021-05-21 12:19:11 +00:00
apache_restart_flagfile = 'restart_apache'
2021-05-31 13:08:10 +00:00
def infobase_data_blank(name: str):
return {
'name': name,
'publicated': False,
'url': '',
'directory': '',
'vrd_filename': '',
}
def publication_data(publication: WebPublication):
return {
'name': publication.name,
'publicated': True,
'url': publication.url_path,
'directory': publication.directory,
'vrd_filename': publication.vrd_filename,
}
2021-05-21 12:19:11 +00:00
class WebPub1C:
def __init__(self, config, verbose: bool = False):
level = logging.INFO if verbose else logging.WARNING
logging.basicConfig(level=level)
self._log = logging.getLogger("webpub1c")
self._log.setLevel(level)
self._config = config
vrd_params: Optional[VRDConfig] = self._config.get('vrd_params', None)
apache_config: str = self._config.get('apache_config', '')
self._vrd_path: str = self._config.get('vrd_path', '')
self._dir_path: str = self._config.get('dir_path', '')
self._url_base: str = self._config.get('url_base', '')
self._apache_cfg = ApacheConfig(apache_config, self._vrd_path,
self._dir_path, self._url_base,
vrd_params)
def _is_vrd_path_valid(self) -> bool:
return os.path.isdir(self._vrd_path)
def _is_dir_path_valid(self) -> bool:
return os.path.isdir(self._dir_path)
def _is_url_base_valid(self) -> bool:
return is_valid_filepath(self._url_base, platform='posix') and self._url_base.startswith('/')
def _is_module_valid(self) -> bool:
if 'platform_path' not in self._config:
return False
if 'ws_module' not in self._config:
return False
return os.path.isfile(os.path.join(self._config['platform_path'], self._config['ws_module']))
def check(self):
return {
'is_apache_cfg_valid': self._apache_cfg.is_valid(),
'is_vrd_path_valid': self._is_vrd_path_valid(),
'is_dir_path_valid': self._is_dir_path_valid(),
'is_url_base_valid': self._is_url_base_valid(),
'is_module_valid': self._is_module_valid()
}
def has_module(self):
return self._apache_cfg.has_1cws_module()
def add_module(self):
""" Add 1cws module to apache config """
if self._apache_cfg.has_1cws_module():
self._log.info('config unchanged')
else:
module: str = os.path.join(self._config['platform_path'], self._config['ws_module'])
self._apache_cfg.add_1cws_module(module)
self._log.info('module added')
def list(self) -> List[str]:
2021-05-31 13:08:10 +00:00
""" List publication names """
2021-05-21 12:19:11 +00:00
return list(self._apache_cfg.publications)
2021-05-31 13:08:10 +00:00
def publications(self) -> List[Dict[str, str]]:
""" List of publications """
pubs = map(lambda p: publication_data(p), self._apache_cfg.iter())
return list(pubs)
2021-05-21 12:19:11 +00:00
def get(self, ibname: str):
""" Get publication info """
publication = self._apache_cfg.get_publication(ibname)
if publication is None:
return publication
2021-05-31 13:08:10 +00:00
return publication_data(publication)
2021-05-21 12:19:11 +00:00
def add(self, ibname: str, url: Optional[str] = None) -> str:
""" Add new publication """
publication = self._apache_cfg.create_publication(ibname, url)
self._apache_cfg.add_publication(publication)
self._log.info(f'publication added: {ibname}')
return publication.url_path
def set_url(self, ibname: str, url: str) -> None:
""" Set publication url """
publication = self._apache_cfg.get_publication(ibname)
if publication is None:
raise KeyError(f'infobase "{ibname}" not publicated')
publication.url_path = urlpath_join(self._url_base, url)
self._apache_cfg.remove_publication(publication.name, destroy=False)
self._apache_cfg.add_publication(publication)
self._log.info(f'publication changed: {ibname}')
def remove(self, ibname: str):
""" Remove publication """
self._apache_cfg.remove_publication(ibname)
self._log.info(f'publication removed: {ibname}')
pub_parser = reqparse.RequestParser()
pub_parser.add_argument('name', required=True, type=str, help='name required')
pub_parser.add_argument('url', type=str)
url_parser = reqparse.RequestParser()
url_parser.add_argument('url', type=str)
url_req_parser = reqparse.RequestParser()
url_req_parser.add_argument('url', type=str, required=True)
def load_config(filename: str = 'config.yml'):
with open(filename, 'r', encoding='utf-8') as cfg_file:
return yaml.safe_load(cfg_file)
def get_config():
if 'config' not in g:
g.config = load_config()
return g.config
def get_webpub1c():
if 'webpub1c' not in g:
g.webpub1c = WebPub1C(get_config())
return g.webpub1c
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
def validate_url(url: Optional[str]):
if url is not None:
if not is_valid_filepath(url, platform='posix'):
abort(400, message='invalid url')
2021-05-31 13:08:10 +00:00
class InfobasesAvailable(Resource):
2021-05-21 12:19:11 +00:00
def get(self) -> List[str]:
cfg = get_config()
bases = load_infobases(cfg)
return bases
2021-05-31 13:08:10 +00:00
class Config(Resource):
def get(self):
return get_config()
2021-05-21 12:19:11 +00:00
class ConfigTest(Resource):
def get(self):
webpub = get_webpub1c()
return webpub.check()
2021-05-31 13:08:10 +00:00
class InfobasesAll(Resource):
def get(self) -> List[Dict[str, str]]:
cfg = get_config()
webpub = get_webpub1c()
infobase_names = load_infobases(cfg)
pubs = webpub.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
2021-05-21 12:19:11 +00:00
class Publications(Resource):
2021-05-31 13:08:10 +00:00
def get(self) -> List[Dict[str, str]]:
2021-05-21 12:19:11 +00:00
webpub = get_webpub1c()
2021-05-31 13:08:10 +00:00
return webpub.publications()
2021-05-21 12:19:11 +00:00
def put(self):
args = pub_parser.parse_args()
validate_url(args.url)
webpub = get_webpub1c()
url = args.url
if args.name in webpub.list():
abort(409, message=f'publication exists: {args.name}')
try:
url = webpub.add(args.name, url)
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):
webpub = get_webpub1c()
if name not in webpub.list():
abort(404, message=f'publication not found: {name}')
return {
name: webpub.get(name)
}
def put(self, name: str):
args = url_parser.parse_args()
validate_url(args.url)
webpub = get_webpub1c()
url = args.url
if name in webpub.list():
abort(409, message=f'publication exists: {name}')
try:
url = webpub.add(name, url)
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):
webpub = get_webpub1c()
if name not in webpub.list():
abort(404, message=f'publication not found: {name}')
try:
webpub.remove(name)
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):
webpub = get_webpub1c()
if name not in webpub.list():
abort(404, message=f'publication not found: {name}')
return {
'url': webpub.get(name)['url']
}
def post(self, name: str):
args = url_req_parser.parse_args()
validate_url(args.url)
webpub = get_webpub1c()
url = args.url
if name not in webpub.list():
abort(404, message=f'publication not found: {name}')
try:
webpub.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):
webpub = get_webpub1c()
if not webpub.has_module():
abort(404, message='not found')
return {
'message': 'found'
}
def put(self):
webpub = get_webpub1c()
if webpub.has_module():
abort(304, message='already added')
webpub.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):
abort(304, message='found')
with open(flagfile, 'a'):
pass
return {
'message': 'success'
}
class APIIndex(Resource):
def get(self):
2021-05-31 13:08:10 +00:00
return ['infobases-available', 'infobases-all', 'publications',
'module', 'config', 'config-test', 'apache-restart']
2021-05-21 12:19:11 +00:00
2021-06-15 14:04:36 +00:00
frontend_dir = 'frontend/dist'
app = Flask(__name__, static_url_path='/', static_folder=frontend_dir, template_folder=frontend_dir)
2021-05-21 12:19:11 +00:00
api = Api(app, '/api/v1/')
2021-05-31 13:08:10 +00:00
cors = CORS(app, resources={r"/api/*": {"origins": "*"}})
2021-05-21 12:19:11 +00:00
2021-05-31 13:08:10 +00:00
api.add_resource(InfobasesAvailable, '/infobases-available')
api.add_resource(InfobasesAll, '/infobases-all')
api.add_resource(Publications, '/publications', '/publications/')
2021-05-21 12:19:11 +00:00
api.add_resource(Publication, '/publications/<string:name>')
api.add_resource(PublicationURL, '/publications/<string:name>/url')
api.add_resource(EnterpriseModule, '/module')
2021-05-31 13:08:10 +00:00
api.add_resource(Config, '/config')
api.add_resource(ConfigTest, '/config-test')
api.add_resource(ApacheRestartFlag, '/apache-restart')
2021-05-21 12:19:11 +00:00
api.add_resource(APIIndex, '/')
2021-06-15 14:04:36 +00:00
@app.route('/')
def index():
return index_content
2021-05-21 12:19:11 +00:00
with app.app_context():
get_config()
get_webpub1c()
2021-06-15 14:04:36 +00:00
index_content = render_template("index.html")
2021-05-21 12:19:11 +00:00
if __name__ == '__main__':
app.run(debug=True)