Inheritance diagram for lsst.dax.apdb.sql.apdbSqlSchema.ApdbSqlSchema:

Public Member Functions
	__init__ (self, sqlalchemy.engine.Engine engine, str dia_object_index, str htm_index_column, str schema_file, str schema_name="ApdbSchema", str prefix="", str\|None namespace=None, bool enable_replica=False)

bool	empty (self)

None	makeSchema (self, bool drop=False)

sqlalchemy.schema.Table	get_table (self, ApdbTables\|ExtraTables table_enum)

list[sqlalchemy.schema.Column]	get_apdb_columns (self, ApdbTables\|ExtraTables table_enum)

bool	has_replica_chunks (self)

Static Public Attributes
tuple	pixel_id_tables = (ApdbTables.DiaObject, ApdbTables.DiaObjectLast, ApdbTables.DiaSource)

Protected Member Functions
bool	_check_replica_chunks (self)

Mapping[ApdbTables, schema_model.Table]	_make_apdb_tables (self, str mysql_engine="InnoDB")

Mapping[ExtraTables, schema_model.Table]	_make_extra_tables (self, Mapping[ApdbTables, schema_model.Table] apdb_tables)

list[schema_model.Column]	_replicaChunkColumns (self, ExtraTables table_enum, ApdbTables apdb_enum)

list[schema_model.Index]	_replicaChunkIndices (self, ExtraTables table_enum, Mapping[str, schema_model.Column] column_map)

list[schema_model.Constraint]	_replicaChunkConstraints (self, ExtraTables table_enum, schema_model.Table apdb_table, schema_model.Table parent_table, Mapping[str, schema_model.Column] column_map)

Protected Attributes
	_engine

	_dia_object_index

	_htm_index_column

	_prefix

	_enable_replica

	_metadata

	_apdb_tables

	_extra_tables

	_has_replica_chunks

	_metadata_check

Detailed Description

Class for management of APDB schema.

Attributes
----------
objects : `sqlalchemy.Table`
    DiaObject table instance
objects_last : `sqlalchemy.Table`
    DiaObjectLast table instance, may be None
sources : `sqlalchemy.Table`
    DiaSource table instance
forcedSources : `sqlalchemy.Table`
    DiaForcedSource table instance
has_replica_chunks : `bool`
    If true then schema has tables for replication chunks.

Parameters
----------
engine : `sqlalchemy.engine.Engine`
    SQLAlchemy engine instance
dia_object_index : `str`
    Indexing mode for DiaObject table, see `ApdbSqlConfig.dia_object_index`
    for details.
htm_index_column : `str`
    Name of a HTM index column for DiaObject and DiaSource tables.
schema_file : `str`
    Name of the YAML schema file.
schema_name : `str`, optional
    Name of the schema in YAML files.
prefix : `str`, optional
    Prefix to add to all schema elements.
namespace : `str`, optional
    Namespace (or schema name) to use for all APDB tables.
enable_replica : `bool`, optional
    If `True` then use additional tables for replica chunks.

Definition at line 80 of file apdbSqlSchema.py.

Constructor & Destructor Documentation

◆ init()

lsst.dax.apdb.sql.apdbSqlSchema.ApdbSqlSchema.__init__	(		self,
		sqlalchemy.engine.Engine	engine,
		str	dia_object_index,
		str	htm_index_column,
		str	schema_file,
		str	schema_name = "ApdbSchema",
		str	prefix = "",
		str \| None	namespace = None,
		bool	enable_replica = False )

Reimplemented from lsst.dax.apdb.apdbSchema.ApdbSchema.

