Differential D6273 Diff 23009 swh/provenance/api/server.py

Changeset View

Standalone View

swh/provenance/api/server.py

Show All 17 Lines

import pika		import pika
import pika.channel		import pika.channel
import pika.connection		import pika.connection
import pika.exceptions		import pika.exceptions
from pika.exchange_type import ExchangeType		from pika.exchange_type import ExchangeType
import pika.frame		import pika.frame
import pika.spec		import pika.spec
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.core.api.serializers import encode_data_client as encode_data		from swh.core.api.serializers import encode_data_client as encode_data
from swh.core.api.serializers import msgpack_loads as decode_data		from swh.core.api.serializers import msgpack_loads as decode_data
from swh.model.model import Sha1Git		from swh.model.model import Sha1Git
from swh.provenance import get_provenance_storage		from swh.provenance import get_provenance_storage
from swh.provenance.interface import (		from swh.provenance.interface import (
EntityType,		EntityType,
ProvenanceStorageInterface,		ProvenanceStorageInterface,
RelationData,		RelationData,
RelationType,		RelationType,
RevisionData,		RevisionData,
)		)

from .serializers import DECODERS, ENCODERS		from .serializers import DECODERS, ENCODERS

LOG_FORMAT = (		LOG_FORMAT = (
"%(levelname) -10s %(asctime)s %(name) -30s %(funcName) "		"%(levelname) -10s %(asctime)s %(name) -30s %(funcName) "
"-35s %(lineno) -5d: %(message)s"		"-35s %(lineno) -5d: %(message)s"
)		)
LOGGER = logging.getLogger(__name__)		LOGGER = logging.getLogger(__name__)

TERMINATE = object()		TERMINATE = object()

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 ProvenanceStorageRPCServerApp(RPCServerApp):
extra_type_decoders = DECODERS
extra_type_encoders = ENCODERS


app = ProvenanceStorageRPCServerApp(
__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(
config_path: Optional[str], type: str = "local"
) -> Dict[str, Any]:
"""Check the minimal configuration is set to run the api or raise an
error explanation.

Args:
config_path (str): Path to the configuration file to load
type (str): configuration type. For 'local' type, more
checks are done.

Raises:
Error if the setup is not as expected

Returns:
configuration as a dict

"""
if config_path is None:
raise EnvironmentError("Configuration file must be defined")

if not os.path.exists(config_path):
raise FileNotFoundError(f"Configuration file {config_path} does not exist")

cfg = config.read(config_path)

pcfg: Optional[Dict[str, Any]] = cfg.get("provenance")
if pcfg is None:
raise KeyError("Missing 'provenance' configuration")

scfg: Optional[Dict[str, Any]] = pcfg.get("storage")
if scfg is None:
raise KeyError("Missing 'provenance.storage' configuration")

if type == "local":
cls = scfg.get("cls")
if cls != "postgresql":
raise ValueError(
"The provenance backend can only be started with a 'postgresql' "
"configuration"
)

db = scfg.get("db")
if not db:
raise KeyError("Invalid configuration; missing 'db' config entry")

return cfg


api_cfg: Optional[Dict[str, Any]] = None


def make_app_from_configfile() -> ProvenanceStorageRPCServerApp:
"""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


class ServerCommand(Enum):		class ServerCommand(Enum):
TERMINATE = "terminate"		TERMINATE = "terminate"
CONSUMING = "consuming"		CONSUMING = "consuming"


class TerminateSignal(BaseException):		class TerminateSignal(BaseException):
pass		pass
▲ Show 20 Lines • Show All 723 Lines • ▼ Show 20 Lines	):
),		),
body=encode_data(		body=encode_data(
response,		response,
extra_encoders=ProvenanceStorageRabbitMQServer.extra_type_encoders,		extra_encoders=ProvenanceStorageRabbitMQServer.extra_type_encoders,
),		),
)		)


		def load_and_check_config(
		config_path: Optional[str], type: str = "local"
		) -> Dict[str, Any]:
		"""Check the minimal configuration is set to run the api or raise an
		error explanation.

		Args:
		config_path (str): Path to the configuration file to load
		type (str): configuration type. For 'local' type, more
		checks are done.

		Raises:
		Error if the setup is not as expected

		Returns:
		configuration as a dict

		"""
		if config_path is None:
		raise EnvironmentError("Configuration file must be defined")

		if not os.path.exists(config_path):
		raise FileNotFoundError(f"Configuration file {config_path} does not exist")

		cfg = config.read(config_path)

		pcfg: Optional[Dict[str, Any]] = cfg.get("provenance")
		if pcfg is None:
		raise KeyError("Missing 'provenance' configuration")

		scfg: Optional[Dict[str, Any]] = pcfg.get("storage")
		if scfg is None:
		raise KeyError("Missing 'provenance.storage' configuration")

		if type == "local":
		cls = scfg.get("cls")
		if cls != "postgresql":
		raise ValueError(
		"The provenance backend can only be started with a 'postgresql' "
		"configuration"
		)

		db = scfg.get("db")
		if not db:
		raise KeyError("Invalid configuration; missing 'db' config entry")

		return cfg


def make_server_from_configfile() -> ProvenanceStorageRabbitMQServer:		def make_server_from_configfile() -> ProvenanceStorageRabbitMQServer:
config_path = os.environ.get("SWH_CONFIG_FILENAME")		config_path = os.environ.get("SWH_CONFIG_FILENAME")
server_cfg = load_and_check_config(config_path)		server_cfg = load_and_check_config(config_path)
return ProvenanceStorageRabbitMQServer(		return ProvenanceStorageRabbitMQServer(
url=server_cfg["provenance"]["rabbitmq"]["url"],		url=server_cfg["provenance"]["rabbitmq"]["url"],
storage_config=server_cfg["provenance"]["storage"],		storage_config=server_cfg["provenance"]["storage"],
)		)