lsst.dax.apdb.apdbIndex.ApdbIndex Class Reference

Public Member Functions
	__init__ (self, str|None index_path=None)
ResourcePath	get_apdb_uri (self, str label, str|None format=None)
Mapping[str, str]	get_entries (self)

Static Public Attributes
ClassVar	index_env_var = "DAX_APDB_INDEX_URI"

Protected Member Functions
Mapping[str, str]	_read_index (self, str|None index_path=None)

Protected Attributes
	_index_path
	_cache

Static Protected Attributes
Mapping	_cache = None

Detailed Description

Index of well-known Apdb instances.

Parameters
----------
index_path : `str`, optional
    Path to the index configuration file, if not provided then the value
    of ``DAX_APDB_INDEX_URI`` environment variable is used to locate
    configuration file.

The index is configured from a simple YAML file whose location is
determined from ``DAX_APDB_INDEX_URI`` environment variable. The content
of the index file is a mapping of the labels to URIs in YAML format, e.g.:

.. code-block:: yaml

   dev: "/path/to/config-file.yaml"
   "prod/pex_config": "s3://bucket/apdb-prod.py"
   "prod/yaml": "s3://bucket/apdb-prod.yaml"

The labels in the index file consists of the label name and an optional
format name separated from label by slash. `get_apdb_uri` method can
use its ``format`` argument to return either a format-specific
configuration or a label-only configuration if format-specific is not
defined.

Definition at line 39 of file apdbIndex.py.

Constructor & Destructor Documentation

__init__()

lsst.dax.apdb.apdbIndex.ApdbIndex.__init__	(		self,
		str | None	index_path = None )

Definition at line 74 of file apdbIndex.py.

    def __init__(self, index_path: str | None = None):
        self._index_path = index_path
 

Member Function Documentation

_read_index()

Mapping[str, str] lsst.dax.apdb.apdbIndex.ApdbIndex._read_index ( self, str | None index_path = None )

protected

Return contents of the index file.

Parameters
----------
index_path : `str`, optional
    Location of the index file, if not provided then default location
    is used.

Returns
-------
entries : `~collections.abc.Mapping` [`str`, `str`]
    All known entries. Can be empty if no index can be found.

Raises
------
RuntimeError
    Raised if ``index_path`` is not provided and environment variable
    is not set.
TypeError
    Raised if content of the configuration file is incorrect.

Definition at line 77 of file apdbIndex.py.

    def _read_index(self, index_path: str | None = None) -> Mapping[str, str]:
        """Return contents of the index file.
 
        Parameters
        ----------
        index_path : `str`, optional
            Location of the index file, if not provided then default location
            is used.
 
        Returns
        -------
        entries : `~collections.abc.Mapping` [`str`, `str`]
            All known entries. Can be empty if no index can be found.
 
        Raises
        ------
        RuntimeError
            Raised if ``index_path`` is not provided and environment variable
            is not set.
        TypeError
            Raised if content of the configuration file is incorrect.
        """
        if self._cache is not None:
            return self._cache
        if index_path is None:
            index_path = os.getenv(self.index_env_var)
            if not index_path:
                raise RuntimeError(
                    f"No repository index defined, environment variable {self.index_env_var} is not set."
                )
        index_uri = ResourcePath(index_path)
        _LOG.debug("Opening YAML index file: %s", index_uri.geturl())
        try:
            content = index_uri.read()
        except IsADirectoryError as exc:
            raise FileNotFoundError(f"Index file {index_uri.geturl()} is a directory") from exc
        stream = io.BytesIO(content)
        if index_data := yaml.load(stream, Loader=yaml.SafeLoader):
            try:
                self._cache = TypeAdapter(dict[str, str]).validate_python(index_data)
            except ValidationError as e:
                raise TypeError(f"Repository index {index_uri.geturl()} not in expected format") from e
        else:
            self._cache = {}
        return self._cache
 

get_apdb_uri()

ResourcePath lsst.dax.apdb.apdbIndex.ApdbIndex.get_apdb_uri	(		self,
		str	label,
		str | None	format = None )

Return URI for APDB configuration file given its label.

Parameters
----------
label : `str`
    Label of APDB instance.