Definition at line 120 of file apdbSqlSchema.py.

    ):
        super().__init__(schema_file, schema_name)
 
        self._engine = engine
        self._dia_object_index = dia_object_index
        self._htm_index_column = htm_index_column
        self._prefix = prefix
        self._enable_replica = enable_replica
 
        self._metadata = sqlalchemy.schema.MetaData(schema=namespace)
 
        # Add pixelId column and index to tables that need it
        for table in self.pixel_id_tables:
            tableDef = self.tableSchemas.get(table)
            if not tableDef:
                continue
            column = schema_model.Column(
                id=f"#{htm_index_column}",
                name=htm_index_column,
                datatype=felis.datamodel.DataType.long,
                nullable=False,
                value=None,
                description="Pixelization index column.",
                table=tableDef,
            )
            tableDef.columns.append(column)
 
            # Adjust index if needed
            if table == ApdbTables.DiaObject and self._dia_object_index == "pix_id_iov":
                tableDef.primary_key.insert(0, column)
 
            if table is ApdbTables.DiaObjectLast:
                # use it as a leading PK column
                tableDef.primary_key.insert(0, column)
            else:
                # make a regular index
                name = f"IDX_{tableDef.name}_{htm_index_column}"
                index = schema_model.Index(id=f"#{name}", name=name, columns=[column])
                tableDef.indexes.append(index)
 
        # generate schema for all tables, must be called last
        apdb_tables = self._make_apdb_tables()
        extra_tables = self._make_extra_tables(apdb_tables)
 
        converter = ModelToSql(metadata=self._metadata, prefix=self._prefix)
        id_to_table = converter.make_tables(itertools.chain(apdb_tables.values(), extra_tables.values()))
 
        self._apdb_tables = {
            apdb_enum: id_to_table[table_model.id] for apdb_enum, table_model in apdb_tables.items()
        }
        self._extra_tables = {
            extra_enum: id_to_table[table_model.id] for extra_enum, table_model in extra_tables.items()
        }
 
        self._has_replica_chunks: bool | None = None
        self._metadata_check: bool | None = None
 

Member Function Documentation

◆ _check_replica_chunks()

bool lsst.dax.apdb.sql.apdbSqlSchema.ApdbSqlSchema._check_replica_chunks ( self )

protected

Check whether database has tables for tracking insert IDs.

Definition at line 323 of file apdbSqlSchema.py.

    def _check_replica_chunks(self) -> bool:
        """Check whether database has tables for tracking insert IDs."""
        inspector = sqlalchemy.inspect(self._engine)
        db_tables = set(inspector.get_table_names(schema=self._metadata.schema))
        return ExtraTables.ApdbReplicaChunks.table_name(self._prefix) in db_tables
 

◆ _make_apdb_tables()

Mapping[ApdbTables, schema_model.Table] lsst.dax.apdb.sql.apdbSqlSchema.ApdbSqlSchema._make_apdb_tables	(		self,
		str	mysql_engine = "InnoDB" )

protected

Generate schema for regular tables.

Parameters
----------
mysql_engine : `str`, optional
    MySQL engine type to use for new tables.

Definition at line 329 of file apdbSqlSchema.py.

    def _make_apdb_tables(self, mysql_engine: str = "InnoDB") -> Mapping[ApdbTables, schema_model.Table]:
        """Generate schema for regular tables.
 
        Parameters
        ----------
        mysql_engine : `str`, optional
            MySQL engine type to use for new tables.
        """
        tables = {}
        for table_enum in ApdbTables:
            if table_enum is ApdbTables.DiaObjectLast and self._dia_object_index != "last_object_table":
                continue
            if table_enum is ApdbTables.metadata and table_enum not in self.tableSchemas:
                # Schema does not define metadata.
                continue
            table = self.tableSchemas[table_enum]
            tables[table_enum] = table
 
        return tables
 

◆ _make_extra_tables()

Mapping[ExtraTables, schema_model.Table] lsst.dax.apdb.sql.apdbSqlSchema.ApdbSqlSchema._make_extra_tables	(		self,
		Mapping[ApdbTables, schema_model.Table]	apdb_tables )

protected

Generate schema for insert ID tables.

