Inheritance diagram for lsst.dax.apdb.sql.apdbSql.ApdbSql:

Public Member Functions
	__init__ (self, ApdbSqlConfig config)

VersionTuple	apdbImplementationVersion (cls)

ApdbSqlConfig	init_database (cls, str db_url, *, str\|None schema_file=None, str\|None ss_schema_file=None, int\|None read_sources_months=None, int\|None read_forced_sources_months=None, bool enable_replica=False, int\|None connection_timeout=None, str\|None dia_object_index=None, int\|None htm_level=None, str\|None htm_index_column=None, tuple[str, str]\|None ra_dec_columns=None, str\|None prefix=None, str\|None namespace=None, bool drop=False)

ApdbSqlReplica	get_replica (self)

dict[str, int]	tableRowCount (self)

ApdbSqlConfig	getConfig (self)

Table\|None	tableDef (self, ApdbTables table)

pandas.DataFrame	getDiaObjects (self, Region region)

pandas.DataFrame\|None	getDiaSources (self, Region region, Iterable[int]\|None object_ids, astropy.time.Time visit_time, astropy.time.Time\|None start_time=None)

pandas.DataFrame\|None	getDiaForcedSources (self, Region region, Iterable[int]\|None object_ids, astropy.time.Time visit_time, astropy.time.Time\|None start_time=None)

bool	containsVisitDetector (self, int visit, int detector, Region region, astropy.time.Time visit_time)

None	store (self, astropy.time.Time visit_time, pandas.DataFrame objects, pandas.DataFrame\|None sources=None, pandas.DataFrame\|None forced_sources=None)

None	reassignDiaSources (self, Mapping[int, int] idMap)

None	dailyJob (self)

int	countUnassociatedObjects (self)

ApdbMetadata	metadata (self)

ApdbSqlAdmin	admin (self)

Apdb	from_config (cls, ApdbConfig config)

Apdb	from_uri (cls, ResourcePathExpression uri)

Public Attributes
	config = freezer.update(config, config_json)

	pixelator = HtmPixelization(self.config.pixelization.htm_level)

Static Public Attributes
str	metadataSchemaVersionKey = "version:schema"

str	metadataCodeVersionKey = "version:ApdbSql"

str	metadataReplicaVersionKey = "version:ApdbSqlReplica"

str	metadataConfigKey = "config:apdb-sql.json"

Protected Member Functions
Timer	_timer (self, str name, *, Mapping[str, str\|int]\|None tags=None)

sqlalchemy.engine.Engine	_makeEngine (cls, ApdbSqlConfig config, *, bool create)

sqlalchemy.engine.URL\|str	_connection_url (cls, str config_url, *, bool create)

str	_update_sqlite_url (cls, str url_string)

VersionTuple	_versionCheck (self, ApdbMetadataSql metadata)

None	_makeSchema (cls, ApdbConfig config, bool drop=False)

pandas.DataFrame	_getDiaSourcesInRegion (self, Region region, float start_time_mjdTai)

pandas.DataFrame	_getDiaSourcesByIDs (self, list[int] object_ids, float start_time_mjdTai)

pandas.DataFrame	_getSourcesByIDs (self, ApdbTables table_enum, list[int] object_ids, float midpointMjdTai_start)

None	_storeReplicaChunk (self, ReplicaChunk replica_chunk, sqlalchemy.engine.Connection connection)

None	_storeDiaObjects (self, pandas.DataFrame objs, astropy.time.Time visit_time, ReplicaChunk\|None replica_chunk, sqlalchemy.engine.Connection connection)

None	_storeDiaSources (self, pandas.DataFrame sources, ReplicaChunk\|None replica_chunk, sqlalchemy.engine.Connection connection)

None	_storeDiaForcedSources (self, pandas.DataFrame sources, ReplicaChunk\|None replica_chunk, sqlalchemy.engine.Connection connection)

None	_storeUpdateRecords (self, Iterable[ApdbUpdateRecord] records, ReplicaChunk chunk, *, bool store_chunk=False, sqlalchemy.engine.Connection\|None connection=None)

list[tuple[int, int]]	_htm_indices (self, Region region)

sql.ColumnElement	_filterRegion (self, sqlalchemy.schema.Table table, Region region)

pandas.DataFrame	_add_spatial_index (self, pandas.DataFrame df)

pandas.DataFrame	_fix_input_timestamps (self, pandas.DataFrame df)

pandas.DataFrame	_fix_result_timestamps (self, pandas.DataFrame df)

Protected Attributes
sqlalchemy.engine.Engine	_engine = self._makeEngine(config, create=False)

	_metadata = ApdbMetadataSql(self._engine, meta_table)

	_schema

VersionTuple	_db_schema_version = self._versionCheck(self._metadata)

Static Protected Attributes
tuple	_frozen_parameters

Detailed Description

Implementation of APDB interface based on SQL database.

The implementation is configured via standard ``pex_config`` mechanism
using `ApdbSqlConfig` configuration class. For an example of different
configurations check ``config/`` folder.

Parameters
----------
config : `ApdbSqlConfig`
    Configuration object.

Definition at line 118 of file apdbSql.py.

Constructor & Destructor Documentation

◆ init()

lsst.dax.apdb.sql.apdbSql.ApdbSql.__init__	(		self,
		ApdbSqlConfig	config )

Definition at line 152 of file apdbSql.py.

    def __init__(self, config: ApdbSqlConfig):
        self._engine = self._makeEngine(config, create=False)
 
        sa_metadata = sqlalchemy.MetaData(schema=config.namespace)
        meta_table_name = ApdbTables.metadata.table_name(prefix=config.prefix)
        meta_table = sqlalchemy.schema.Table(meta_table_name, sa_metadata, autoload_with=self._engine)
        self._metadata = ApdbMetadataSql(self._engine, meta_table)
 
        # Read frozen config from metadata.
        config_json = self._metadata.get(self.metadataConfigKey)
        if config_json is not None:
            # Update config from metadata.
            freezer = ApdbConfigFreezer[ApdbSqlConfig](self._frozen_parameters)
            self.config = freezer.update(config, config_json)
        else:
            self.config = config
 
        self._schema = ApdbSqlSchema(
            engine=self._engine,
            dia_object_index=self.config.dia_object_index,
            schema_file=self.config.schema_file,
            ss_schema_file=self.config.ss_schema_file,
            prefix=self.config.prefix,
            namespace=self.config.namespace,
            htm_index_column=self.config.pixelization.htm_index_column,
            enable_replica=self.config.enable_replica,
        )
 
        self._db_schema_version = self._versionCheck(self._metadata)
 
        self.pixelator = HtmPixelization(self.config.pixelization.htm_level)
 
        if _LOG.isEnabledFor(logging.DEBUG):
            _LOG.debug("ApdbSql Configuration: %s", self.config.model_dump())
 

Member Function Documentation

◆ _add_spatial_index()

pandas.DataFrame lsst.dax.apdb.sql.apdbSql.ApdbSql._add_spatial_index	(		self,
		pandas.DataFrame	df )

protected

Calculate spatial index for each record and add it to a DataFrame.

Parameters
----------
df : `pandas.DataFrame`
    DataFrame which has to contain ra/dec columns, names of these
    columns are defined by configuration ``ra_dec_columns`` field.

Returns
-------
df : `pandas.DataFrame`
    DataFrame with ``pixelId`` column which contains pixel index
    for ra/dec coordinates.

Notes
-----
This overrides any existing column in a DataFrame with the same name
(pixelId). Original DataFrame is not changed, copy of a DataFrame is
returned.

Definition at line 1204 of file apdbSql.py.

    def _add_spatial_index(self, df: pandas.DataFrame) -> pandas.DataFrame:
        """Calculate spatial index for each record and add it to a DataFrame.
 
        Parameters
        ----------
        df : `pandas.DataFrame`
            DataFrame which has to contain ra/dec columns, names of these
            columns are defined by configuration ``ra_dec_columns`` field.
 
        Returns
        -------
        df : `pandas.DataFrame`
            DataFrame with ``pixelId`` column which contains pixel index
            for ra/dec coordinates.
 
        Notes
        -----
        This overrides any existing column in a DataFrame with the same name
        (pixelId). Original DataFrame is not changed, copy of a DataFrame is
        returned.
        """
        # calculate HTM index for every DiaObject
        htm_index = np.zeros(df.shape[0], dtype=np.int64)
        ra_col, dec_col = self.config.ra_dec_columns
        for i, (ra, dec) in enumerate(zip(df[ra_col], df[dec_col])):
            uv3d = UnitVector3d(LonLat.fromDegrees(ra, dec))
            idx = self.pixelator.index(uv3d)
            htm_index[i] = idx
        df = df.copy()
        df[self.config.pixelization.htm_index_column] = htm_index
        return df
 

◆ _connection_url()

sqlalchemy.engine.URL \| str lsst.dax.apdb.sql.apdbSql.ApdbSql._connection_url	(		cls,
		str	config_url,
		*	,
		bool	create )

protected

Generate a complete URL for database with proper credentials.

Parameters
----------
config_url : `str`
    Database URL as specified in configuration.
