registrar package

Subpackages

• registrar.backend package
  • Submodules
  • registrar.backend.eoxserver module
    • backend.py
    • AssetsToCoverageMapping
      • AssetsToCoverageMapping.assets
      • AssetsToCoverageMapping.assets
    • ItemBackend
      • ItemBackend.deregister()
      • ItemBackend.deregister_identifier()
      • ItemBackend.exists()
      • ItemBackend.register()
    • ItemToProductTypeMapping
      • ItemToProductTypeMapping.product_type
      • ItemToProductTypeMapping.filter
      • ItemToProductTypeMapping.coverages
      • ItemToProductTypeMapping.collections
      • ItemToProductTypeMapping.collections
      • ItemToProductTypeMapping.coverages
      • ItemToProductTypeMapping.filter
      • ItemToProductTypeMapping.name
      • ItemToProductTypeMapping.product_type
  • registrar.backend.utils module
    • get_backends()
  • Module contents
    • get_backends()
• registrar.route package
  • Submodules
  • registrar.route.json module
    • JSONRoute
      • JSONRoute.get_source()
      • JSONRoute.parse()
  • registrar.route.path module
    • PathRoute
      • PathRoute.get_source()
      • PathRoute.parse()
  • registrar.route.stac module
    • CollectionRoute
      • CollectionRoute.get_source()
      • CollectionRoute.parse()
    • ItemRoute
      • ItemRoute.get_source()
      • ItemRoute.parse()
  • Module contents

Submodules

registrar.abc module

Module for registrar routes

class registrar.abc.Backend(*args, **kwds)

Bases: ABC, Generic[T]

A backend registers the given item

abstract deregister(source: Source | None, item: T)

Deregisters an item using its specification.

abstract deregister_identifier(identifier: str)

Deregisters an item using its identifier.

abstract exists(source: Source | None, item: T) → bool

Checks if an item exists in the backend.

abstract register(source: Source | None, item: T, replace: bool)

Registers an item on the backend.

class registrar.abc.Route(route_config: RouteConfig)

Bases: ABC, Generic[TR]

A route is of a specific type and manages the registration from a specific queue to a list of backends.

abstract get_source(source_cfgs: List[SourceConfig], item: TR) → Source | None

Determines the source of the given parsed item.

abstract parse(raw: str) → TR

Parses the value from the given raw input.

property replace

Whether to allow replacement for this route

registrar.cli module

Command line interfaces for the registrar

registrar.cli.setup_logging(debug=False)

Sets up the logging configuration

registrar.config module

config.py

Contains configuration functions for the registrar

class registrar.config.BackendConfig(path: str, args: ~typing.List[~typing.Any] = <factory>, kwargs: ~typing.Dict[str, ~typing.Any] = <factory>)

Bases: object

Configuration for a specific backend.

path

the dotpath to the implementing class

Type:

str

args

any arguments to be passed to the class when instantiating it

Type:

dict

kwargs

any keyword arguments to be passed to the class when instantiating it

Type:

dict

args: List[Any]

kwargs: Dict[str, Any]

path: str

class registrar.config.HandlerConfig(path: str, args: ~typing.List[~typing.Any] = <factory>, kwargs: ~typing.Dict[str, ~typing.Any] = <factory>)

Bases: object

A handler configuration

path

the dotpath of the handler class

Type:

str

args

any handler arguments

Type:

List[Any]

kwargs

any handler keyword-arguments

Type:

dict

args: List[Any]

kwargs: Dict[str, Any]

path: str

class registrar.config.HandlersConfig(pre: ~typing.List[~registrar.config.HandlerConfig] = <factory>, post: ~typing.List[~registrar.config.HandlerConfig] = <factory>, error: ~typing.List[~registrar.config.HandlerConfig] = <factory>)

Bases: object

Pre-/success-/error-handlers for a given route.

pre

the pre-handlers configuration

Type:

List[HandlerConfig]

post

the post-handlers configuration

Type:

List[HandlerConfig]

error

the error-handlers configuration

Type:

List[HandlerConfig]

error: List[HandlerConfig]

classmethod from_dict(values: dict) → HandlersConfig

Constructs a handler config from a dict

post: List[HandlerConfig]

pre: List[HandlerConfig]

