Registration routes

Adding implementation routes

A Route is bound to a queue and operates on a particular data-type (such as STAC Item, Collection, paths, arbitrary JSON structures).

Backends are also now generic according to its datatype and are now part of the route.

Each route also encapsulates the pre, post, and error handlers.

Example configuration:

routes:
  items:
    path: registrar.route.stac.ItemRoute
    queue: register
    output_queue: next_queue
    backends:
      - path: registrar.backend.eoxserver.ItemBackend
        kwargs:
          instance_base_path: /some/dir
          instance_name: name

    handlers:
      pre:
        - path: some.pre.handler
      post:
        - path: some.post.handler
      error:
        - path: some.error.handler

  collections:
    path: registrar.route.stac.CollectionRoute
    queue: register-collections
    backends:
      - path: registrar.backend.eoxserver.CollectionBackend

  path:
    path: registrar.route.path.PathRoute
    queue: register-paths
    backends:
      - path: some.path.backend

sources: [...]

This is how a STAC Item route can be implemented:

import json
import pystac

class ItemRoute(Route[pystac.Item]):
    """A route implementation for STAC Items"""

    def parse(self, raw: str) -> pystac.Item:
        return pystac.Item.from_dict(json.loads(raw))

    def get_source(
        self, source_cfgs: List[SourceConfig], item: pystac.Item
    ) -> Optional[Source]:
        return get_source(
            source_cfgs, [asset.href for asset in item.assets.values()]
        )

This is how the Backend interface can be satisfied:

from pystac import Item

# ...

class MyItemBackend(Backend[Item]):
    def exists(self, source: Optional[Source], item: Item) -> bool:
        ...

    def register(self, source: Optional[Source], item: Item, replace: bool):
        ...

    def deregister(self, source: Optional[Source], item: Item):
        ...

    def deregister_identifier(self, identifier: str):
        ...

registrar/backend/__init__.py

-from typing import List
+"""Registrar backends.
+"""
 
-from ..utils import import_by_path
-from .abc import ItemBackend, PathBackend
 
+from .utils import get_backends
 
-def get_backends(config: dict) -> List[ItemBackend]:
-    """Gets the backends for registering into
 
-    Args:
-        config (dict): registrar configuration
-
-    Returns:
-        List[ItemBackend]: list of backends where registration will be
-            carried out
-    """
-    backends: List[ItemBackend] = []
-    for cfg_backend in config["backends"]:
-        # construct backend
-        backend_cls = import_by_path(cfg_backend["path"])
-
-        backends.append(
-            backend_cls(
-                *cfg_backend.get("args", []),
-                **cfg_backend.get("kwargs", {}),
-            )
-        )
-
-    return backends
-
-
-def get_path_backends(config: dict) -> List[PathBackend]:
-    """Gets the backends for registering the path into
-
-    Args:
-        config (dict): registrar configuration
-
-    Returns:
-        List[PathBackend]: list of backends where registration will be
-            carried out
-    """
-    backends = []
-    for cfg_backend in config.get("path_backends", []):
-        # construct backend
-        backend_cls = import_by_path(cfg_backend["path"])
-
-        backends.append(
-            backend_cls(
-                *cfg_backend.get("args", []),
-                **cfg_backend.get("kwargs", {}),
-            )
-        )
-
-    return backends
+__all__ = ["get_backends"]