create : `bool`
    Whether to try to create new database file, only relevant for
    SQLite backend which always creates new files by default.

Returns
-------
connection_url : `sqlalchemy.engine.URL` or `str`
    Connection URL including credentials.

Definition at line 229 of file apdbSql.py.

    def _connection_url(cls, config_url: str, *, create: bool) -> sqlalchemy.engine.URL | str:
        """Generate a complete URL for database with proper credentials.
 
        Parameters
        ----------
        config_url : `str`
            Database URL as specified in configuration.
        create : `bool`
            Whether to try to create new database file, only relevant for
            SQLite backend which always creates new files by default.
 
        Returns
        -------
        connection_url : `sqlalchemy.engine.URL` or `str`
            Connection URL including credentials.
        """
        # Allow 3rd party authentication mechanisms by assuming connection
        # string is correct when we can not recognize (dialect, host, database)
        # matching keys.
        components = urllib.parse.urlparse(config_url)
        if all((components.scheme is not None, components.hostname is not None, components.path is not None)):
            try:
                db_auth = DbAuth()
                config_url = db_auth.getUrl(config_url)
            except DbAuthNotFoundError:
                # Credentials file doesn't exist or no matching credentials,
                # use default auth.
                pass
 
        # SQLite has a nasty habit creating empty databases when they do not
        # exist, tell it not to do that unless we do need to create it.
        if not create:
            config_url = cls._update_sqlite_url(config_url)
 
        return config_url
 

◆ _filterRegion()

sql.ColumnElement lsst.dax.apdb.sql.apdbSql.ApdbSql._filterRegion	(		self,
		sqlalchemy.schema.Table	table,
		Region	region )

protected

Make SQLAlchemy expression for selecting records in a region.

Definition at line 1190 of file apdbSql.py.

    def _filterRegion(self, table: sqlalchemy.schema.Table, region: Region) -> sql.ColumnElement:
        """Make SQLAlchemy expression for selecting records in a region."""
        htm_index_column = table.columns[self.config.pixelization.htm_index_column]
        exprlist = []
        pixel_ranges = self._htm_indices(region)
        for low, upper in pixel_ranges:
            upper -= 1
            if low == upper:
                exprlist.append(htm_index_column == low)
            else:
                exprlist.append(sql.expression.between(htm_index_column, low, upper))
 
        return sql.expression.or_(*exprlist)
 

◆ _fix_input_timestamps()

pandas.DataFrame lsst.dax.apdb.sql.apdbSql.ApdbSql._fix_input_timestamps	(		self,
		pandas.DataFrame	df )

protected

Update timestamp columns in input DataFrame to be aware datetime
type in in UTC.

AP pipeline generates naive datetime instances, we want them to be
aware before they go to database. All naive timestamps are assumed to
be in UTC timezone (they should be TAI).

Definition at line 1236 of file apdbSql.py.

    def _fix_input_timestamps(self, df: pandas.DataFrame) -> pandas.DataFrame:
        """Update timestamp columns in input DataFrame to be aware datetime
        type in in UTC.
 
        AP pipeline generates naive datetime instances, we want them to be
        aware before they go to database. All naive timestamps are assumed to
        be in UTC timezone (they should be TAI).
        """
        # Find all columns with aware non-UTC timestamps and convert to UTC.
        columns = [
            column
            for column, dtype in df.dtypes.items()
            if isinstance(dtype, pandas.DatetimeTZDtype) and dtype.tz is not datetime.UTC
        ]
        for column in columns:
            df[column] = df[column].dt.tz_convert(datetime.UTC)
        # Find all columns with naive timestamps and add UTC timezone.
        columns = [
            column for column, dtype in df.dtypes.items() if pandas.api.types.is_datetime64_dtype(dtype)
        ]
        for column in columns:
            df[column] = df[column].dt.tz_localize(datetime.UTC)
        return df
 

◆ _fix_result_timestamps()

pandas.DataFrame lsst.dax.apdb.sql.apdbSql.ApdbSql._fix_result_timestamps	(		self,
		pandas.DataFrame	df )

protected

Update timestamp columns to be naive datetime type in returned
DataFrame.

AP pipeline code expects DataFrames to contain naive datetime columns,
while Postgres queries return timezone-aware type. This method converts
those columns to naive datetime in UTC timezone.

Definition at line 1260 of file apdbSql.py.

    def _fix_result_timestamps(self, df: pandas.DataFrame) -> pandas.DataFrame:
        """Update timestamp columns to be naive datetime type in returned
        DataFrame.
 
        AP pipeline code expects DataFrames to contain naive datetime columns,
        while Postgres queries return timezone-aware type. This method converts
        those columns to naive datetime in UTC timezone.
        """
        # Find all columns with aware timestamps.
        columns = [column for column, dtype in df.dtypes.items() if isinstance(dtype, pandas.DatetimeTZDtype)]
        for column in columns:
            # tz_convert(None) will convert to UTC and drop timezone.
            df[column] = df[column].dt.tz_convert(None)
        return df

◆ _getDiaSourcesByIDs()

pandas.DataFrame lsst.dax.apdb.sql.apdbSql.ApdbSql._getDiaSourcesByIDs	(		self,
		list[int]	object_ids,
		float	start_time_mjdTai )

protected

Return catalog of DiaSource instances given set of DiaObject IDs.

Parameters
----------
object_ids :
    Collection of DiaObject IDs
start_time_mjdTai : `float`
    Lower bound of time window for the query.

Returns
-------
catalog : `pandas.DataFrame`
    Catalog containing DiaSource records.

Definition at line 784 of file apdbSql.py.

    def _getDiaSourcesByIDs(self, object_ids: list[int], start_time_mjdTai: float) -> pandas.DataFrame:
        """Return catalog of DiaSource instances given set of DiaObject IDs.
 
        Parameters
        ----------
        object_ids :
            Collection of DiaObject IDs
        start_time_mjdTai : `float`
            Lower bound of time window for the query.
 
        Returns
        -------
        catalog : `pandas.DataFrame`
            Catalog containing DiaSource records.
        """
        with self._timer("select_time", tags={"table": "DiaSource"}) as timer:
            sources = self._getSourcesByIDs(ApdbTables.DiaSource, object_ids, start_time_mjdTai)
            timer.add_values(row_count=len(sources))
 
        _LOG.debug("found %s DiaSources", len(sources))
        return sources
 

◆ _getDiaSourcesInRegion()

pandas.DataFrame lsst.dax.apdb.sql.apdbSql.ApdbSql._getDiaSourcesInRegion	(		self,
		Region	region,
		float	start_time_mjdTai )

protected

Return catalog of DiaSource instances from given region.

Parameters
----------
region : `lsst.sphgeom.Region`
    Region to search for DIASources.
start_time_mjdTai : `float`
    Lower bound of time window for the query.

Returns
-------
catalog : `pandas.DataFrame`
    Catalog containing DiaSource records.

Definition at line 752 of file apdbSql.py.

    def _getDiaSourcesInRegion(self, region: Region, start_time_mjdTai: float) -> pandas.DataFrame:
        """Return catalog of DiaSource instances from given region.
 
        Parameters
        ----------
        region : `lsst.sphgeom.Region`
            Region to search for DIASources.
        start_time_mjdTai : `float`
            Lower bound of time window for the query.
 
        Returns
        -------
        catalog : `pandas.DataFrame`
            Catalog containing DiaSource records.
        """
        table = self._schema.get_table(ApdbTables.DiaSource)
        columns = self._schema.get_apdb_columns(ApdbTables.DiaSource)
        query = sql.select(*columns)
 
        # build selection
        time_filter = table.columns["midpointMjdTai"] > start_time_mjdTai
        where = sql.expression.and_(self._filterRegion(table, region), time_filter)
        query = query.where(where)
 
        # execute select
        with self._timer("DiaSource_select_time", tags={"table": "DiaSource"}) as timer:
            with self._engine.begin() as conn:
                sources = pandas.read_sql_query(query, conn)
            timer.add_values(row_counts=len(sources))
        _LOG.debug("found %s DiaSources", len(sources))
        return self._fix_result_timestamps(sources)
 

◆ _getSourcesByIDs()

pandas.DataFrame lsst.dax.apdb.sql.apdbSql.ApdbSql._getSourcesByIDs	(		self,
		ApdbTables	table_enum,
		list[int]	object_ids,
		float	midpointMjdTai_start )

protected

Return catalog of DiaSource or DiaForcedSource instances given set
of DiaObject IDs.

Parameters
----------
table : `sqlalchemy.schema.Table`
    Database table.
object_ids :
    Collection of DiaObject IDs
midpointMjdTai_start : `float`
    Earliest midpointMjdTai to retrieve.

Returns
-------
catalog : `pandas.DataFrame`
    Catalog contaning DiaSource records. `None` is returned if
    ``read_sources_months`` configuration parameter is set to 0 or
    when ``object_ids`` is empty.

