lsst.dax.apdb.sql.apdbSqlReplica.ApdbSqlReplica Class Reference

Inheritance diagram for lsst.dax.apdb.sql.apdbSqlReplica.ApdbSqlReplica:

Public Member Functions
	__init__ (self, ApdbSqlSchema schema, sqlalchemy.engine.Engine engine, bool timer=False)
VersionTuple	apdbReplicaImplementationVersion (cls)
list[ReplicaChunk]|None	getReplicaChunks (self)
None	deleteReplicaChunks (self, Iterable[int] chunks)
ApdbTableData	getDiaObjectsChunks (self, Iterable[int] chunks)
ApdbTableData	getDiaSourcesChunks (self, Iterable[int] chunks)
ApdbTableData	getDiaForcedSourcesChunks (self, Iterable[int] chunks)

Protected Member Functions
Timer	_timer (self, str name, *Mapping[str, str|int]|None tags=None)
ApdbTableData	_get_chunks (self, Iterable[int] chunks, ApdbTables table_enum, ExtraTables chunk_table_enum)

Protected Attributes
	_schema
	_engine
	_timer_args

Detailed Description

Implementation of `ApdbReplica` for SQL backend.

Parameters
----------
schema : `ApdbSqlSchema`
    Instance of `ApdbSqlSchema` class for APDB database.
engine : `sqlalchemy.engine.Engine`
    Engine for database access.
timer : `bool`, optional
    If `True` then log timing information.

Definition at line 73 of file apdbSqlReplica.py.

Constructor & Destructor Documentation

__init__()

lsst.dax.apdb.sql.apdbSqlReplica.ApdbSqlReplica.__init__	(		self,
		ApdbSqlSchema	schema,
		sqlalchemy.engine.Engine	engine,
		bool	timer = False )

Definition at line 86 of file apdbSqlReplica.py.

    def __init__(self, schema: ApdbSqlSchema, engine: sqlalchemy.engine.Engine, timer: bool = False):
        self._schema = schema
        self._engine = engine
 
        self._timer_args: list[MonAgent | logging.Logger] = [_MON]
        if timer:
            self._timer_args.append(_LOG)
 

Member Function Documentation

_get_chunks()

ApdbTableData lsst.dax.apdb.sql.apdbSqlReplica.ApdbSqlReplica._get_chunks ( self, Iterable[int] chunks, ApdbTables table_enum, ExtraTables chunk_table_enum )

protected

Return catalog of records for given insert identifiers, common
implementation for all DIA tables.

Definition at line 146 of file apdbSqlReplica.py.

    ) -> ApdbTableData:
        """Return catalog of records for given insert identifiers, common
        implementation for all DIA tables.
        """
        if not self._schema.has_replica_chunks:
            raise ValueError("APDB is not configured for replication")
 
        table = self._schema.get_table(table_enum)
        chunk_table = self._schema.get_table(chunk_table_enum)
 
        join = table.join(chunk_table)
        chunk_id_column = chunk_table.columns["apdb_replica_chunk"]
        apdb_columns = self._schema.get_apdb_columns(table_enum)
        where_clause = chunk_id_column.in_(chunks)
        query = sql.select(chunk_id_column, *apdb_columns).select_from(join).where(where_clause)
 
        # execute select
        with self._timer("table_chunk_select_time", tags={"table": table.name}):
            with self._engine.begin() as conn:
                result = conn.execution_options(stream_results=True, max_row_buffer=10000).execute(query)
                return ApdbSqlTableData(result)

_timer()

Timer lsst.dax.apdb.sql.apdbSqlReplica.ApdbSqlReplica._timer ( self, str name, *Mapping[str, str | int] | None tags = None )

protected

Create `Timer` instance given its name.

Definition at line 94 of file apdbSqlReplica.py.

    def _timer(self, name: str, *, tags: Mapping[str, str | int] | None = None) -> Timer:
        """Create `Timer` instance given its name."""
        return Timer(name, *self._timer_args, tags=tags)
 

apdbReplicaImplementationVersion()

VersionTuple lsst.dax.apdb.sql.apdbSqlReplica.ApdbSqlReplica.apdbReplicaImplementationVersion

(

cls

)

Return version number for current ApdbReplica implementation.

Returns
-------
version : `VersionTuple`
    Version of the code defined in implementation class.

Reimplemented from lsst.dax.apdb.apdbReplica.ApdbReplica.

Definition at line 99 of file apdbSqlReplica.py.

    def apdbReplicaImplementationVersion(cls) -> VersionTuple:
        # Docstring inherited from base class.
        return VERSION
 

deleteReplicaChunks()

None lsst.dax.apdb.sql.apdbSqlReplica.ApdbSqlReplica.deleteReplicaChunks	(		self,
		Iterable[int]	chunks )

Remove replication chunks from the database.

Parameters
----------
chunks : `~collections.abc.Iterable` [`int`]
    Chunk identifiers to remove.

Notes
-----
This method causes Apdb to forget about specified chunks. If there
are any auxiliary data associated with the identifiers, it is also
removed from database (but data in regular tables is not removed).
This method should be called after successful transfer of data from
APDB to PPDB to free space used by replicas.

Reimplemented from lsst.dax.apdb.apdbReplica.ApdbReplica.