class registrar.config.RegistrarConfig(routes: Dict[str, RouteConfig], sources: List[SourceConfig], redis_host: str | None = 'redis', redis_port: int | None = 6379)

Bases: object

The root registration configuration object.

routes

all routes

Type:

Dict[str, RouteConfig]

sources

the sources

Type:

List[SourceConfig]

classmethod from_dict(values: dict, validate: bool = False) → RegistrarConfig

Parses a RegistrarConfig from a dictionary

classmethod from_file(input_file: TextIO, validate: bool = False) → RegistrarConfig

Parses a RegistrarConfig from a file

get_route(queue: str) → RouteConfig

Returns the RouteConfig for the given queue name

redis_host: str | None = 'redis'

redis_port: int | None = 6379

routes: Dict[str, RouteConfig]

sources: List[SourceConfig]

class registrar.config.RouteConfig(path: str, queue: str, mode: ~registrar.config.RouteMode = RouteMode.REGISTER, success_queue: str | None = None, error_queue: str | None = None, replace: bool = False, simplify_footprint_tolerance: float | None = None, backends: ~typing.List[~registrar.config.BackendConfig] = <factory>, handlers: ~registrar.config.HandlersConfig = <factory>, args: ~typing.List[~typing.Any] = <factory>, kwargs: ~typing.Dict[str, ~typing.Any] = <factory>)

Bases: object

A registration route configuration

path

the dotpath to the implementing class

Type:

str

queue

the queue for this route to listen on

Type:

str

mode

(RouteMode, optional): the default mode to use for this route

Type:

registrar.config.RouteMode

success_queue

the queue to put successfully registered items on

Type:

str, optional

error_queue

the queue to put the items on upon eror

Type:

str, optional

replace

whether replacement of items is allowed

Type:

bool, optional

backends

all associated backends for that route

Type:

List[BackendConfig], optional

handlers

any handlers associated

Type:

HandlerConfig

args

any arguments to supply to the implementing class

Type:

list, optional

kwargs

any keyword arguments to supply to the implementing class

Type:

list, optional

args: List[Any]

backends: List[BackendConfig]

error_queue: str | None = None

classmethod from_dict(values: dict) → RouteConfig

Parses a RouteConfig from a dictionary

handlers: HandlersConfig

kwargs: Dict[str, Any]

mode: RouteMode = 'REGISTER'

path: str

queue: str

replace: bool = False

simplify_footprint_tolerance: float | None = None

success_queue: str | None = None

class registrar.config.RouteMode(value)

Bases: Enum

The route mode.

DEREGISTER = 'DEREGISTER'

DEREGISTER_IDENTIFIER = 'DEREGISTER_IDENTIFIER'

REGISTER = 'REGISTER'

class registrar.config.SourceConfig(type: str, name: str, filter: str | None = None, args: ~typing.List[~typing.Any] = <factory>, kwargs: ~typing.Dict[str, str] = <factory>)

Bases: object

The configuration of a specific remote source.

type

the type name of the source

Type:

str

name

the given name of the source

Type:

str

filter

a filter to see if a path is relative to this source

Type:

str, optional

args

additional options for the implementing source class constructor

Type:

list

kwargs

additional options for the implementing source class constructor

Type:

dict

args: List[Any]

filter: str | None = None

kwargs: Dict[str, str]

name: str

type: str

registrar.config.constructor_env_variables(loader: Loader, node)

Extracts the environment variable from the node’s value

Parameters:

• loader (yaml.Loader) – the yaml loader

• node (str) – the current node in the yaml

Returns:

the parsed string that contains the value of the environment variable

Return type:

str

registrar.config.load_config(input_file: TextIO) → dict

Attempts to load config from yaml file

Parameters:

input_file (TextIO) – the file to be read

Returns:

dictionary containing configuration values

Return type:

dict

registrar.config.validate_config(config)

Validate against the config-schema.yaml

registrar.daemon module

daemon.py

Contains the daemon functions for the registrar

registrar.daemon.run_daemon(config: RegistrarConfig)

Run the registration daemon

Parameters:

config (RegistarConfig) – the root configuration.

registrar.exceptions module

exceptions.py

Contains exceptions that may occur during registration

exception registrar.exceptions.RegistrationError

Bases: Exception

Registration error class