Definition at line 806 of file apdbSql.py.

    ) -> pandas.DataFrame:
        """Return catalog of DiaSource or DiaForcedSource instances given set
        of DiaObject IDs.
 
        Parameters
        ----------
        table : `sqlalchemy.schema.Table`
            Database table.
        object_ids :
            Collection of DiaObject IDs
        midpointMjdTai_start : `float`
            Earliest midpointMjdTai to retrieve.
 
        Returns
        -------
        catalog : `pandas.DataFrame`
            Catalog contaning DiaSource records. `None` is returned if
            ``read_sources_months`` configuration parameter is set to 0 or
            when ``object_ids`` is empty.
        """
        table = self._schema.get_table(table_enum)
        columns = self._schema.get_apdb_columns(table_enum)
 
        sources: pandas.DataFrame | None = None
        if len(object_ids) <= 0:
            _LOG.debug("ID list is empty, just fetch empty result")
            query = sql.select(*columns).where(sql.literal(False))
            with self._engine.begin() as conn:
                sources = pandas.read_sql_query(query, conn)
        else:
            data_frames: list[pandas.DataFrame] = []
            for ids in chunk_iterable(sorted(object_ids), 1000):
                query = sql.select(*columns)
 
                # Some types like np.int64 can cause issues with
                # sqlalchemy, convert them to int.
                int_ids = [int(oid) for oid in ids]
 
                # select by object id
                query = query.where(
                    sql.expression.and_(
                        table.columns["diaObjectId"].in_(int_ids),
                        table.columns["midpointMjdTai"] > midpointMjdTai_start,
                    )
                )
 
                # execute select
                with self._engine.begin() as conn:
                    data_frames.append(pandas.read_sql_query(query, conn))
 
            if len(data_frames) == 1:
                sources = data_frames[0]
            else:
                sources = pandas.concat(data_frames)
        assert sources is not None, "Catalog cannot be None"
        return self._fix_result_timestamps(sources)
 

◆ _htm_indices()

list[tuple[int, int]] lsst.dax.apdb.sql.apdbSql.ApdbSql._htm_indices	(		self,
		Region	region )

protected

Generate a set of HTM indices covering specified region.

Parameters
----------
region: `sphgeom.Region`
    Region that needs to be indexed.

Returns
-------
Sequence of ranges, range is a tuple (minHtmID, maxHtmID).

Definition at line 1173 of file apdbSql.py.

    def _htm_indices(self, region: Region) -> list[tuple[int, int]]:
        """Generate a set of HTM indices covering specified region.
 
        Parameters
        ----------
        region: `sphgeom.Region`
            Region that needs to be indexed.
 
        Returns
        -------
        Sequence of ranges, range is a tuple (minHtmID, maxHtmID).
        """
        _LOG.debug("region: %s", region)
        indices = self.pixelator.envelope(region, self.config.pixelization.htm_max_ranges)
 
        return indices.ranges()
 

◆ _makeEngine()

sqlalchemy.engine.Engine lsst.dax.apdb.sql.apdbSql.ApdbSql._makeEngine	(		cls,
		ApdbSqlConfig	config,
		*	,
		bool	create )

protected

Make SQLALchemy engine based on configured parameters.

Parameters
----------
config : `ApdbSqlConfig`
    Configuration object.
create : `bool`
    Whether to try to create new database file, only relevant for
    SQLite backend which always creates new files by default.

Definition at line 192 of file apdbSql.py.

    def _makeEngine(cls, config: ApdbSqlConfig, *, create: bool) -> sqlalchemy.engine.Engine:
        """Make SQLALchemy engine based on configured parameters.
 
        Parameters
        ----------
        config : `ApdbSqlConfig`
            Configuration object.
        create : `bool`
            Whether to try to create new database file, only relevant for
            SQLite backend which always creates new files by default.
        """
        # engine is reused between multiple processes, make sure that we don't
        # share connections by disabling pool (by using NullPool class)
        kw: MutableMapping[str, Any] = dict(config.connection_config.extra_parameters)
        conn_args: dict[str, Any] = {}
        if not config.connection_config.connection_pool:
            kw.update(poolclass=NullPool)
        if config.connection_config.isolation_level is not None:
            kw.update(isolation_level=config.connection_config.isolation_level)
        elif config.db_url.startswith("sqlite"):
            # Use READ_UNCOMMITTED as default value for sqlite.
            kw.update(isolation_level="READ_UNCOMMITTED")
        if config.connection_config.connection_timeout is not None:
            if config.db_url.startswith("sqlite"):
                conn_args.update(timeout=config.connection_config.connection_timeout)
            elif config.db_url.startswith(("postgresql", "mysql")):
                conn_args.update(connect_timeout=int(config.connection_config.connection_timeout))
        kw.update(connect_args=conn_args)
        engine = sqlalchemy.create_engine(cls._connection_url(config.db_url, create=create), **kw)
 
        if engine.dialect.name == "sqlite":
            # Need to enable foreign keys on every new connection.
            sqlalchemy.event.listen(engine, "connect", _onSqlite3Connect)
 
        return engine
 

◆ _makeSchema()

None lsst.dax.apdb.sql.apdbSql.ApdbSql._makeSchema	(		cls,
		ApdbConfig	config,
		bool	drop = False )

protected

Definition at line 493 of file apdbSql.py.

    def _makeSchema(cls, config: ApdbConfig, drop: bool = False) -> None:
        # docstring is inherited from a base class
 
        if not isinstance(config, ApdbSqlConfig):
            raise TypeError(f"Unexpected type of configuration object: {type(config)}")
 
        engine = cls._makeEngine(config, create=True)
 
        # Ask schema class to create all tables.
        schema = ApdbSqlSchema(
            engine=engine,
            dia_object_index=config.dia_object_index,
            schema_file=config.schema_file,
            ss_schema_file=config.ss_schema_file,
            prefix=config.prefix,
            namespace=config.namespace,
            htm_index_column=config.pixelization.htm_index_column,
            enable_replica=config.enable_replica,
        )
        schema.makeSchema(drop=drop)
 
        # Need metadata table to store few items in it.
        meta_table = schema.get_table(ApdbTables.metadata)
        apdb_meta = ApdbMetadataSql(engine, meta_table)
 
        # Fill version numbers, overwrite if they are already there.
        apdb_meta.set(cls.metadataSchemaVersionKey, str(schema.schemaVersion()), force=True)
        apdb_meta.set(cls.metadataCodeVersionKey, str(cls.apdbImplementationVersion()), force=True)
        if config.enable_replica:
            # Only store replica code version if replica is enabled.
            apdb_meta.set(
                cls.metadataReplicaVersionKey,
                str(ApdbSqlReplica.apdbReplicaImplementationVersion()),
                force=True,
            )
 
        # Store frozen part of a configuration in metadata.
        freezer = ApdbConfigFreezer[ApdbSqlConfig](cls._frozen_parameters)
        apdb_meta.set(cls.metadataConfigKey, freezer.to_json(config), force=True)
 

◆ _storeDiaForcedSources()

None lsst.dax.apdb.sql.apdbSql.ApdbSql._storeDiaForcedSources	(		self,
		pandas.DataFrame	sources,
		ReplicaChunk \| None	replica_chunk,
		sqlalchemy.engine.Connection	connection )

protected

Store a set of DiaForcedSources from current visit.

Parameters
----------
sources : `pandas.DataFrame`
    Catalog containing DiaForcedSource records

Definition at line 1067 of file apdbSql.py.

    ) -> None:
        """Store a set of DiaForcedSources from current visit.
 
        Parameters
        ----------
        sources : `pandas.DataFrame`
            Catalog containing DiaForcedSource records
        """
        table = self._schema.get_table(ApdbTables.DiaForcedSource)
 
        # Insert replica data
        replica_data: list[dict] = []
        replica_stmt: Any = None
        replica_table_name = ""
        if replica_chunk is not None:
            pk_names = [column.name for column in table.primary_key]
            replica_data = sources[pk_names].to_dict("records")
            if replica_data:
                for row in replica_data:
                    row["apdb_replica_chunk"] = replica_chunk.id
                replica_table = self._schema.get_table(ExtraTables.DiaForcedSourceChunks)
                replica_table_name = replica_table.name
                replica_stmt = replica_table.insert()
 
        # everything to be done in single transaction
        with self._timer("insert_time", tags={"table": table.name}) as timer:
            sources = _coerce_uint64(sources)
            sources.to_sql(table.name, connection, if_exists="append", index=False, schema=table.schema)
            timer.add_values(row_count=len(sources))
        if replica_stmt is not None:
            with self._timer("insert_time", tags={"table": replica_table_name}) as timer:
                connection.execute(replica_stmt, replica_data)
                timer.add_values(row_count=len(replica_data))
 

◆ _storeDiaObjects()

None lsst.dax.apdb.sql.apdbSql.ApdbSql._storeDiaObjects	(		self,
		pandas.DataFrame	objs,
		astropy.time.Time	visit_time,
		ReplicaChunk \| None	replica_chunk,
		sqlalchemy.engine.Connection	connection )

protected

Store catalog of DiaObjects from current visit.

Parameters
----------
objs : `pandas.DataFrame`
    Catalog with DiaObject records.
visit_time : `astropy.time.Time`
    Time of the visit.
replica_chunk : `ReplicaChunk`
    Insert identifier.

