lsst.dax.apdb.cassandra.connectionContext.ConnectionContext Class Reference

Public Member Functions
	__init__ (self, Session session, ApdbCassandraConfig config, Mapping[ApdbTables, Table] table_schemas)
ApdbCassandraTimePartitionRange|None	time_partitions_range (self)

Public Attributes
	session = session
	metadata
	config = freezer.update(config, config_json)
DbVersions	db_versions = self._readVersions(self.metadata)
bool	has_chunk_sub_partitions = False
bool	has_update_record_chunks_table = False
DbVersions	has_time_partition_meta = VersionTuple(0, 1, 3)
DbVersions	has_visit_detector_table = VersionTuple(0, 1, 2)
DbVersions	has_dia_object_last_to_partition = VersionTuple(0, 1, 1)
	preparer = PreparedStatementCache(self.session)
	partitioner = Partitioner(self.config)
	schema

Static Public Attributes
str	metadataSchemaVersionKey = "version:schema"
str	metadataCodeVersionKey = "version:ApdbCassandra"
str	metadataReplicaVersionKey = "version:ApdbCassandraReplica"
str	metadataConfigKey = "config:apdb-cassandra.json"
tuple	frozen_parameters

Protected Member Functions
DbVersions	_readVersions (self, ApdbMetadataCassandra metadata)

Detailed Description

Container for all kinds ob objects that are instantiated once the
connection to Cassandra is established.

Parameters
----------
session : `cassandra.cluster.Sesion`
    Cassandra session.
config : `ApdbCassandraConfig`
    Configuration object.
table_schemas : `~collection.abc.Mapping` [`ApdbTables`, `Table`]
    Schema definitions for regular APDB tables.

Definition at line 70 of file connectionContext.py.

Constructor & Destructor Documentation

__init__()

lsst.dax.apdb.cassandra.connectionContext.ConnectionContext.__init__	(		self,
		Session	session,
		ApdbCassandraConfig	config,
		Mapping[ApdbTables, Table]	table_schemas )

Definition at line 108 of file connectionContext.py.

    ):
        self.session = session
 
        meta_table_name = ApdbTables.metadata.table_name(config.prefix)
        self.metadata = ApdbMetadataCassandra(
            self.session, meta_table_name, config.keyspace, "read_tuples", "write"
        )
 
        # Read frozen config from metadata.
        config_json = self.metadata.get(self.metadataConfigKey)
        if config_json is not None:
            # Update config from metadata.
            freezer = ApdbConfigFreezer[ApdbCassandraConfig](self.frozen_parameters)
            self.config = freezer.update(config, config_json)
        else:
            self.config = config
        del config
 
        # Read versions stored in database.
        self.db_versions = self._readVersions(self.metadata)
        _LOG.debug("Database versions: %s", self.db_versions)
 
        # Since replica version 1.1.0 we use finer partitioning for replica
        # chunk tables.
        self.has_chunk_sub_partitions = False
        self.has_update_record_chunks_table = False
        if self.config.enable_replica:
            assert self.db_versions.replica_version is not None, "Replica version must be defined"
            self.has_chunk_sub_partitions = ApdbCassandraReplica.hasChunkSubPartitions(
                self.db_versions.replica_version
            )
            self.has_update_record_chunks_table = ApdbCassandraReplica.hasUpdateRecordChunks(
                self.db_versions.replica_version
            )
 
        # Since version 0.1.3 we have metadata for time partitions.
        self.has_time_partition_meta = self.db_versions.code_version >= VersionTuple(0, 1, 3)
 
        # Since version 0.1.2 we have an extra table for visit/detector.
        self.has_visit_detector_table = self.db_versions.code_version >= VersionTuple(0, 1, 2)
 
        # Support for DiaObjectLastToPartition was added at code version 0.1.1
        # in a backward-compatible way (we only use the table if it is there).
        self.has_dia_object_last_to_partition = self.db_versions.code_version >= VersionTuple(0, 1, 1)
 
        # Cache for prepared statements
        self.preparer = PreparedStatementCache(self.session)
 
        self.partitioner = Partitioner(self.config)
 
        self.schema = ApdbCassandraSchema(
            session=self.session,
            keyspace=self.config.keyspace,
            table_schemas=table_schemas,
            prefix=self.config.prefix,
            time_partition_tables=self.config.partitioning.time_partition_tables,
            enable_replica=self.config.enable_replica,
            replica_skips_diaobjects=self.config.replica_skips_diaobjects,
            has_chunk_sub_partitions=self.has_chunk_sub_partitions,
            has_visit_detector_table=self.has_visit_detector_table,
        )
 

Member Function Documentation

_readVersions()

DbVersions lsst.dax.apdb.cassandra.connectionContext.ConnectionContext._readVersions ( self, ApdbMetadataCassandra metadata )

protected

Read versions of all objects from metadata.

Definition at line 193 of file connectionContext.py.

    def _readVersions(self, metadata: ApdbMetadataCassandra) -> DbVersions:
        """Read versions of all objects from metadata."""
 
        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)
 
        # Check replica code version only if replica is enabled.
        db_replica_version: VersionTuple | None = None
        if self.config.enable_replica:
            db_replica_version = _get_version(self.metadataReplicaVersionKey)
 
        return DbVersions(
            schema_version=db_schema_version, code_version=db_code_version, replica_version=db_replica_version
        )

time_partitions_range()