Definition at line 349 of file apdbSqlSchema.py.

    ) -> Mapping[ExtraTables, schema_model.Table]:
        """Generate schema for insert ID tables."""
        if not self._enable_replica:
            return {}
 
        tables = {}
        column_defs: list[schema_model.Column] = [
            schema_model.Column(
                name="apdb_replica_chunk",
                id="#ApdbReplicaChunks.apdb_replica_chunk",
                datatype=felis.datamodel.DataType.long,
            ),
            schema_model.Column(
                name="last_update_time",
                id="#ApdbReplicaChunks.last_update_time",
                datatype=felis.datamodel.DataType.timestamp,
                nullable=False,
            ),
            schema_model.Column(
                name="unique_id",
                id="#ApdbReplicaChunks.unique_id",
                datatype=schema_model.ExtraDataTypes.UUID,
                nullable=False,
            ),
        ]
        parent_table = schema_model.Table(
            name=ExtraTables.ApdbReplicaChunks.table_name(self._prefix),
            id="#ApdbReplicaChunks",
            columns=column_defs,
            primary_key=[column_defs[0]],
            constraints=[],
            indexes=[],
        )
        tables[ExtraTables.ApdbReplicaChunks] = parent_table
 
        for table_enum, apdb_enum in ExtraTables.replica_chunk_tables().items():
            apdb_table = apdb_tables[apdb_enum]
            table_name = table_enum.table_name(self._prefix)
 
            columns = self._replicaChunkColumns(table_enum, apdb_enum)
            column_map = {column.name: column for column in columns}
            # PK is the same as for original table
            pk_columns = [column_map[column.name] for column in apdb_table.primary_key]
 
            indices = self._replicaChunkIndices(table_enum, column_map)
            constraints = self._replicaChunkConstraints(table_enum, apdb_table, parent_table, column_map)
            table = schema_model.Table(
                name=table_name,
                id=f"#{table_name}",
                columns=columns,
                primary_key=pk_columns,
                indexes=indices,
                constraints=constraints,
            )
            tables[table_enum] = table
 
        return tables
 

◆ _replicaChunkColumns()

list[schema_model.Column] lsst.dax.apdb.sql.apdbSqlSchema.ApdbSqlSchema._replicaChunkColumns	(		self,
		ExtraTables	table_enum,
		ApdbTables	apdb_enum )

protected

Return list of columns for replica chunks tables.

Definition at line 409 of file apdbSqlSchema.py.

    ) -> list[schema_model.Column]:
        """Return list of columns for replica chunks tables."""
        table_name = table_enum.table_name()
        column_defs: list[schema_model.Column] = [
            schema_model.Column(
                name="apdb_replica_chunk",
                id=f"#{table_name}.apdb_replica_chunk",
                datatype=felis.datamodel.DataType.long,
                nullable=False,
            )
        ]
        if table_enum in ExtraTables.replica_chunk_tables():
            table_model = self.tableSchemas[apdb_enum]
            column_defs += [column.clone() for column in table_model.primary_key]
        else:
            assert False, "Above branches have to cover all enum values"
        return column_defs
 

◆ _replicaChunkConstraints()

list[schema_model.Constraint] lsst.dax.apdb.sql.apdbSqlSchema.ApdbSqlSchema._replicaChunkConstraints	(		self,
		ExtraTables	table_enum,
		schema_model.Table	apdb_table,
		schema_model.Table	parent_table,
		Mapping[str, schema_model.Column]	column_map )

protected

Return set of constraints for replica chunk table.