Definition at line 891 of file apdbSql.py.

    ) -> None:
        """Store catalog of DiaObjects from current visit.
 
        Parameters
        ----------
        objs : `pandas.DataFrame`
            Catalog with DiaObject records.
        visit_time : `astropy.time.Time`
            Time of the visit.
        replica_chunk : `ReplicaChunk`
            Insert identifier.
        """
        if len(objs) == 0:
            _LOG.debug("No objects to write to database.")
            return
 
        # Some types like np.int64 can cause issues with sqlalchemy, convert
        # them to int.
        ids = sorted(int(oid) for oid in objs["diaObjectId"])
        _LOG.debug("first object ID: %d", ids[0])
 
        if self._schema.has_mjd_timestamps:
            validity_start_column = "validityStartMjdTai"
            validity_end_column = "validityEndMjdTai"
            timestamp = float(visit_time.tai.mjd)
        else:
            validity_start_column = "validityStart"
            validity_end_column = "validityEnd"
            timestamp = visit_time.datetime
 
        # everything to be done in single transaction
        if self.config.dia_object_index == "last_object_table":
            # Insert and replace all records in LAST table.
            table = self._schema.get_table(ApdbTables.DiaObjectLast)
 
            # DiaObjectLast did not have this column in the past.
            use_validity_start = self._schema.check_column(ApdbTables.DiaObjectLast, validity_start_column)
 
            # Drop the previous objects (pandas cannot upsert).
            query = table.delete().where(table.columns["diaObjectId"].in_(ids))
 
            with self._timer("delete_time", tags={"table": table.name}) as timer:
                res = connection.execute(query)
                timer.add_values(row_count=res.rowcount)
            _LOG.debug("deleted %s objects", res.rowcount)
 
            # DiaObjectLast is a subset of DiaObject, strip missing columns
            last_column_names = [column.name for column in table.columns]
            if validity_start_column in last_column_names and validity_start_column not in objs.columns:
                last_column_names.remove(validity_start_column)
            last_objs = objs[last_column_names]
            last_objs = _coerce_uint64(last_objs)
 
            # Fill validityStart, only when it is in the schema.
            if use_validity_start:
                if validity_start_column in last_objs:
                    last_objs[validity_start_column] = timestamp
                else:
                    extra_column = pandas.Series([timestamp] * len(last_objs), name=validity_start_column)
                    last_objs.set_index(extra_column.index, inplace=True)
                    last_objs = pandas.concat([last_objs, extra_column], axis="columns")
 
            with self._timer("insert_time", tags={"table": "DiaObjectLast"}) as timer:
                last_objs.to_sql(
                    table.name,
                    connection,
                    if_exists="append",
                    index=False,
                    schema=table.schema,
                )
                timer.add_values(row_count=len(last_objs))
        else:
            # truncate existing validity intervals
            table = self._schema.get_table(ApdbTables.DiaObject)
 
            update = (
                table.update()
                .values(**{validity_end_column: timestamp})
                .where(
                    sql.expression.and_(
                        table.columns["diaObjectId"].in_(ids),
                        table.columns[validity_end_column].is_(None),
                    )
                )
            )
 
            with self._timer("truncate_time", tags={"table": table.name}) as timer:
                res = connection.execute(update)
                timer.add_values(row_count=res.rowcount)
            _LOG.debug("truncated %s intervals", res.rowcount)
 
        objs = _coerce_uint64(objs)
 
        # Fill additional columns
        extra_columns: list[pandas.Series] = []
        if validity_start_column in objs.columns:
            objs[validity_start_column] = timestamp
        else:
            extra_columns.append(pandas.Series([timestamp] * len(objs), name=validity_start_column))
        if validity_end_column in objs.columns:
            objs[validity_end_column] = None
        else:
            extra_columns.append(pandas.Series([None] * len(objs), name=validity_end_column))
        if extra_columns:
            objs.set_index(extra_columns[0].index, inplace=True)
            objs = pandas.concat([objs] + extra_columns, axis="columns")
 
        # Insert replica data
        table = self._schema.get_table(ApdbTables.DiaObject)
        replica_data: list[dict] = []
        replica_stmt: Any = None
        replica_table_name = ""
        if replica_chunk is not None:
            pk_names = [column.name for column in table.primary_key]
            replica_data = objs[pk_names].to_dict("records")
            if replica_data:
                for row in replica_data:
                    row["apdb_replica_chunk"] = replica_chunk.id
                replica_table = self._schema.get_table(ExtraTables.DiaObjectChunks)
                replica_table_name = replica_table.name
                replica_stmt = replica_table.insert()
 
        # insert new versions
        with self._timer("insert_time", tags={"table": table.name}) as timer:
            objs.to_sql(table.name, connection, if_exists="append", index=False, schema=table.schema)
            timer.add_values(row_count=len(objs))
        if replica_stmt is not None:
            with self._timer("insert_time", tags={"table": replica_table_name}) as timer:
                connection.execute(replica_stmt, replica_data)
                timer.add_values(row_count=len(replica_data))
 

◆ _storeDiaSources()

None lsst.dax.apdb.sql.apdbSql.ApdbSql._storeDiaSources	(		self,
		pandas.DataFrame	sources,
		ReplicaChunk \| None	replica_chunk,
		sqlalchemy.engine.Connection	connection )

protected

Store catalog of DiaSources from current visit.

Parameters
----------
sources : `pandas.DataFrame`
    Catalog containing DiaSource records

Definition at line 1028 of file apdbSql.py.

    ) -> None:
        """Store catalog of DiaSources from current visit.
 
        Parameters
        ----------
        sources : `pandas.DataFrame`
            Catalog containing DiaSource records
        """
        table = self._schema.get_table(ApdbTables.DiaSource)
 
        # Insert replica data
        replica_data: list[dict] = []
        replica_stmt: Any = None
        replica_table_name = ""
        if replica_chunk is not None:
            pk_names = [column.name for column in table.primary_key]
            replica_data = sources[pk_names].to_dict("records")
            if replica_data:
                for row in replica_data:
                    row["apdb_replica_chunk"] = replica_chunk.id
                replica_table = self._schema.get_table(ExtraTables.DiaSourceChunks)
                replica_table_name = replica_table.name
                replica_stmt = replica_table.insert()
 
        # everything to be done in single transaction
        with self._timer("insert_time", tags={"table": table.name}) as timer:
            sources = _coerce_uint64(sources)
            sources.to_sql(table.name, connection, if_exists="append", index=False, schema=table.schema)
            timer.add_values(row_count=len(sources))
        if replica_stmt is not None:
            with self._timer("replica_insert_time", tags={"table": replica_table_name}) as timer:
                connection.execute(replica_stmt, replica_data)
                timer.add_values(row_count=len(replica_data))
 

◆ _storeReplicaChunk()

None lsst.dax.apdb.sql.apdbSql.ApdbSql._storeReplicaChunk	(		self,
		ReplicaChunk	replica_chunk,
		sqlalchemy.engine.Connection	connection )

protected

Definition at line 865 of file apdbSql.py.

    ) -> None:
        # `visit_time.datetime` returns naive datetime, even though all astropy
        # times are in UTC. Add UTC timezone to timestamp so that database
        # can store a correct value.
        dt = datetime.datetime.fromtimestamp(replica_chunk.last_update_time.unix_tai, tz=datetime.UTC)
 
        table = self._schema.get_table(ExtraTables.ApdbReplicaChunks)
 
        # We need UPSERT which is dialect-specific construct
        values = {"last_update_time": dt, "unique_id": replica_chunk.unique_id}
        row = {"apdb_replica_chunk": replica_chunk.id} | values
        if connection.dialect.name == "sqlite":
            insert_sqlite = sqlalchemy.dialects.sqlite.insert(table)
            insert_sqlite = insert_sqlite.on_conflict_do_update(index_elements=table.primary_key, set_=values)
            connection.execute(insert_sqlite, row)
        elif connection.dialect.name == "postgresql":
            insert_pg = sqlalchemy.dialects.postgresql.dml.insert(table)
            insert_pg = insert_pg.on_conflict_do_update(constraint=table.primary_key, set_=values)
            connection.execute(insert_pg, row)
        else:
            raise TypeError(f"Unsupported dialect {connection.dialect.name} for upsert.")
 

◆ _storeUpdateRecords()

None lsst.dax.apdb.sql.apdbSql.ApdbSql._storeUpdateRecords	(		self,
		Iterable[ApdbUpdateRecord]	records,
		ReplicaChunk	chunk,
		*	,
		bool	store_chunk = False,
		sqlalchemy.engine.Connection \| None	connection = None )

protected

Store ApdbUpdateRecords in the replica table for those records.

Parameters
----------
records : `list` [`ApdbUpdateRecord`]
    Records to store.
chunk : `ReplicaChunk`
    Replica chunk for these records.
store_chunk : `bool`
    If True then also store replica chunk.
connection : `sqlalchemy.engine.Connection`
    SQLALchemy connection to use, if `None` the new connection will be
    made. `None` is useful for tests only, regular use will call this
    method in the same transaction that saves other types of records.

Raises
------
TypeError
    Raised if replication is not enabled for this instance.