ApdbCassandraTimePartitionRange | None lsst.dax.apdb.cassandra.connectionContext.ConnectionContext.time_partitions_range

(

self

)

Time partition range or None if instance does not use
time-partitioned tables (`ApdbCassandraTimePartitionRange` or `None`).

Definition at line 173 of file connectionContext.py.

    def time_partitions_range(self) -> ApdbCassandraTimePartitionRange | None:
        """Time partition range or None if instance does not use
        time-partitioned tables (`ApdbCassandraTimePartitionRange` or `None`).
        """
        if not self.config.partitioning.time_partition_tables:
            return None
 
        if self.has_time_partition_meta:
            return ApdbCassandraTimePartitionRange.from_meta(self.metadata)
        else:
            # Scan DiaSource tables and see which partitions are present.
            partitions = set()
            tables = self.schema.existing_tables(ApdbTables.DiaSource)
            for table_name in tables[ApdbTables.DiaSource]:
                _, _, part_str = table_name.rpartition("_")
                partitions.add(int(part_str))
            if not partitions:
                raise LookupError("Failed to find any partitioned DiaSource table.")
            return ApdbCassandraTimePartitionRange(start=min(partitions), end=max(partitions))
 

Member Data Documentation

config

lsst.dax.apdb.cassandra.connectionContext.ConnectionContext.config = freezer.update(config, config_json)

Definition at line 123 of file connectionContext.py.

db_versions

lsst.dax.apdb.cassandra.connectionContext.ConnectionContext.db_versions = self._readVersions(self.metadata)

Definition at line 129 of file connectionContext.py.

frozen_parameters

tuple lsst.dax.apdb.cassandra.connectionContext.ConnectionContext.frozen_parameters

static

Initial value:

=  (
        "enable_replica",
        "ra_dec_columns",
        "replica_skips_diaobjects",
        "replica_sub_chunk_count",
        "partitioning.part_pixelization",
        "partitioning.part_pix_level",
        "partitioning.time_partition_tables",
        "partitioning.time_partition_days",
    )

Definition at line 96 of file connectionContext.py.

has_chunk_sub_partitions

bool lsst.dax.apdb.cassandra.connectionContext.ConnectionContext.has_chunk_sub_partitions = False

Definition at line 134 of file connectionContext.py.

has_dia_object_last_to_partition

DbVersions lsst.dax.apdb.cassandra.connectionContext.ConnectionContext.has_dia_object_last_to_partition = VersionTuple(0, 1, 1)

Definition at line 153 of file connectionContext.py.

has_time_partition_meta

DbVersions lsst.dax.apdb.cassandra.connectionContext.ConnectionContext.has_time_partition_meta = VersionTuple(0, 1, 3)

Definition at line 146 of file connectionContext.py.

has_update_record_chunks_table

bool lsst.dax.apdb.cassandra.connectionContext.ConnectionContext.has_update_record_chunks_table = False

Definition at line 135 of file connectionContext.py.

has_visit_detector_table

DbVersions lsst.dax.apdb.cassandra.connectionContext.ConnectionContext.has_visit_detector_table = VersionTuple(0, 1, 2)

Definition at line 149 of file connectionContext.py.

metadata

lsst.dax.apdb.cassandra.connectionContext.ConnectionContext.metadata

Initial value:

=  ApdbMetadataCassandra(
            self.session, meta_table_name, config.keyspace, "read_tuples", "write"
        )

Definition at line 114 of file connectionContext.py.

metadataCodeVersionKey

str lsst.dax.apdb.cassandra.connectionContext.ConnectionContext.metadataCodeVersionKey = "version:ApdbCassandra"

static

Definition at line 87 of file connectionContext.py.

metadataConfigKey

str lsst.dax.apdb.cassandra.connectionContext.ConnectionContext.metadataConfigKey = "config:apdb-cassandra.json"

static

Definition at line 93 of file connectionContext.py.

metadataReplicaVersionKey

str lsst.dax.apdb.cassandra.connectionContext.ConnectionContext.metadataReplicaVersionKey = "version:ApdbCassandraReplica"

static

Definition at line 90 of file connectionContext.py.

metadataSchemaVersionKey

str lsst.dax.apdb.cassandra.connectionContext.ConnectionContext.metadataSchemaVersionKey = "version:schema"

static

Definition at line 84 of file connectionContext.py.

partitioner

lsst.dax.apdb.cassandra.connectionContext.ConnectionContext.partitioner = Partitioner(self.config)

Definition at line 158 of file connectionContext.py.

preparer

lsst.dax.apdb.cassandra.connectionContext.ConnectionContext.preparer = PreparedStatementCache(self.session)

Definition at line 156 of file connectionContext.py.

schema

lsst.dax.apdb.cassandra.connectionContext.ConnectionContext.schema

Initial value:

=  ApdbCassandraSchema(
            session=self.session,
            keyspace=self.config.keyspace,
            table_schemas=table_schemas,
            prefix=self.config.prefix,
            time_partition_tables=self.config.partitioning.time_partition_tables,
            enable_replica=self.config.enable_replica,
            replica_skips_diaobjects=self.config.replica_skips_diaobjects,
            has_chunk_sub_partitions=self.has_chunk_sub_partitions,
            has_visit_detector_table=self.has_visit_detector_table,
        )

Definition at line 160 of file connectionContext.py.

session

lsst.dax.apdb.cassandra.connectionContext.ConnectionContext.session = session

Definition at line 111 of file connectionContext.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/cassandra/connectionContext.py