Definition at line 443 of file apdbSqlSchema.py.

    ) -> list[schema_model.Constraint]:
        """Return set of constraints for replica chunk table."""
        constraints: list[schema_model.Constraint] = []
        replica_chunk_tables = ExtraTables.replica_chunk_tables()
        if table_enum in replica_chunk_tables:
            # Foreign key to original table
            name = f"{table_enum.table_name()}_fk_{apdb_table.name}"
            other_columns = apdb_table.primary_key
            this_columns = [column_map[column.name] for column in apdb_table.primary_key]
            constraints.append(
                schema_model.ForeignKeyConstraint(
                    name=name,
                    id=f"#{name}",
                    columns=this_columns,
                    referenced_columns=other_columns,
                    onupdate="CASCADE",
                    ondelete="CASCADE",
                )
            )
 
            # Foreign key to parent chunk ID table
            name = f"{table_enum.table_name()}_fk_{parent_table.name}"
            other_columns = parent_table.primary_key
            this_columns = [column_map[column.name] for column in parent_table.primary_key]
            constraints.append(
                schema_model.ForeignKeyConstraint(
                    name=name,
                    id=f"#{name}",
                    columns=this_columns,
                    referenced_columns=other_columns,
                    onupdate="CASCADE",
                    ondelete="CASCADE",
                )
            )
        return constraints

◆ _replicaChunkIndices()

list[schema_model.Index] lsst.dax.apdb.sql.apdbSqlSchema.ApdbSqlSchema._replicaChunkIndices	(		self,
		ExtraTables	table_enum,
		Mapping[str, schema_model.Column]	column_map )

protected

Return set of indices for replica chunk table.

Definition at line 429 of file apdbSqlSchema.py.

    ) -> list[schema_model.Index]:
        """Return set of indices for replica chunk table."""
        index_defs: list[schema_model.Index] = []
        if table_enum in ExtraTables.replica_chunk_tables():
            # Non-unique index on replica chunk column.
            name = self._prefix + table_enum.name + "_apdb_replica_chunk_idx"
            column = column_map["apdb_replica_chunk"]
            index_defs.append(schema_model.Index(name=name, id=f"#{name}", columns=[column]))
        return index_defs
 

◆ empty()

bool lsst.dax.apdb.sql.apdbSqlSchema.ApdbSqlSchema.empty ( self )

Return True if database schema is empty.

Returns
-------
empty : `bool`
    `True` if none of the required APDB tables exist in the database,
    `False` if all required tables exist.

Raises
------
InconsistentSchemaError
    Raised when some of the required tables exist but not all.

Definition at line 187 of file apdbSqlSchema.py.

    def empty(self) -> bool:
        """Return True if database schema is empty.
 
        Returns
        -------
        empty : `bool`
            `True` if none of the required APDB tables exist in the database,
            `False` if all required tables exist.
 
        Raises
        ------
        InconsistentSchemaError
            Raised when some of the required tables exist but not all.
        """
        inspector = sqlalchemy.inspect(self._engine)
        table_names = set(inspector.get_table_names(self._metadata.schema))
 
        existing_tables = []
        missing_tables = []
        for table_enum in self._apdb_tables:
            table_name = table_enum.table_name(self._prefix)
            if table_name in table_names:
                existing_tables.append(table_name)
            else:
                missing_tables.append(table_name)
 
        if not missing_tables:
            return False
        elif not existing_tables:
            return True
        else:
            raise InconsistentSchemaError(
                f"Only some required APDB tables exist: {existing_tables}, missing tables: {missing_tables}"
            )
 

◆ get_apdb_columns()

list[sqlalchemy.schema.Column] lsst.dax.apdb.sql.apdbSqlSchema.ApdbSqlSchema.get_apdb_columns	(		self,
		ApdbTables \| ExtraTables	table_enum )

Return list of columns defined for a table in APDB schema.

Returned list excludes columns that are implementation-specific, e.g.
``pixelId`` column is not include in the returned list.

Parameters
----------
table_enum : `ApdbTables` or `ExtraTables`
    Type of table.

Returns
-------
table : `list` [`sqlalchemy.schema.Column`]
    Table instance.

Raises
------
ValueError
    Raised if ``table_enum`` is not valid for this database.

