swh.storage.api.server module

swh.storage.api.server.get_storage()

class swh.storage.api.server.StorageServerApp(*args, backend_class=None, backend_factory=None, **kwargs)

Bases: RPCServerApp

extra_type_decoders: Dict[str, Callable] = {'blocking_status': <function _decode_blocking_status>, 'branch_by_name_response': <function _decode_snapshot_branch_by_name_response>, 'core_swhid': <bound method _BaseSWHID.from_string of <class 'swh.model.swhids.CoreSWHID'>>, 'extended_swhid': <bound method _BaseSWHID.from_string of <class 'swh.model.swhids.ExtendedSWHID'>>, 'identifiers_enum': <function _decode_swhids_enum>, 'masked_status': <function _decode_masked_status>, 'model': <function <lambda>>, 'model_enum': <function _decode_model_enum>, 'object_reference': <function _decode_object_reference>, 'origin_visit_with_statuses': <function _decode_origin_visit_with_statuses>, 'qualified_swhid': <bound method QualifiedSWHID.from_string of <class 'swh.model.swhids.QualifiedSWHID'>>, 'storage_enum': <function _decode_storage_enum>, 'swhids_enum': <function _decode_swhids_enum>}

Value of extra_decoders passed to json_loads or msgpack_loads to be able to deserialize more object types.

extra_type_encoders: List[Tuple[type, str, Callable]] = [(<class 'swh.model.model.BaseModel'>, 'model', <function _encode_model_object>), (<class 'swh.model.swhids.CoreSWHID'>, 'core_swhid', <class 'str'>), (<class 'swh.model.swhids.ExtendedSWHID'>, 'extended_swhid', <class 'str'>), (<class 'swh.model.swhids.QualifiedSWHID'>, 'qualified_swhid', <class 'str'>), (<enum 'ObjectType'>, 'identifiers_enum', <function _encode_enum>), (<enum 'MetadataAuthorityType'>, 'model_enum', <function _encode_enum>), (<enum 'ListOrder'>, 'storage_enum', <function _encode_enum>), (<class 'swh.storage.interface.OriginVisitWithStatuses'>, 'origin_visit_with_statuses', <function _encode_origin_visit_with_statuses>), (<class 'swh.storage.interface.ObjectReference'>, 'object_reference', <function _encode_object_reference>), (<class 'swh.storage.interface.SnapshotBranchByNameResponse'>, 'branch_by_name_response', <function _encode_snapshot_branch_by_name_response>), (<class 'swh.storage.proxies.masking.db.MaskedStatus'>, 'masked_status', <function _encode_masked_status>), (<class 'swh.storage.proxies.blocking.db.BlockingStatus'>, 'blocking_status', <function _encode_blocking_status>)]

Value of extra_encoders passed to json_dumps or msgpack_dumps to be able to serialize more object types.

method_decorators: List[Callable[[Callable], Callable]] = [<function timed>]

List of decorators to all methods generated from the backend_class.

post_content_add(ret, kw)

post_content_add_metadata(ret, kw)

post_skipped_content_add(ret, kw)

post_directory_add(ret, kw)

post_revision_add(ret, kw)

post_release_add(ret, kw)

post_snapshot_add(ret, kw)

post_origin_visit_status_add(ret, kw)

post_origin_add(ret, kw)

post_raw_extrinsic_metadata_add(ret, kw)

post_metadata_fetcher_add(ret, kw)

post_metadata_authority_add(ret, kw)

post_extid_add(ret, kw)

post_origin_visit_add(ret, kw)

swh.storage.api.server.non_retryable_error_handler(exception)

Send all non-retryable errors with a 400 status code so the client can re-raise them.

swh.storage.api.server.default_error_handler(exception)

swh.storage.api.server.index()

swh.storage.api.server.stat_counters()

swh.storage.api.server.refresh_stat_counters()

swh.storage.api.server.load_and_check_config(config_path: str | None) → Dict[str, Any]

Check the minimal configuration is set to run the api or raise an

error explanation.

Parameters:

config_path – Path to the configuration file to load

Raises:

Error if the setup is not as expected –

Returns:

configuration as a dict

swh.storage.api.server.make_app_from_configfile() → StorageServerApp

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.