registrar.json module

json.py

Contains helper functions for reading json using concepts from the registrar

registrar.json.read_json(source: Source, path: str) → Dict | List

Helper to easily read a file on a source as JSON and decode it.

Parameters:

• source (Source) – From which source to read the file

• path (str) – Path to file

Returns:

json converted to dict

Return type:

Union[Dict, List]

registrar.post_handlers module

post_handlers.py

Contains different handlers that trigger post registration

class registrar.post_handlers.ReportingPostHandler(service_url: str, reporting_dir: str)

Bases: object

sanitize_path(path)

registrar.registrar module

Contains all functions relevant for registration

registrar.registrar.deregister(route_cfg: RouteConfig, source_cfgs: List[SourceConfig], value: str, use_id: bool = False)

Handles the deregistration of a specific item.

Parameters:

• route_cfg (RouteConfig) – the used route configuration

• source_cfgs (List[SourceConfig]) – the source configs

• value (str) – the raw value to be parsed or the identifier if use_id is used.

• use_id (bool) – to deregister using the identifier, or a parsed item.

registrar.registrar.get_error_handlers(config: HandlersConfig) → List[Callable]

Instantiates error error handlers.

registrar.registrar.get_post_handlers(config: HandlersConfig) → List[Callable]

Instantiates post error handlers.

registrar.registrar.get_pre_handlers(config: HandlersConfig) → List[Callable]

Instantiates pre error handlers.

registrar.registrar.register(route_cfg: RouteConfig, source_cfgs: List[SourceConfig], value: str)

Handles the registration of a specific item

Parameters:

• route_cfg (RouteConfig) – the used route configuration

• source_cfgs (List[SourceConfig]) – the source configs

• value (str) – the raw value to be parsed

registrar.source module

source.py

Contains different file source concepts where the data may be stored such as local storage, S3, swift

class registrar.source.HTTPSource(name: str, endpoint_url: str, streaming: bool)

Bases: Source

Source class for HTTP locations

get_container_and_path(path: str)

Split the input path into a container and a path part

get_file(path: str, target_path: str)

Download the given file to the target location

get_vsi_env_and_path(path: str)

Get a VSI conformant path. See https://gdal.org/user/virtual_file_systems.html

list_files(path: str, glob_patterns: list | None = None)

Return a list of file references for the given base path and glob pattern

class registrar.source.LocalSource(name: str, root_directory: str)

Bases: Source

Handles data on local filesystem

Parameters:

• name (str) – Name of the filesystem

• root_directory (str) – path to root

get_container_and_path(path: str)

Split the input path into a container and a path part

get_file(path: str, target_path: str)

Download the given file to the target location

get_vsi_env_and_path(path: str)

Get a VSI conformant path. See https://gdal.org/user/virtual_file_systems.html

list_files(path: str, glob_patterns: list | None = None)

Return a list of file references for the given base path and glob pattern

class registrar.source.S3Source(name: str | None = None, bucket_name: str | None = None, secret_access_key: str | None = None, access_key_id: str | None = None, endpoint_url: str = '', strip_bucket: bool = True, validate_bucket_name: bool = True, region_name: str | None = None, public: bool = False, streaming: bool = False, **client_kwargs)

Bases: Source

Handles data located on an S3 bucket

Parameters:

• name (str, optional) – Name of the source. Defaults to None.

• bucket_name (str, optional) – Name of the bucket. Defaults to None.

• secret_access_key (str, optional) – secret access key. Defaults to None.

• access_key_id (str, optional) – access key identifier. Defaults to None.

• endpoint_url (str, optional) – endpoint url. Defaults to None.

• strip_bucket (bool, optional) – whether to strip bucket name when constructing paths. Defaults to True.

• validate_bucket_name (bool, optional) – whether to validate the name of bucket. Defaults to True.

• region_name (str, optional) – name of aws s3 region. Defaults to None.

• public (bool, optional) – whether the data is public or not. Defaults to False.

• streaming (bool, optional) – If streaming is to be used by GDAL. Defaults to False.

get_container_and_path(path: str)

Split the input path into a container and a path part

get_file(path: str, target_path: str)

Download the given file to the target location

get_vsi_env_and_path(path: str)

Get a VSI conformant path. See https://gdal.org/user/virtual_file_systems.html

