Inheritance diagram for lsst.dax.apdb.apdbSchema.ApdbSchema:

Public Member Functions
	__init__ (self, str schema_file, str ss_schema_file)

type\|str	column_dtype (self, felis.datamodel.DataType\|ExtraDataTypes felis_type)

VersionTuple	schemaVersion (self)

bool	has_mjd_timestamps (self)

Public Attributes
	tableSchemas

Protected Member Functions
tuple[Mapping[ApdbTables, Table], VersionTuple\|None]	_buildSchemas (cls, str schema_file)

Protected Attributes
tuple[Mapping[ApdbTables, Table], VersionTuple\|None]	_schemaVersion = self._buildSchemas(schema_file)

Detailed Description

Class for management of APDB schema.

Attributes
----------
tableSchemas : `dict`
    Maps table name to `TableDef` instance.

Parameters
----------
schema_file : `str`
    Location of the YAML file with APDB schema.
ss_schema_file : `str`
    Location of the YAML file with SSO schema. File will be loaded if APDB
    schema file does not contain SSObject/SSSource tables. Can be set to
    empty string to skip loading of SSObject/SSSource schema.

Definition at line 115 of file apdbSchema.py.

Constructor & Destructor Documentation

◆ init()

lsst.dax.apdb.apdbSchema.ApdbSchema.__init__	(		self,
		str	schema_file,
		str	ss_schema_file )

Definition at line 133 of file apdbSchema.py.

    ):
        # build complete table schema
        self.tableSchemas, self._schemaVersion = self._buildSchemas(schema_file)
        if ss_schema_file:
            if ApdbTables.SSObject not in self.tableSchemas or ApdbTables.SSSource not in self.tableSchemas:
                # Read additional SSP schema.
                ssp_tables, _ = self._buildSchemas(ss_schema_file)
                if ApdbTables.SSObject not in ssp_tables or ApdbTables.SSSource not in ssp_tables:
                    raise LookupError(f"Cannot locate SSObject/SSSource table in {ss_schema_file}")
                self.tableSchemas = dict(self.tableSchemas) | {
                    ApdbTables.SSObject: ssp_tables[ApdbTables.SSObject],
                    ApdbTables.SSSource: ssp_tables[ApdbTables.SSSource],
                }
 

Member Function Documentation

◆ _buildSchemas()

tuple[Mapping[ApdbTables, Table], VersionTuple \| None] lsst.dax.apdb.apdbSchema.ApdbSchema._buildSchemas	(		cls,
		str	schema_file )

protected

Create schema definitions for tables from felis schema.

Reads YAML schema and builds a dictionary containing
`.schema_model.Table` instances for each APDB table appearing in that
schema.

Parameters
----------
schema_file : `str`
    Name of YAML file with ``felis`` schema.

Returns
-------
tables : `dict` [`ApdbTables`, `schema_model.Table`]
    Mapping of table names to `.schema_model.Table` instances.
version : `VersionTuple` or `None`
    Schema version defined in schema file, `None` if version is not
    defined.

Definition at line 189 of file apdbSchema.py.

    def _buildSchemas(cls, schema_file: str) -> tuple[Mapping[ApdbTables, Table], VersionTuple | None]:
        """Create schema definitions for tables from felis schema.
 
        Reads YAML schema and builds a dictionary containing
        `.schema_model.Table` instances for each APDB table appearing in that
        schema.
 
        Parameters
        ----------
        schema_file : `str`
            Name of YAML file with ``felis`` schema.
 
        Returns
        -------
        tables : `dict` [`ApdbTables`, `schema_model.Table`]
            Mapping of table names to `.schema_model.Table` instances.
        version : `VersionTuple` or `None`
            Schema version defined in schema file, `None` if version is not
            defined.
        """
        _LOG.debug("Loading felis schema from %s", schema_file)
        felis_schema = felis.datamodel.Schema.from_uri(schema_file, context={"id_generation": True})
        schema = Schema.from_felis(felis_schema)
 
        # convert all dicts into classes
        tables: MutableMapping[ApdbTables, Table] = {}
        for table in schema.tables:
            try:
                table_enum = ApdbTables(table.name)
            except ValueError:
                # There may be other tables in the schema that do not belong
                # to APDB.
                continue
            else:
                tables[table_enum] = table
 
        version: VersionTuple | None = None
        if schema.version is not None:
            version = VersionTuple.fromString(schema.version.current)
 
        _LOG.debug("Loaded schema for tables %s", list(tables))
        return tables, version
 