Definition at line 1106 of file apdbSql.py.

    ) -> None:
        """Store ApdbUpdateRecords in the replica table for those records.
 
        Parameters
        ----------
        records : `list` [`ApdbUpdateRecord`]
            Records to store.
        chunk : `ReplicaChunk`
            Replica chunk for these records.
        store_chunk : `bool`
            If True then also store replica chunk.
        connection : `sqlalchemy.engine.Connection`
            SQLALchemy connection to use, if `None` the new connection will be
            made. `None` is useful for tests only, regular use will call this
            method in the same transaction that saves other types of records.
 
        Raises
        ------
        TypeError
            Raised if replication is not enabled for this instance.
        """
        if not self._schema.replication_enabled:
            raise TypeError("Replication is not enabled for this APDB instance.")
 
        apdb_replica_chunk = chunk.id
        # Do not use unique_if from ReplicaChunk as it could be reused in
        # multiple calls to this method.
        update_unique_id = uuid.uuid4()
 
        record_dicts = []
        for record in records:
            record_dicts.append(
                {
                    "apdb_replica_chunk": apdb_replica_chunk,
                    "update_time_ns": record.update_time_ns,
                    "update_order": record.update_order,
                    "update_unique_id": update_unique_id,
                    "update_payload": record.to_json(),
                }
            )
 
        if not record_dicts:
            return
 
        # TODO: Need to check that table exists.
        table = self._schema.get_table(ExtraTables.ApdbUpdateRecordChunks)
 
        def _do_store(connection: sqlalchemy.engine.Connection) -> None:
            if store_chunk:
                self._storeReplicaChunk(chunk, connection)
            with self._timer("insert_time", tags={"table": table.name}) as timer:
                connection.execute(table.insert(), record_dicts)
                timer.add_values(row_count=len(record_dicts))
 
        if connection is None:
            with self._engine.begin() as connection:
                _do_store(connection)
        else:
            _do_store(connection)
 

◆ _timer()

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

protected

Create `Timer` instance given its name.

Definition at line 187 of file apdbSql.py.

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

◆ _update_sqlite_url()

str lsst.dax.apdb.sql.apdbSql.ApdbSql._update_sqlite_url	(		cls,
		str	url_string )

protected

If URL refers to sqlite dialect, update it so that the backend does
not try to create database file if it does not exist already.

Parameters
----------
url_string : `str`
    Connection string.

Returns
-------
url_string : `str`
    Possibly updated connection string.

Definition at line 266 of file apdbSql.py.

    def _update_sqlite_url(cls, url_string: str) -> str:
        """If URL refers to sqlite dialect, update it so that the backend does
        not try to create database file if it does not exist already.
 
        Parameters
        ----------
        url_string : `str`
            Connection string.
 
        Returns
        -------
        url_string : `str`
            Possibly updated connection string.
        """
        try:
            url = sqlalchemy.make_url(url_string)
        except sqlalchemy.exc.SQLAlchemyError:
            # If parsing fails it means some special format, likely not
            # sqlite so we just return it unchanged.
            return url_string
 
        if url.get_backend_name() == "sqlite":
            # Massage url so that database name starts with "file:" and
            # option string has "mode=rw&uri=true". Database name
            # should look like a path (:memory: is not supported by
            # Apdb, but someone could still try to use it).
            database = url.database
            if database and not database.startswith((":", "file:")):
                query = dict(url.query, mode="rw", uri="true")
                # If ``database`` is an absolute path then original URL should
                # include four slashes after "sqlite:". Humans are bad at
                # counting things beyond four and sometimes an extra slash gets
                # added unintentionally, which causes sqlite to treat initial
                # element as "authority" and to complain. Strip extra slashes
                # at the start of the path to avoid that (DM-46077).
                if database.startswith("//"):
                    warnings.warn(
                        f"Database URL contains extra leading slashes which will be removed: {url}",
                        stacklevel=3,
                    )
                    database = "/" + database.lstrip("/")
                url = url.set(database=f"file:{database}", query=query)
                url_string = url.render_as_string()
 
        return url_string
 

◆ _versionCheck()

VersionTuple lsst.dax.apdb.sql.apdbSql.ApdbSql._versionCheck	(		self,
		ApdbMetadataSql	metadata )

protected

Check schema version compatibility and return the database schema
version.

Definition at line 312 of file apdbSql.py.

    def _versionCheck(self, metadata: ApdbMetadataSql) -> VersionTuple:
        """Check schema version compatibility and return the database schema
        version.
        """
 
        def _get_version(key: str) -> VersionTuple:
            """Retrieve version number from given metadata key."""
            version_str = metadata.get(key)
            if version_str is None:
                # Should not happen with existing metadata table.
                raise RuntimeError(f"Version key {key!r} does not exist in metadata table.")
            return VersionTuple.fromString(version_str)
 
        db_schema_version = _get_version(self.metadataSchemaVersionKey)
        db_code_version = _get_version(self.metadataCodeVersionKey)
 
        # For now there is no way to make read-only APDB instances, assume that
        # any access can do updates.
        if not self._schema.schemaVersion().checkCompatibility(db_schema_version):
            raise IncompatibleVersionError(
                f"Configured schema version {self._schema.schemaVersion()} "
                f"is not compatible with database version {db_schema_version}"
            )
        if not self.apdbImplementationVersion().checkCompatibility(db_code_version):
            raise IncompatibleVersionError(
                f"Current code version {self.apdbImplementationVersion()} "
                f"is not compatible with database version {db_code_version}"
            )
 
        # Check replica code version only if replica is enabled.
        if self._schema.replication_enabled:
            db_replica_version = _get_version(self.metadataReplicaVersionKey)
            code_replica_version = ApdbSqlReplica.apdbReplicaImplementationVersion()
            if not code_replica_version.checkCompatibility(db_replica_version):
                raise IncompatibleVersionError(
                    f"Current replication code version {code_replica_version} "
                    f"is not compatible with database version {db_replica_version}"
                )
 
        return db_schema_version
 

◆ admin()

ApdbSqlAdmin lsst.dax.apdb.sql.apdbSql.ApdbSql.admin ( self )

Object providing adminitrative interface for APDB (`ApdbAdmin`).

Reimplemented from lsst.dax.apdb.apdb.Apdb.

Definition at line 748 of file apdbSql.py.

    def admin(self) -> ApdbSqlAdmin:
        # docstring is inherited from a base class
        return ApdbSqlAdmin(self.pixelator)
 

◆ apdbImplementationVersion()

VersionTuple lsst.dax.apdb.sql.apdbSql.ApdbSql.apdbImplementationVersion ( cls )

Return version number for current APDB implementation.

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

Definition at line 354 of file apdbSql.py.

    def apdbImplementationVersion(cls) -> VersionTuple:
        """Return version number for current APDB implementation.
 
        Returns
        -------
        version : `VersionTuple`
            Version of the code defined in implementation class.
        """
        return VERSION
 

◆ containsVisitDetector()

bool lsst.dax.apdb.sql.apdbSql.ApdbSql.containsVisitDetector	(		self,
		int	visit,
		int	detector,
		Region	region,
		astropy.time.Time	visit_time )

Test whether any sources for a given visit-detector are present in
the APDB.

Parameters
----------
visit, detector : `int`
    The ID of the visit-detector to search for.
region : `lsst.sphgeom.Region`
    Region corresponding to the visit/detector combination.
visit_time : `astropy.time.Time`
    Visit time (as opposed to visit processing time). This can be any
    timestamp in the visit timespan, e.g. its begin or end time.

Returns
-------
present : `bool`
    `True` if at least one DiaSource or DiaForcedSource record
    may exist for the specified observation, `False` otherwise.

Reimplemented from lsst.dax.apdb.apdb.Apdb.

Definition at line 628 of file apdbSql.py.

    ) -> bool:
        # docstring is inherited from a base class
        src_table: sqlalchemy.schema.Table = self._schema.get_table(ApdbTables.DiaSource)
        frcsrc_table: sqlalchemy.schema.Table = self._schema.get_table(ApdbTables.DiaForcedSource)
        # Query should load only one leaf page of the index
        query1 = sql.select(src_table.c.visit).filter_by(visit=visit, detector=detector).limit(1)
 
        with self._engine.begin() as conn:
            result = conn.execute(query1).scalar_one_or_none()
            if result is not None:
                return True
            else:
                # Backup query if an image was processed but had no diaSources
                query2 = sql.select(frcsrc_table.c.visit).filter_by(visit=visit, detector=detector).limit(1)
                result = conn.execute(query2).scalar_one_or_none()
                return result is not None
 

◆ countUnassociatedObjects()

int lsst.dax.apdb.sql.apdbSql.ApdbSql.countUnassociatedObjects ( self )

Return the number of DiaObjects that have only one DiaSource
associated with them.

Used as part of ap_verify metrics.

Returns
-------
count : `int`
    Number of DiaObjects with exactly one associated DiaSource.

Notes
-----
This method can be very inefficient or slow in some implementations.

Reimplemented from lsst.dax.apdb.apdb.Apdb.