Definition at line 289 of file apdbSqlSchema.py.

    def get_apdb_columns(self, table_enum: ApdbTables | ExtraTables) -> list[sqlalchemy.schema.Column]:
        """Return list of columns defined for a table in APDB schema.
 
        Returned list excludes columns that are implementation-specific, e.g.
        ``pixelId`` column is not include in the returned list.
 
        Parameters
        ----------
        table_enum : `ApdbTables` or `ExtraTables`
            Type of table.
 
        Returns
        -------
        table : `list` [`sqlalchemy.schema.Column`]
            Table instance.
 
        Raises
        ------
        ValueError
            Raised if ``table_enum`` is not valid for this database.
        """
        table = self.get_table(table_enum)
        exclude_columns = set()
        if table_enum in self.pixel_id_tables:
            exclude_columns.add(self._htm_index_column)
        return [column for column in table.columns if column.name not in exclude_columns]
 

◆ get_table()

sqlalchemy.schema.Table lsst.dax.apdb.sql.apdbSqlSchema.ApdbSqlSchema.get_table	(		self,
		ApdbTables \| ExtraTables	table_enum )

Return SQLAlchemy table instance for a specified table type/enum.

Parameters
----------
table_enum : `ApdbTables` or `ExtraTables`
    Type of table to return.

Returns
-------
table : `sqlalchemy.schema.Table`
    Table instance.

Raises
------
ValueError
    Raised if ``table_enum`` is not valid for this database.

Definition at line 251 of file apdbSqlSchema.py.

    def get_table(self, table_enum: ApdbTables | ExtraTables) -> sqlalchemy.schema.Table:
        """Return SQLAlchemy table instance for a specified table type/enum.
 
        Parameters
        ----------
        table_enum : `ApdbTables` or `ExtraTables`
            Type of table to return.
 
        Returns
        -------
        table : `sqlalchemy.schema.Table`
            Table instance.
 
        Raises
        ------
        ValueError
            Raised if ``table_enum`` is not valid for this database.
        """
        try:
            if isinstance(table_enum, ApdbTables):
                if table_enum is ApdbTables.metadata:
                    # There may be cases when schema is configured with the
                    # metadata table but database is still missing it. Check
                    # that table actually exists in the database. Note that
                    # this may interact with `makeSchema`.
                    if self._metadata_check is None:
                        inspector = sqlalchemy.inspect(self._engine)
                        table_name = table_enum.table_name(self._prefix)
                        self._metadata_check = inspector.has_table(table_name, schema=self._metadata.schema)
                    if not self._metadata_check:
                        # this will be caught below
                        raise LookupError("metadata table is missing")
                return self._apdb_tables[table_enum]
            else:
                return self._extra_tables[table_enum]
        except LookupError:
            raise ValueError(f"Table type {table_enum} does not exist in the schema") from None
 

◆ has_replica_chunks()

bool lsst.dax.apdb.sql.apdbSqlSchema.ApdbSqlSchema.has_replica_chunks ( self )

Whether insert ID tables are to be used (`bool`).

Definition at line 317 of file apdbSqlSchema.py.

    def has_replica_chunks(self) -> bool:
        """Whether insert ID tables are to be used (`bool`)."""
        if self._has_replica_chunks is None:
            self._has_replica_chunks = self._enable_replica and self._check_replica_chunks()
        return self._has_replica_chunks
 

◆ makeSchema()

None lsst.dax.apdb.sql.apdbSqlSchema.ApdbSqlSchema.makeSchema	(		self,
		bool	drop = False )

Create or re-create all tables.

Parameters
----------
drop : `bool`, optional
    If True then drop tables before creating new ones.