list_files(path: str, glob_patterns: list | None = None)

Return a list of file references for the given base path and glob pattern

class registrar.source.Source(name: str | None = None, endpoint_url: str = '')

Bases: ABC

Abstract base class for all sources

abstract get_container_and_path(path: str)

Split the input path into a container and a path part

abstract get_file(path: str, target_path: str)

Download the given file to the target location

abstract get_vsi_env_and_path(path: str)

Get a VSI conformant path. See https://gdal.org/user/virtual_file_systems.html

abstract list_files(path: str, glob_patterns: list | None = None)

Return a list of file references for the given base path and glob pattern

class registrar.source.SwiftSource(name: str | None = None, username: str | None = None, password: str | None = None, project_name: str | None = None, project_id: str | None = None, region_name: str | None = None, project_domain_id: str | None = None, project_domain_name: str | None = None, user_domain_id: str | None = None, user_domain_name: str | None = None, auth_url: str | None = None, auth_url_short: str | None = None, auth_version: str | None = None, container: str | None = None, streaming: bool = False)

Bases: Source

Handles data located on a openstack swift bucket

Parameters:

• name (str, optional) – Name of bucket. Defaults to None.

• username (str, optional) – username for authentication. Defaults to None.

• password (str, optional) – password for authentication Defaults to None.

• project_name (str, optional) – name of swift tenant. Defaults to None.

• project_id (str, optional) – id of swift tenant. Defaults to None.

• region_name (str, optional) – name of region. Defaults to None.

• project_domain_id (str, optional) – project domain identifier. Defaults to None.

• project_domain_name (str, optional) – name of project domain. Defaults to None.

• user_domain_id (str, optional) – user domain identifier. Defaults to None.

• user_domain_name (str, optional) – name of user domain. Defaults to None.

• auth_url (str, optional) – url to authenticate to. Defaults to None.

• auth_url_short (str, optional) – short url to auth to. Defaults to None.

• auth_version (str, optional) – swift auth version. Defaults to None.

• container (str, optional) – name of swift container. Defaults to None.

• streaming (bool, optional) – If streaming is to be used by GDAL. Defaults to False.

get_container_and_path(path: str)

Split the input path into a container and a path part

get_file(path: str, target_path: str)

Download the given file to the target location

get_service()

Returns the swiftclient.SwiftService for the options.

get_vsi_env_and_path(path: str)

Get a VSI conformant path. See https://gdal.org/user/virtual_file_systems.html

list_files(path: str, glob_patterns: list | None = None)

Return a list of file references for the given base path and glob pattern

registrar.source.get_source(source_cfgs: List[SourceConfig], hrefs: List[str]) → Source | None

Retrieves a Source from a given list of SourceConfigs and a list of hrefs to test against.

Parameters:

• source_cfgs (List[SourceConfig]) – the source configs to test

• hrefs (List[str]) – the hrefs to test the sources against

Returns:

the constructed source from the tested source configuration

Return type:

Source

registrar.utils module

utils.py

Contains helper functions

registrar.utils.extract_footprint(env: dict, vsipath: str) → str

Extracts the footprint from the extent of file. This method is currently imprecise due to the extent being different from the actual footprint in cases of nodata, unless the geotiff has geometry stored as geojson in a metadata attribute geometry

Parameters:

• env (dict) – environment vars for gdal

• vsipath (str) – vsi path to file

Returns:

WKT formatted footprint

Return type:

str

registrar.utils.extract_metadata(env: dict, vsipath: str) → dict

Extracts metadata from file

Parameters:

• env (dict) – environment vars for

• vsipath (str) – vsi path to file

Returns:

metadata as key value pairs

Return type:

dict

registrar.utils.import_by_path(path: str)

Imports the object from the referenced module.

Parameters:

• path (str) – the dotted Python path, where the last element is the

• module. (object in the referenced) –

registrar.utils.isoformat(dt: datetime) → str

Formats a datetime object to an ISO string. Timezone naive datetimes are

are treated as UTC Zulu. UTC Zulu is expressed with the proper “Z” ending and not with the “+00:00” offset declaration.

Parameters:

dt (datetime) – datetime to encode

Returns:

iso formatted datetime

Return type:

str

registrar.xml module

Module contents