Definition at line 721 of file apdbSql.py.

    def countUnassociatedObjects(self) -> int:
        # docstring is inherited from a base class
 
        # Retrieve the DiaObject table.
        table: sqlalchemy.schema.Table = self._schema.get_table(ApdbTables.DiaObject)
 
        if self._schema.has_mjd_timestamps:
            validity_end_column = "validityEndMjdTai"
        else:
            validity_end_column = "validityEnd"
 
        # Construct the sql statement.
        stmt = sql.select(func.count()).select_from(table).where(table.c.nDiaSources == 1)
        stmt = stmt.where(table.columns[validity_end_column] == None)  # noqa: E711
 
        # Return the count.
        with self._engine.begin() as conn:
            count = conn.execute(stmt).scalar_one()
 
        return count
 

◆ dailyJob()

None lsst.dax.apdb.sql.apdbSql.ApdbSql.dailyJob ( self )

Implement daily activities like cleanup/vacuum.

What should be done during daily activities is determined by
specific implementation.

Reimplemented from lsst.dax.apdb.apdb.Apdb.

Definition at line 717 of file apdbSql.py.

    def dailyJob(self) -> None:
        # docstring is inherited from a base class
        pass
 

◆ from_config()

Apdb lsst.dax.apdb.apdb.Apdb.from_config	(		cls,
		ApdbConfig	config )

inherited

Create Ppdb instance from configuration object.

Parameters
----------
config : `ApdbConfig`
    Configuration object, type of this object determines type of the
    Apdb implementation.

Returns
-------
apdb : `apdb`
    Instance of `Apdb` class.

Definition at line 50 of file apdb.py.

    def from_config(cls, config: ApdbConfig) -> Apdb:
        """Create Ppdb instance from configuration object.
 
        Parameters
        ----------
        config : `ApdbConfig`
            Configuration object, type of this object determines type of the
            Apdb implementation.
 
        Returns
        -------
        apdb : `apdb`
            Instance of `Apdb` class.
        """
        return make_apdb(config)
 

◆ from_uri()

Apdb lsst.dax.apdb.apdb.Apdb.from_uri	(		cls,
		ResourcePathExpression	uri )

inherited

Make Apdb instance from a serialized configuration.

Parameters
----------
uri : `~lsst.resources.ResourcePathExpression`
    URI or local file path pointing to a file with serialized
    configuration, or a string with a "label:" prefix. In the latter
    case, the configuration will be looked up from an APDB index file
    using the label name that follows the prefix. The APDB index file's
    location is determined by the ``DAX_APDB_INDEX_URI`` environment
    variable.

Returns
-------
apdb : `apdb`
    Instance of `Apdb` class, the type of the returned instance is
    determined by configuration.

Definition at line 67 of file apdb.py.

    def from_uri(cls, uri: ResourcePathExpression) -> Apdb:
        """Make Apdb instance from a serialized configuration.
 
        Parameters
        ----------
        uri : `~lsst.resources.ResourcePathExpression`
            URI or local file path pointing to a file with serialized
            configuration, or a string with a "label:" prefix. In the latter
            case, the configuration will be looked up from an APDB index file
            using the label name that follows the prefix. The APDB index file's
            location is determined by the ``DAX_APDB_INDEX_URI`` environment
            variable.
 
        Returns
        -------
        apdb : `apdb`
            Instance of `Apdb` class, the type of the returned instance is
            determined by configuration.
        """
        config = ApdbConfig.from_uri(uri)
        return make_apdb(config)
 

◆ get_replica()

ApdbSqlReplica lsst.dax.apdb.sql.apdbSql.ApdbSql.get_replica ( self )

Return `ApdbReplica` instance for this database.

Definition at line 456 of file apdbSql.py.

    def get_replica(self) -> ApdbSqlReplica:
        """Return `ApdbReplica` instance for this database."""
        return ApdbSqlReplica(self._schema, self._engine, self._db_schema_version)
 

◆ getConfig()

ApdbSqlConfig lsst.dax.apdb.sql.apdbSql.ApdbSql.getConfig ( self )

Return APDB configuration for this instance, including any updates
that may be read from database.

Returns
-------
config : `ApdbConfig`
    APDB configuration.

Reimplemented from lsst.dax.apdb.apdb.Apdb.

Definition at line 484 of file apdbSql.py.

    def getConfig(self) -> ApdbSqlConfig:
        # docstring is inherited from a base class
        return self.config
 

◆ getDiaForcedSources()

pandas.DataFrame \| None lsst.dax.apdb.sql.apdbSql.ApdbSql.getDiaForcedSources	(		self,
		Region	region,
		Iterable[int] \| None	object_ids,
		astropy.time.Time	visit_time,
		astropy.time.Time \| None	start_time = None )

Return catalog of DiaForcedSource instances from a given region.

Parameters
----------
region : `lsst.sphgeom.Region`
    Region to search for DIASources.
object_ids : iterable [ `int` ], optional
    List of DiaObject IDs to further constrain the set of returned
    sources. If list is empty then empty catalog is returned with a
    correct schema. If `None` then returned sources are not
    constrained.
visit_time : `astropy.time.Time`
    Time of the current visit. If APDB contains records later than this
    time they may also be returned.
start_time : `astropy.time.Time`, optional
    Lower bound of time window for the query. If not specified then
    it is calculated using ``visit_time`` and
    ``read_forced_sources_months`` configuration parameter.

Returns
-------
catalog : `pandas.DataFrame`, or `None`
    Catalog containing DiaForcedSource records. `None` is returned if
    ``start_time`` is not specified and ``read_forced_sources_months``
    configuration parameter is set to 0.

Raises
------
NotImplementedError
    May be raised by some implementations if ``object_ids`` is `None`.

Notes
-----
This method returns DiaForcedSource catalog for a region with
additional filtering based on DiaObject IDs. Only a subset of DiaSource
history is returned limited by ``read_forced_sources_months`` config
parameter, w.r.t. ``visit_time``. If ``object_ids`` is empty then an
empty catalog is always returned with the correct schema
(columns/types). If ``object_ids`` is `None` then no filtering is
performed and some of the returned records may be outside the specified
region.

Reimplemented from lsst.dax.apdb.apdb.Apdb.

Definition at line 594 of file apdbSql.py.

    ) -> pandas.DataFrame | None:
        # docstring is inherited from a base class
        if start_time is None and self.config.read_forced_sources_months == 0:
            _LOG.debug("Skip DiaForceSources fetching")
            return None
 
        if object_ids is None:
            # This implementation does not support region-based selection. In
            # the past DiaForcedSource schema did not have ra/dec columns (it
            # had x/y columns). ra/dec were added at some point, so we could
            # add pixelOd column to this table if/when needed.
            raise NotImplementedError("Region-based selection is not supported")
 
        # TODO: DateTime.MJD must be consistent with code in ap_association,
        # alternatively we can fill midpointMjdTai ourselves in store()
        if start_time is None:
            start_time_mjdTai = _make_midpointMjdTai_start(visit_time, self.config.read_forced_sources_months)
        else:
            start_time_mjdTai = float(start_time.tai.mjd)
        _LOG.debug("start_time_mjdTai = %.6f", start_time_mjdTai)
 
        with self._timer("select_time", tags={"table": "DiaForcedSource"}) as timer:
            sources = self._getSourcesByIDs(ApdbTables.DiaForcedSource, list(object_ids), start_time_mjdTai)
            timer.add_values(row_count=len(sources))
 
        _LOG.debug("found %s DiaForcedSources", len(sources))
        return sources
 

◆ getDiaObjects()

pandas.DataFrame lsst.dax.apdb.sql.apdbSql.ApdbSql.getDiaObjects	(		self,
		Region	region )

Return catalog of DiaObject instances from a given region.

This method returns only the last version of each DiaObject,
and may return only the subset of the DiaObject columns needed
for AP association. Some
records in a returned catalog may be outside the specified region, it
is up to a client to ignore those records or cleanup the catalog before
futher use.

Parameters
----------
region : `lsst.sphgeom.Region`
    Region to search for DIAObjects.

Returns
-------
catalog : `pandas.DataFrame`
    Catalog containing DiaObject records for a region that may be a
    superset of the specified region.

Reimplemented from lsst.dax.apdb.apdb.Apdb.

Definition at line 533 of file apdbSql.py.

    def getDiaObjects(self, region: Region) -> pandas.DataFrame:
        # docstring is inherited from a base class
 
        # decide what columns we need
        if self.config.dia_object_index == "last_object_table":
            table_enum = ApdbTables.DiaObjectLast
        else:
            table_enum = ApdbTables.DiaObject
        table = self._schema.get_table(table_enum)
        if not self.config.dia_object_columns:
            columns = self._schema.get_apdb_columns(table_enum)
        else:
            columns = [table.c[col] for col in self.config.dia_object_columns]
        query = sql.select(*columns)
 
        # build selection
        query = query.where(self._filterRegion(table, region))
 
        if self._schema.has_mjd_timestamps:
            validity_end_column = "validityEndMjdTai"
        else:
            validity_end_column = "validityEnd"
 
        # select latest version of objects
        if self.config.dia_object_index != "last_object_table":
            query = query.where(table.columns[validity_end_column] == None)  # noqa: E711
 
        # _LOG.debug("query: %s", query)
 
        # execute select
        with self._timer("select_time", tags={"table": "DiaObject"}) as timer:
            with self._engine.begin() as conn:
                objects = pandas.read_sql_query(query, conn)
            timer.add_values(row_count=len(objects))
        _LOG.debug("found %s DiaObjects", len(objects))
        return self._fix_result_timestamps(objects)
 