format : `str`
    Format of the APDB configuration file, arbitrary string. This can
    be used to support an expected migration from pex_config to YAML
    configuration for APDB, code that uses pex_config could provide
    "pex_config" for ``format``. The actual key in the index is
    either a slash-separated label and format, or, if that is missing,
    just a label.

Returns
-------
uri : `~lsst.resources.ResourcePath`
    URI for the configuration file for APDB instance.

Raises
------
FileNotFoundError
    Raised if an index is defined in the environment but it
    can not be found.
ValueError
    Raised if the label is not found in the index.
RuntimeError
    Raised if ``index_path`` is not provided and environment variable
    is not set.
TypeError
    Raised if the format of the index file is incorrect.

Definition at line 123 of file apdbIndex.py.

    def get_apdb_uri(self, label: str, format: str | None = None) -> ResourcePath:
        """Return URI for APDB configuration file given its label.
 
        Parameters
        ----------
        label : `str`
            Label of APDB instance.
        format : `str`
            Format of the APDB configuration file, arbitrary string. This can
            be used to support an expected migration from pex_config to YAML
            configuration for APDB, code that uses pex_config could provide
            "pex_config" for ``format``. The actual key in the index is
            either a slash-separated label and format, or, if that is missing,
            just a label.
 
        Returns
        -------
        uri : `~lsst.resources.ResourcePath`
            URI for the configuration file for APDB instance.
 
        Raises
        ------
        FileNotFoundError
            Raised if an index is defined in the environment but it
            can not be found.
        ValueError
            Raised if the label is not found in the index.
        RuntimeError
            Raised if ``index_path`` is not provided and environment variable
            is not set.
        TypeError
            Raised if the format of the index file is incorrect.
        """
        index = self._read_index(self._index_path)
        labels: list[str] = [label]
        if format:
            labels.insert(0, f"{label}/{format}")
        for label in labels:
            if (uri_str := index.get(label)) is not None:
                return ResourcePath(uri_str)
        if len(labels) == 1:
            message = f"Label {labels[0]} is not defined in index file"
        else:
            labels_str = ", ".join(labels)
            message = f"None of labels {labels_str} is defined in index file"
        all_labels = set(index)
        raise ValueError(f"{message}, labels known to index: {all_labels}")
 

get_entries()

Mapping[str, str] lsst.dax.apdb.apdbIndex.ApdbIndex.get_entries

(

self

)

Retrieve all entries defined in index.

Returns
-------
entries : `~collections.abc.Mapping` [`str`, `str`]
    All known index entries.

Raises
------
RuntimeError
    Raised if ``index_path`` is not provided and environment variable
    is not set.
TypeError
    Raised if content of the configuration file is incorrect.

Definition at line 171 of file apdbIndex.py.

    def get_entries(self) -> Mapping[str, str]:
        """Retrieve all entries defined in index.
 
        Returns
        -------
        entries : `~collections.abc.Mapping` [`str`, `str`]
            All known index entries.
 
        Raises
        ------
        RuntimeError
            Raised if ``index_path`` is not provided and environment variable
            is not set.
        TypeError
            Raised if content of the configuration file is incorrect.
        """
        return self._read_index(self._index_path)

Member Data Documentation

_cache [1/2]

Mapping lsst.dax.apdb.apdbIndex.ApdbIndex._cache = None

staticprotected

Definition at line 71 of file apdbIndex.py.

_cache [2/2]

lsst.dax.apdb.apdbIndex.ApdbIndex._cache

protected

Definition at line 116 of file apdbIndex.py.

_index_path

lsst.dax.apdb.apdbIndex.ApdbIndex._index_path

protected

Definition at line 75 of file apdbIndex.py.

index_env_var

ClassVar lsst.dax.apdb.apdbIndex.ApdbIndex.index_env_var = "DAX_APDB_INDEX_URI"

static

Definition at line 66 of file apdbIndex.py.

---

The documentation for this class was generated from the following file:

• /j/snowflake/release/lsstsw/stack/lsst-scipipe-8.0.0/Linux64/dax_apdb/g88963caddf+0cb8e002cc/python/lsst/dax/apdb/apdbIndex.py