◆ column_dtype()

type \| str lsst.dax.apdb.apdbSchema.ApdbSchema.column_dtype	(		self,
		felis.datamodel.DataType \| ExtraDataTypes	felis_type )

Return Pandas data type for a given Felis column type.

Parameters
----------
felis_type : `felis.datamodel.DataType`
    Felis type, on of the enums defined in `felis.datamodel` module.

Returns
-------
column_dtype : `type` or `str`
    Type that can be used for columns in Pandas.

Raises
------
TypeError
    Raised if type is cannot be handled.

Definition at line 151 of file apdbSchema.py.

    def column_dtype(self, felis_type: felis.datamodel.DataType | ExtraDataTypes) -> type | str:
        """Return Pandas data type for a given Felis column type.
 
        Parameters
        ----------
        felis_type : `felis.datamodel.DataType`
            Felis type, on of the enums defined in `felis.datamodel` module.
 
        Returns
        -------
        column_dtype : `type` or `str`
            Type that can be used for columns in Pandas.
 
        Raises
        ------
        TypeError
            Raised if type is cannot be handled.
        """
        try:
            return _dtype_map[felis_type]
        except KeyError:
            raise TypeError(f"Unexpected Felis type: {felis_type}")
 

◆ has_mjd_timestamps()

bool lsst.dax.apdb.apdbSchema.ApdbSchema.has_mjd_timestamps ( self )

True if timestamps columns are in MJD (`bool`).

Definition at line 233 of file apdbSchema.py.

    def has_mjd_timestamps(self) -> bool:
        """True if timestamps columns are in MJD (`bool`)."""
        table = self.tableSchemas[ApdbTables.DiaObject]
        # Look for validityStartMjdTai or validityStart
        for column in table.columns:
            if column.name == "validityStartMjdTai":
                return True
            elif column.name == "validityStart":
                return False
        raise LookupError(
            "Could not find validityStart or validityStartMjdTai column in DiaObject table schema."
        )

◆ schemaVersion()

VersionTuple lsst.dax.apdb.apdbSchema.ApdbSchema.schemaVersion ( self )

Return schema version as defined in YAML schema file.

Returns
-------
version : `VersionTuple`
    Version number read from YAML file, if YAML file does not define
    schema version then "0.1.0" is returned.

Definition at line 174 of file apdbSchema.py.

    def schemaVersion(self) -> VersionTuple:
        """Return schema version as defined in YAML schema file.
 
        Returns
        -------
        version : `VersionTuple`
            Version number read from YAML file, if YAML file does not define
            schema version then "0.1.0" is returned.
        """
        if self._schemaVersion is None:
            return VersionTuple(0, 1, 0)
        else:
            return self._schemaVersion
 

Member Data Documentation

◆ _schemaVersion

tuple[Mapping[ApdbTables, Table], VersionTuple | None] lsst.dax.apdb.apdbSchema.ApdbSchema._schemaVersion = self._buildSchemas(schema_file)

protected

Definition at line 139 of file apdbSchema.py.

◆ tableSchemas

lsst.dax.apdb.apdbSchema.ApdbSchema.tableSchemas

Definition at line 139 of file apdbSchema.py.

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

/j/snowflake/release/lsstsw/stack/lsst-scipipe-12.1.0/Linux/dax_apdb/gb88ae4c679+7dec8f19df/python/lsst/dax/apdb/apdbSchema.py

Public Member Functions

Public Attributes

Protected Member Functions

Protected Attributes