◆ getDiaSources()

pandas.DataFrame \| None lsst.dax.apdb.sql.apdbSql.ApdbSql.getDiaSources	(		self,
		Region	region,
		Iterable[int] \| None	object_ids,
		astropy.time.Time	visit_time,
		astropy.time.Time \| None	start_time = None )

Return catalog of DiaSource instances from a given region.

Parameters
----------
region : `lsst.sphgeom.Region`
    Region to search for DIASources.
object_ids : iterable [ `int` ], optional
    List of DiaObject IDs to further constrain the set of returned
    sources. If `None` then returned sources are not constrained. If
    list is empty then empty catalog is returned with a correct
    schema.
visit_time : `astropy.time.Time`
    Time of the current visit. If APDB contains records later than this
    time they may also be returned.
start_time : `astropy.time.Time`, optional
    Lower bound of time window for the query. If not specified then
    it is calculated using ``visit_time`` and
    ``read_forced_sources_months`` configuration parameter.

Returns
-------
catalog : `pandas.DataFrame`, or `None`
    Catalog containing DiaSource records. `None` is returned if
    ``start_time`` is not specified and ``read_sources_months``
    configuration parameter is set to 0.

Notes
-----
This method returns DiaSource catalog for a region with additional
filtering based on DiaObject IDs. Only a subset of DiaSource history
is returned limited by ``read_sources_months`` config parameter, w.r.t.
``visit_time``. If ``object_ids`` is empty then an empty catalog is
always returned with the correct schema (columns/types). If
``object_ids`` is `None` then no filtering is performed and some of the
returned records may be outside the specified region.

Reimplemented from lsst.dax.apdb.apdb.Apdb.

Definition at line 570 of file apdbSql.py.

    ) -> pandas.DataFrame | None:
        # docstring is inherited from a base class
        if start_time is None and self.config.read_sources_months == 0:
            _LOG.debug("Skip DiaSources fetching")
            return None
 
        if start_time is None:
            start_time_mjdTai = _make_midpointMjdTai_start(visit_time, self.config.read_sources_months)
        else:
            start_time_mjdTai = float(start_time.tai.mjd)
        _LOG.debug("start_time_mjdTai = %.6f", start_time_mjdTai)
 
        if object_ids is None:
            # region-based select
            return self._getDiaSourcesInRegion(region, start_time_mjdTai)
        else:
            return self._getDiaSourcesByIDs(list(object_ids), start_time_mjdTai)
 

◆ init_database()

ApdbSqlConfig lsst.dax.apdb.sql.apdbSql.ApdbSql.init_database	(		cls,
		str	db_url,
		*	,
		str \| None	schema_file = None,
		str \| None	ss_schema_file = None,
		int \| None	read_sources_months = None,
		int \| None	read_forced_sources_months = None,
		bool	enable_replica = False,
		int \| None	connection_timeout = None,
		str \| None	dia_object_index = None,
		int \| None	htm_level = None,
		str \| None	htm_index_column = None,
		tuple[str, str] \| None	ra_dec_columns = None,
		str \| None	prefix = None,
		str \| None	namespace = None,
		bool	drop = False )

Initialize new APDB instance and make configuration object for it.

Parameters
----------
db_url : `str`
    SQLAlchemy database URL.
schema_file : `str`, optional
    Location of (YAML) configuration file with APDB schema. If not
    specified then default location will be used.
ss_schema_file : `str`, optional
    Location of (YAML) configuration file with SSO schema. If not
    specified then default location will be used.
read_sources_months : `int`, optional
    Number of months of history to read from DiaSource.
read_forced_sources_months : `int`, optional
    Number of months of history to read from DiaForcedSource.
enable_replica : `bool`, optional
    If True, make additional tables used for replication to PPDB.
connection_timeout : `int`, optional
    Database connection timeout in seconds.
dia_object_index : `str`, optional
    Indexing mode for DiaObject table.
htm_level : `int`, optional
    HTM indexing level.
htm_index_column : `str`, optional
    Name of a HTM index column for DiaObject and DiaSource tables.
ra_dec_columns : `tuple` [`str`, `str`], optional
    Names of ra/dec columns in DiaObject table.
prefix : `str`, optional
    Optional prefix for all table names.
namespace : `str`, optional
    Name of the database schema for all APDB tables. If not specified
    then default schema is used.
drop : `bool`, optional
    If `True` then drop existing tables before re-creating the schema.

Returns
-------
config : `ApdbSqlConfig`
    Resulting configuration object for a created APDB instance.

Definition at line 365 of file apdbSql.py.

    ) -> ApdbSqlConfig:
        """Initialize new APDB instance and make configuration object for it.
 
        Parameters
        ----------
        db_url : `str`
            SQLAlchemy database URL.
        schema_file : `str`, optional
            Location of (YAML) configuration file with APDB schema. If not
            specified then default location will be used.
        ss_schema_file : `str`, optional
            Location of (YAML) configuration file with SSO schema. If not
            specified then default location will be used.
        read_sources_months : `int`, optional
            Number of months of history to read from DiaSource.
        read_forced_sources_months : `int`, optional
            Number of months of history to read from DiaForcedSource.
        enable_replica : `bool`, optional
            If True, make additional tables used for replication to PPDB.
        connection_timeout : `int`, optional
            Database connection timeout in seconds.
        dia_object_index : `str`, optional
            Indexing mode for DiaObject table.
        htm_level : `int`, optional
            HTM indexing level.
        htm_index_column : `str`, optional
            Name of a HTM index column for DiaObject and DiaSource tables.
        ra_dec_columns : `tuple` [`str`, `str`], optional
            Names of ra/dec columns in DiaObject table.
        prefix : `str`, optional
            Optional prefix for all table names.
        namespace : `str`, optional
            Name of the database schema for all APDB tables. If not specified
            then default schema is used.
        drop : `bool`, optional
            If `True` then drop existing tables before re-creating the schema.
 
        Returns
        -------
        config : `ApdbSqlConfig`
            Resulting configuration object for a created APDB instance.
        """
        config = ApdbSqlConfig(db_url=db_url, enable_replica=enable_replica)
        if schema_file is not None:
            config.schema_file = schema_file
        if ss_schema_file is not None:
            config.ss_schema_file = ss_schema_file
        if read_sources_months is not None:
            config.read_sources_months = read_sources_months
        if read_forced_sources_months is not None:
            config.read_forced_sources_months = read_forced_sources_months
        if connection_timeout is not None:
            config.connection_config.connection_timeout = connection_timeout
        if dia_object_index is not None:
            config.dia_object_index = dia_object_index
        if htm_level is not None:
            config.pixelization.htm_level = htm_level
        if htm_index_column is not None:
            config.pixelization.htm_index_column = htm_index_column
        if ra_dec_columns is not None:
            config.ra_dec_columns = ra_dec_columns
        if prefix is not None:
            config.prefix = prefix
        if namespace is not None:
            config.namespace = namespace
 
        cls._makeSchema(config, drop=drop)
 
        # SQLite has a nasty habit of creating empty database by default,
        # update URL in config file to disable that behavior.
        config.db_url = cls._update_sqlite_url(config.db_url)
 
        return config
 

◆ metadata()

ApdbMetadata lsst.dax.apdb.sql.apdbSql.ApdbSql.metadata ( self )

Object controlling access to APDB metadata (`ApdbMetadata`).

Reimplemented from lsst.dax.apdb.apdb.Apdb.

Definition at line 743 of file apdbSql.py.

    def metadata(self) -> ApdbMetadata:
        # docstring is inherited from a base class
        return self._metadata
 

◆ reassignDiaSources()

None lsst.dax.apdb.sql.apdbSql.ApdbSql.reassignDiaSources	(		self,
		Mapping[int, int]	idMap )

Associate DiaSources with SSObjects, dis-associating them
from DiaObjects.

Parameters
----------
idMap : `Mapping`
    Maps DiaSource IDs to their new SSObject IDs.

Raises
------
ValueError
    Raised if DiaSource ID does not exist in the database.

Reimplemented from lsst.dax.apdb.apdb.Apdb.

Definition at line 684 of file apdbSql.py.

    def reassignDiaSources(self, idMap: Mapping[int, int]) -> None:
        # docstring is inherited from a base class
 
        timestamp: float | datetime.datetime
        if self._schema.has_mjd_timestamps:
            timestamp_column = "ssObjectReassocTimeMjdTai"
            timestamp = float(astropy.time.Time.now().tai.mjd)
        else:
            timestamp_column = "ssObjectReassocTime"
            timestamp = datetime.datetime.now(tz=datetime.UTC)
 
        table = self._schema.get_table(ApdbTables.DiaSource)
        query = table.update().where(table.columns["diaSourceId"] == sql.bindparam("srcId"))
 
        with self._engine.begin() as conn:
            # Need to make sure that every ID exists in the database, but
            # executemany may not support rowcount, so iterate and check what
            # is missing.
            missing_ids: list[int] = []
            for key, value in idMap.items():
                params = {
                    "srcId": key,
                    "diaObjectId": 0,
                    "ssObjectId": value,
                    timestamp_column: timestamp,
                }
                result = conn.execute(query, params)
                if result.rowcount == 0:
                    missing_ids.append(key)
            if missing_ids:
                missing = ",".join(str(item) for item in missing_ids)
                raise ValueError(f"Following DiaSource IDs do not exist in the database: {missing}")
 