Definition at line 222 of file apdbSqlSchema.py.

    def makeSchema(self, drop: bool = False) -> None:
        """Create or re-create all tables.
 
        Parameters
        ----------
        drop : `bool`, optional
            If True then drop tables before creating new ones.
        """
        # Create namespace if it does not exist yet, for now this only makes
        # sense for postgres.
        if self._metadata.schema:
            dialect = self._engine.dialect
            quoted_schema = dialect.preparer(dialect).quote_schema(self._metadata.schema)
            create_schema = sqlalchemy.schema.DDL(
                "CREATE SCHEMA IF NOT EXISTS %(schema)s", context={"schema": quoted_schema}
            ).execute_if(dialect="postgresql")
            sqlalchemy.event.listen(self._metadata, "before_create", create_schema)
 
        # create all tables (optionally drop first)
        if drop:
            _LOG.info("dropping all tables")
            self._metadata.drop_all(self._engine)
        _LOG.info("creating all tables")
        self._metadata.create_all(self._engine)
 
        # Reset possibly cached value.
        self._has_replica_chunks = None
        self._metadata_check = None
 

Member Data Documentation

◆ _apdb_tables

lsst.dax.apdb.sql.apdbSqlSchema.ApdbSqlSchema._apdb_tables

protected

Definition at line 177 of file apdbSqlSchema.py.

◆ _dia_object_index

lsst.dax.apdb.sql.apdbSqlSchema.ApdbSqlSchema._dia_object_index

protected

Definition at line 134 of file apdbSqlSchema.py.

◆ _enable_replica

lsst.dax.apdb.sql.apdbSqlSchema.ApdbSqlSchema._enable_replica

protected

Definition at line 137 of file apdbSqlSchema.py.

◆ _engine

lsst.dax.apdb.sql.apdbSqlSchema.ApdbSqlSchema._engine

protected

Definition at line 133 of file apdbSqlSchema.py.

◆ _extra_tables

lsst.dax.apdb.sql.apdbSqlSchema.ApdbSqlSchema._extra_tables

protected

Definition at line 180 of file apdbSqlSchema.py.

◆ _has_replica_chunks

lsst.dax.apdb.sql.apdbSqlSchema.ApdbSqlSchema._has_replica_chunks

protected

Definition at line 248 of file apdbSqlSchema.py.

◆ _htm_index_column

lsst.dax.apdb.sql.apdbSqlSchema.ApdbSqlSchema._htm_index_column

protected

Definition at line 135 of file apdbSqlSchema.py.

◆ _metadata

lsst.dax.apdb.sql.apdbSqlSchema.ApdbSqlSchema._metadata

protected

Definition at line 139 of file apdbSqlSchema.py.

◆ _metadata_check

lsst.dax.apdb.sql.apdbSqlSchema.ApdbSqlSchema._metadata_check

protected

Definition at line 249 of file apdbSqlSchema.py.

◆ _prefix

lsst.dax.apdb.sql.apdbSqlSchema.ApdbSqlSchema._prefix

protected

Definition at line 136 of file apdbSqlSchema.py.

◆ pixel_id_tables

tuple lsst.dax.apdb.sql.apdbSqlSchema.ApdbSqlSchema.pixel_id_tables = (ApdbTables.DiaObject, ApdbTables.DiaObjectLast, ApdbTables.DiaSource)

static

Definition at line 117 of file apdbSqlSchema.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/g5ea96fc03c+33ab2bc355/python/lsst/dax/apdb/sql/apdbSqlSchema.py

Public Member Functions

Static Public Attributes

Protected Member Functions

Protected Attributes

Detailed Description

Constructor & Destructor Documentation

◆ __init__()

Member Function Documentation

◆ _check_replica_chunks()

◆ _make_apdb_tables()

◆ _make_extra_tables()

◆ _replicaChunkColumns()

◆ _replicaChunkConstraints()

◆ _replicaChunkIndices()

◆ empty()

◆ get_apdb_columns()

◆ get_table()

◆ has_replica_chunks()

◆ makeSchema()

Member Data Documentation

◆ _apdb_tables

◆ _dia_object_index

◆ _enable_replica

◆ _engine

◆ _extra_tables

◆ _has_replica_chunks

◆ _htm_index_column

◆ _metadata

◆ _metadata_check

◆ _prefix

◆ pixel_id_tables

◆ init()