Changeset View
Changeset View
Standalone View
Standalone View
swh/provenance/api/server.py
# Copyright (C) 2021 The Software Heritage developers | # Copyright (C) 2021 The Software Heritage developers | ||||
# See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
# License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
# See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
import logging | |||||
import os | import os | ||||
from typing import Any, Dict, List, Optional | from typing import Any, Dict, Optional | ||||
from werkzeug.routing import Rule | |||||
from swh.core import config | from swh.core import config | ||||
from swh.core.api import JSONFormatter, MsgpackFormatter, RPCServerApp, negotiate | |||||
from swh.provenance import get_provenance_storage | |||||
from swh.provenance.interface import ProvenanceStorageInterface | |||||
from .serializers import DECODERS, ENCODERS | |||||
storage: Optional[ProvenanceStorageInterface] = None | |||||
def get_global_provenance_storage() -> ProvenanceStorageInterface: | |||||
global storage | |||||
if storage is None: | |||||
storage = get_provenance_storage(**app.config["provenance"]["storage"]) | |||||
return storage | |||||
class ProvenanceStorageServerApp(RPCServerApp): | |||||
extra_type_decoders = DECODERS | |||||
extra_type_encoders = ENCODERS | |||||
app = ProvenanceStorageServerApp( | |||||
__name__, | |||||
backend_class=ProvenanceStorageInterface, | |||||
backend_factory=get_global_provenance_storage, | |||||
) | |||||
def has_no_empty_params(rule: Rule) -> bool: | |||||
return len(rule.defaults or ()) >= len(rule.arguments or ()) | |||||
@app.route("/") | |||||
def index() -> str: | |||||
return """<html> | |||||
<head><title>Software Heritage provenance storage RPC server</title></head> | |||||
<body> | |||||
<p>You have reached the | |||||
<a href="https://www.softwareheritage.org/">Software Heritage</a> | |||||
provenance storage RPC server.<br /> | |||||
See its | |||||
<a href="https://docs.softwareheritage.org/devel/swh-provenance/">documentation | |||||
and API</a> for more information</p> | |||||
</body> | |||||
</html>""" | |||||
@app.route("/site-map") | |||||
@negotiate(MsgpackFormatter) | |||||
@negotiate(JSONFormatter) | |||||
def site_map() -> List[Dict[str, Any]]: | |||||
links = [] | |||||
for rule in app.url_map.iter_rules(): | |||||
if has_no_empty_params(rule) and hasattr( | |||||
ProvenanceStorageInterface, rule.endpoint | |||||
): | |||||
links.append( | |||||
dict( | |||||
rule=rule.rule, | |||||
description=getattr( | |||||
ProvenanceStorageInterface, rule.endpoint | |||||
).__doc__, | |||||
) | |||||
) | |||||
# links is now a list of url, endpoint tuples | |||||
return links | |||||
def load_and_check_config( | def load_and_check_config( | ||||
config_path: Optional[str], type: str = "local" | config_path: Optional[str], type: str = "local" | ||||
) -> Dict[str, Any]: | ) -> Dict[str, Any]: | ||||
"""Check the minimal configuration is set to run the api or raise an | """Check the minimal configuration is set to run the api or raise an | ||||
error explanation. | error explanation. | ||||
Show All 33 Lines | if type == "local": | ||||
"configuration" | "configuration" | ||||
) | ) | ||||
db = scfg.get("db") | db = scfg.get("db") | ||||
if not db: | if not db: | ||||
raise KeyError("Invalid configuration; missing 'db' config entry") | raise KeyError("Invalid configuration; missing 'db' config entry") | ||||
return cfg | return cfg | ||||
api_cfg: Optional[Dict[str, Any]] = None | |||||
def make_app_from_configfile() -> ProvenanceStorageServerApp: | |||||
"""Run the WSGI app from the webserver, loading the configuration from | |||||
a configuration file. | |||||
SWH_CONFIG_FILENAME environment variable defines the | |||||
configuration path to load. | |||||
""" | |||||
global api_cfg | |||||
if api_cfg is None: | |||||
config_path = os.environ.get("SWH_CONFIG_FILENAME") | |||||
api_cfg = load_and_check_config(config_path) | |||||
app.config.update(api_cfg) | |||||
handler = logging.StreamHandler() | |||||
app.logger.addHandler(handler) | |||||
return app |