Definition at line 122 of file apdbSqlReplica.py.

    def deleteReplicaChunks(self, chunks: Iterable[int]) -> None:
        # docstring is inherited from a base class
        if not self._schema.has_replica_chunks:
            raise ValueError("APDB is not configured for replication")
 
        table = self._schema.get_table(ExtraTables.ApdbReplicaChunks)
        where_clause = table.columns["apdb_replica_chunk"].in_(chunks)
        stmt = table.delete().where(where_clause)
        with self._timer("chunks_delete_time"):
            with self._engine.begin() as conn:
                conn.execute(stmt)
 

getDiaForcedSourcesChunks()

ApdbTableData lsst.dax.apdb.sql.apdbSqlReplica.ApdbSqlReplica.getDiaForcedSourcesChunks	(		self,
		Iterable[int]	chunks )

Return catalog of DiaForcedSource records from given replica chunks.

Parameters
----------
chunks : `~collections.abc.Iterable` [`int`]
    Chunk identifiers to return.

Returns
-------
data : `ApdbTableData`
    Catalog containing DiaForcedSource records. In addition to all
    regular columns it will contain ``apdb_replica_chunk`` column.

Notes
-----
This part of API may not be very stable and can change before the
implementation finalizes.

Reimplemented from lsst.dax.apdb.apdbReplica.ApdbReplica.

Definition at line 142 of file apdbSqlReplica.py.

    def getDiaForcedSourcesChunks(self, chunks: Iterable[int]) -> ApdbTableData:
        # docstring is inherited from a base class
        return self._get_chunks(chunks, ApdbTables.DiaForcedSource, ExtraTables.DiaForcedSourceChunks)
 

getDiaObjectsChunks()

ApdbTableData lsst.dax.apdb.sql.apdbSqlReplica.ApdbSqlReplica.getDiaObjectsChunks	(		self,
		Iterable[int]	chunks )

Return catalog of DiaObject records from given replica chunks.

Parameters
----------
chunks : `~collections.abc.Iterable` [`int`]
    Chunk identifiers to return.

Returns
-------
data : `ApdbTableData`
    Catalog containing DiaObject records. In addition to all regular
    columns it will contain ``apdb_replica_chunk`` column.

Notes
-----
This part of API may not be very stable and can change before the
implementation finalizes.

Reimplemented from lsst.dax.apdb.apdbReplica.ApdbReplica.

Definition at line 134 of file apdbSqlReplica.py.

    def getDiaObjectsChunks(self, chunks: Iterable[int]) -> ApdbTableData:
        # docstring is inherited from a base class
        return self._get_chunks(chunks, ApdbTables.DiaObject, ExtraTables.DiaObjectChunks)
 

getDiaSourcesChunks()

ApdbTableData lsst.dax.apdb.sql.apdbSqlReplica.ApdbSqlReplica.getDiaSourcesChunks	(		self,
		Iterable[int]	chunks )

Return catalog of DiaSource records from given replica chunks.

Parameters
----------
chunks : `~collections.abc.Iterable` [`int`]
    Chunk identifiers to return.

Returns
-------
data : `ApdbTableData`
    Catalog containing DiaSource records. In addition to all regular
    columns it will contain ``apdb_replica_chunk`` column.

Notes
-----
This part of API may not be very stable and can change before the
implementation finalizes.

Reimplemented from lsst.dax.apdb.apdbReplica.ApdbReplica.

Definition at line 138 of file apdbSqlReplica.py.

    def getDiaSourcesChunks(self, chunks: Iterable[int]) -> ApdbTableData:
        # docstring is inherited from a base class
        return self._get_chunks(chunks, ApdbTables.DiaSource, ExtraTables.DiaSourceChunks)
 

getReplicaChunks()

list[ReplicaChunk] | None lsst.dax.apdb.sql.apdbSqlReplica.ApdbSqlReplica.getReplicaChunks

(

self

)

Return collection of replication chunks known to the database.

Returns
-------
chunks : `list` [`ReplicaChunk`] or `None`
    List of chunks, they may be time-ordered if database supports
    ordering. `None` is returned if database is not configured for
    replication.

Reimplemented from lsst.dax.apdb.apdbReplica.ApdbReplica.

Definition at line 103 of file apdbSqlReplica.py.

    def getReplicaChunks(self) -> list[ReplicaChunk] | None:
        # docstring is inherited from a base class
        if not self._schema.has_replica_chunks:
            return None
 
        table = self._schema.get_table(ExtraTables.ApdbReplicaChunks)
        assert table is not None, "has_replica_chunks=True means it must be defined"
        query = sql.select(
            table.columns["apdb_replica_chunk"], table.columns["last_update_time"], table.columns["unique_id"]
        ).order_by(table.columns["last_update_time"])
        with self._timer("chunks_select_time"):
            with self._engine.connect() as conn:
                result = conn.execution_options(stream_results=True, max_row_buffer=10000).execute(query)
                ids = []
                for row in result:
                    last_update_time = astropy.time.Time(row[1].timestamp(), format="unix_tai")
                    ids.append(ReplicaChunk(id=row[0], last_update_time=last_update_time, unique_id=row[2]))
                return ids
 

Member Data Documentation

_engine

lsst.dax.apdb.sql.apdbSqlReplica.ApdbSqlReplica._engine

protected

Definition at line 88 of file apdbSqlReplica.py.

_schema

lsst.dax.apdb.sql.apdbSqlReplica.ApdbSqlReplica._schema

protected

Definition at line 87 of file apdbSqlReplica.py.

_timer_args

lsst.dax.apdb.sql.apdbSqlReplica.ApdbSqlReplica._timer_args

protected

Definition at line 96 of file apdbSqlReplica.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/sql/apdbSqlReplica.py