◆ store()

None lsst.dax.apdb.sql.apdbSql.ApdbSql.store	(		self,
		astropy.time.Time	visit_time,
		pandas.DataFrame	objects,
		pandas.DataFrame \| None	sources = None,
		pandas.DataFrame \| None	forced_sources = None )

Store all three types of catalogs in the database.

Parameters
----------
visit_time : `astropy.time.Time`
    Time of the visit.
objects : `pandas.DataFrame`
    Catalog with DiaObject records.
sources : `pandas.DataFrame`, optional
    Catalog with DiaSource records.
forced_sources : `pandas.DataFrame`, optional
    Catalog with DiaForcedSource records.

Notes
-----
This methods takes DataFrame catalogs, their schema must be
compatible with the schema of APDB table:

  - column names must correspond to database table columns
  - types and units of the columns must match database definitions,
    no unit conversion is performed presently
  - columns that have default values in database schema can be
    omitted from catalog
  - this method knows how to fill interval-related columns of DiaObject
    (validityStart, validityEnd) they do not need to appear in a
    catalog
  - source catalogs have ``diaObjectId`` column associating sources
    with objects

This operation need not be atomic, but DiaSources and DiaForcedSources
will not be stored until all DiaObjects are stored.

Reimplemented from lsst.dax.apdb.apdb.Apdb.

Definition at line 651 of file apdbSql.py.

    ) -> None:
        # docstring is inherited from a base class
        objects = self._fix_input_timestamps(objects)
        if sources is not None:
            sources = self._fix_input_timestamps(sources)
        if forced_sources is not None:
            forced_sources = self._fix_input_timestamps(forced_sources)
 
        # We want to run all inserts in one transaction.
        with self._engine.begin() as connection:
            replica_chunk: ReplicaChunk | None = None
            if self._schema.replication_enabled:
                replica_chunk = ReplicaChunk.make_replica_chunk(visit_time, self.config.replica_chunk_seconds)
                self._storeReplicaChunk(replica_chunk, connection)
 
            # fill pixelId column for DiaObjects
            objects = self._add_spatial_index(objects)
            self._storeDiaObjects(objects, visit_time, replica_chunk, connection)
 
            if sources is not None:
                # fill pixelId column for DiaSources
                sources = self._add_spatial_index(sources)
                self._storeDiaSources(sources, replica_chunk, connection)
 
            if forced_sources is not None:
                self._storeDiaForcedSources(forced_sources, replica_chunk, connection)
 

◆ tableDef()

Table \| None lsst.dax.apdb.sql.apdbSql.ApdbSql.tableDef	(		self,
		ApdbTables	table )

Return table schema definition for a given table.

Parameters
----------
table : `ApdbTables`
    One of the known APDB tables.

Returns
-------
tableSchema : `.schema_model.Table` or `None`
    Table schema description, `None` is returned if table is not
    defined by this implementation.

Reimplemented from lsst.dax.apdb.apdb.Apdb.

Definition at line 488 of file apdbSql.py.

    def tableDef(self, table: ApdbTables) -> Table | None:
        # docstring is inherited from a base class
        return self._schema.tableSchemas.get(table)
 

◆ tableRowCount()

dict[str, int] lsst.dax.apdb.sql.apdbSql.ApdbSql.tableRowCount ( self )

Return dictionary with the table names and row counts.

Used by ``ap_proto`` to keep track of the size of the database tables.
Depending on database technology this could be expensive operation.

Returns
-------
row_counts : `dict`
    Dict where key is a table name and value is a row count.

Definition at line 460 of file apdbSql.py.

    def tableRowCount(self) -> dict[str, int]:
        """Return dictionary with the table names and row counts.
 
        Used by ``ap_proto`` to keep track of the size of the database tables.
        Depending on database technology this could be expensive operation.
 
        Returns
        -------
        row_counts : `dict`
            Dict where key is a table name and value is a row count.
        """
        res = {}
        tables = [ApdbTables.DiaObject, ApdbTables.DiaSource, ApdbTables.DiaForcedSource]
        if self.config.dia_object_index == "last_object_table":
            tables.append(ApdbTables.DiaObjectLast)
        with self._engine.begin() as conn:
            for table in tables:
                sa_table = self._schema.get_table(table)
                stmt = sql.select(func.count()).select_from(sa_table)
                count: int = conn.execute(stmt).scalar_one()
                res[table.name] = count
 
        return res
 

Member Data Documentation

◆ _db_schema_version

lsst.dax.apdb.sql.apdbSql.ApdbSql._db_schema_version = self._versionCheck(self._metadata)

protected

Definition at line 180 of file apdbSql.py.

◆ _engine

lsst.dax.apdb.sql.apdbSql.ApdbSql._engine = self._makeEngine(config, create=False)

protected

Definition at line 153 of file apdbSql.py.

◆ _frozen_parameters

tuple lsst.dax.apdb.sql.apdbSql.ApdbSql._frozen_parameters

staticprotected

Initial value:

=  (
        "enable_replica",
        "dia_object_index",
        "pixelization.htm_level",
        "pixelization.htm_index_column",
        "ra_dec_columns",
    )

Definition at line 143 of file apdbSql.py.

◆ _metadata

lsst.dax.apdb.sql.apdbSql.ApdbSql._metadata = ApdbMetadataSql(self._engine, meta_table)

protected

Definition at line 158 of file apdbSql.py.

◆ _schema

lsst.dax.apdb.sql.apdbSql.ApdbSql._schema

protected

Initial value:

=  ApdbSqlSchema(
            engine=self._engine,
            dia_object_index=self.config.dia_object_index,
            schema_file=self.config.schema_file,
            ss_schema_file=self.config.ss_schema_file,
            prefix=self.config.prefix,
            namespace=self.config.namespace,
            htm_index_column=self.config.pixelization.htm_index_column,
            enable_replica=self.config.enable_replica,
        )

Definition at line 169 of file apdbSql.py.

◆ config

lsst.dax.apdb.sql.apdbSql.ApdbSql.config = freezer.update(config, config_json)

Definition at line 165 of file apdbSql.py.

◆ metadataCodeVersionKey

lsst.dax.apdb.sql.apdbSql.ApdbSql.metadataCodeVersionKey = "version:ApdbSql"

static

Definition at line 134 of file apdbSql.py.

◆ metadataConfigKey

lsst.dax.apdb.sql.apdbSql.ApdbSql.metadataConfigKey = "config:apdb-sql.json"

static

Definition at line 140 of file apdbSql.py.

◆ metadataReplicaVersionKey

lsst.dax.apdb.sql.apdbSql.ApdbSql.metadataReplicaVersionKey = "version:ApdbSqlReplica"

static

Definition at line 137 of file apdbSql.py.

◆ metadataSchemaVersionKey

lsst.dax.apdb.sql.apdbSql.ApdbSql.metadataSchemaVersionKey = "version:schema"

static

Definition at line 131 of file apdbSql.py.

◆ pixelator

lsst.dax.apdb.sql.apdbSql.ApdbSql.pixelator = HtmPixelization(self.config.pixelization.htm_level)

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

Public Member Functions

Public Attributes

Static Public Attributes

Protected Member Functions

Protected Attributes

Static Protected Attributes

Detailed Description

Constructor & Destructor Documentation

◆ __init__()

Member Function Documentation

◆ _add_spatial_index()

◆ _connection_url()

◆ _filterRegion()

◆ _fix_input_timestamps()

◆ _fix_result_timestamps()

◆ _getDiaSourcesByIDs()

◆ _getDiaSourcesInRegion()

◆ _getSourcesByIDs()

◆ _htm_indices()

◆ _makeEngine()

◆ _makeSchema()

◆ _storeDiaForcedSources()

◆ _storeDiaObjects()

◆ _storeDiaSources()

◆ _storeReplicaChunk()

◆ _storeUpdateRecords()

◆ _timer()

◆ _update_sqlite_url()

◆ _versionCheck()

◆ admin()

◆ apdbImplementationVersion()

◆ containsVisitDetector()

◆ countUnassociatedObjects()

◆ dailyJob()

◆ from_config()

◆ from_uri()

◆ get_replica()

◆ getConfig()

◆ getDiaForcedSources()

◆ getDiaObjects()

◆ getDiaSources()

◆ init_database()

◆ metadata()

◆ reassignDiaSources()

◆ store()

◆ tableDef()

◆ tableRowCount()

Member Data Documentation

◆ _db_schema_version

◆ _engine

◆ _frozen_parameters

◆ _metadata

◆ _schema

◆ config

◆ metadataCodeVersionKey

◆ metadataConfigKey

◆ metadataReplicaVersionKey

◆ metadataSchemaVersionKey

◆ pixelator

◆ init()