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
ApdbTableData	_get_chunks (self, Iterable[int] chunks, ApdbTables table_enum, ExtraTables chunk_table_enum)

Protected Attributes
	_schema

	_engine

	_timer

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 70 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 83 of file apdbSqlReplica.py.

    def __init__(self, schema: ApdbSqlSchema, engine: sqlalchemy.engine.Engine, timer: bool = False):
        self._schema = schema
        self._engine = engine
        self._timer = timer
 

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 135 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 Timer(f"{table.name} replica chunk select", self._timer):
            with self._engine.begin() as conn:
                result = conn.execution_options(stream_results=True, max_row_buffer=10000).execute(query)
                return ApdbSqlTableData(result)

◆ 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 89 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 112 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._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 131 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 123 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 127 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 93 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 Timer("DiaObject insert id select", self._timer):
            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 85 of file apdbSqlReplica.py.

◆ _schema

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

protected

Definition at line 84 of file apdbSqlReplica.py.

◆ _timer

lsst.dax.apdb.sql.apdbSqlReplica.ApdbSqlReplica._timer

protected

Definition at line 86 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/g295015adf3+81dd352a9d/python/lsst/dax/apdb/sql/apdbSqlReplica.py

Public Member Functions

Protected Member Functions

Protected Attributes

Detailed Description

Constructor & Destructor Documentation

◆ __init__()

Member Function Documentation

◆ _get_chunks()

◆ apdbReplicaImplementationVersion()

◆ deleteReplicaChunks()

◆ getDiaForcedSourcesChunks()

◆ getDiaObjectsChunks()

◆ getDiaSourcesChunks()

◆ getReplicaChunks()

Member Data Documentation

◆ _engine

◆ _schema

◆ _timer

◆ init()