lsst.dax.apdb.cassandra.sessionFactory.SessionFactory Class Reference

Public Member Functions
	__init__ (self, ApdbCassandraConfig config)
None	__del__ (self)
Session	session (self)

Protected Member Functions
tuple[Cluster, Session]	_make_session (self)
AuthProvider|None	_make_auth_provider (self)
Mapping[Any, ExecutionProfile]	_make_profiles (self)

Protected Attributes
	_config = config
Cluster|None	_cluster = None
Session|None	_session = None

Detailed Description

Implementation of SessionFactory that uses parameters from Apdb
configuration.

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

Definition at line 78 of file sessionFactory.py.

Constructor & Destructor Documentation

__init__()

lsst.dax.apdb.cassandra.sessionFactory.SessionFactory.__init__	(		self,
		ApdbCassandraConfig	config )

Definition at line 88 of file sessionFactory.py.

    def __init__(self, config: ApdbCassandraConfig):
        self._config = config
        self._cluster: Cluster | None = None
        self._session: Session | None = None
 

__del__()

None lsst.dax.apdb.cassandra.sessionFactory.SessionFactory.__del__

(

self

)

Definition at line 93 of file sessionFactory.py.

    def __del__(self) -> None:
        # Need to call Cluster.shutdown() to avoid warnings.
        if hasattr(self, "_cluster"):
            if self._cluster:
                self._cluster.shutdown()
 

Member Function Documentation

_make_auth_provider()

AuthProvider | None lsst.dax.apdb.cassandra.sessionFactory.SessionFactory._make_auth_provider ( self )

protected

Make Cassandra authentication provider instance.

Definition at line 148 of file sessionFactory.py.

    def _make_auth_provider(self) -> AuthProvider | None:
        """Make Cassandra authentication provider instance."""
        try:
            dbauth = DbAuth()
        except DbAuthNotFoundError:
            # Credentials file doesn't exist, use anonymous login.
            return None
 
        empty_username = True
        # Try every contact point in turn.
        for hostname in self._config.contact_points:
            try:
                username, password = dbauth.getAuth(
                    "cassandra",
                    self._config.connection_config.username,
                    hostname,
                    self._config.connection_config.port,
                    self._config.keyspace,
                )
                if not username:
                    # Password without user name, try next hostname, but give
                    # warning later if no better match is found.
                    empty_username = True
                else:
                    return PlainTextAuthProvider(username=username, password=password)
            except DbAuthNotFoundError:
                pass
 
        if empty_username:
            _LOG.warning(
                f"Credentials file ({dbauth.db_auth_path}) provided password but not "
                "user name, anonymous Cassandra logon will be attempted."
            )
 
        return None
 

_make_profiles()

Mapping[Any, ExecutionProfile] lsst.dax.apdb.cassandra.sessionFactory.SessionFactory._make_profiles ( self )

protected

Make all execution profiles used in the code.

Definition at line 184 of file sessionFactory.py.

    def _make_profiles(self) -> Mapping[Any, ExecutionProfile]:
        """Make all execution profiles used in the code."""
        config = self._config
        if config.connection_config.private_ips:
            loadBalancePolicy = WhiteListRoundRobinPolicy(hosts=config.contact_points)
        else:
            loadBalancePolicy = RoundRobinPolicy()
 
        read_tuples_profile = ExecutionProfile(
            consistency_level=getattr(cassandra.ConsistencyLevel, config.connection_config.read_consistency),
            request_timeout=config.connection_config.read_timeout,
            row_factory=cassandra.query.tuple_factory,
            load_balancing_policy=loadBalancePolicy,
        )
        read_pandas_profile = ExecutionProfile(
            consistency_level=getattr(cassandra.ConsistencyLevel, config.connection_config.read_consistency),
            request_timeout=config.connection_config.read_timeout,
            row_factory=pandas_dataframe_factory,
            load_balancing_policy=loadBalancePolicy,
        )
        read_raw_profile = ExecutionProfile(
            consistency_level=getattr(cassandra.ConsistencyLevel, config.connection_config.read_consistency),
            request_timeout=config.connection_config.read_timeout,
            row_factory=raw_data_factory,
            load_balancing_policy=loadBalancePolicy,
        )
        # Profile to use with select_concurrent to return pandas data frame
        read_pandas_multi_profile = ExecutionProfile(
            consistency_level=getattr(cassandra.ConsistencyLevel, config.connection_config.read_consistency),
            request_timeout=config.connection_config.read_timeout,
            row_factory=pandas_dataframe_factory,
            load_balancing_policy=loadBalancePolicy,
        )
        # Profile to use with select_concurrent to return raw data (columns and
        # rows)
        read_raw_multi_profile = ExecutionProfile(
            consistency_level=getattr(cassandra.ConsistencyLevel, config.connection_config.read_consistency),
            request_timeout=config.connection_config.read_timeout,
            row_factory=raw_data_factory,
            load_balancing_policy=loadBalancePolicy,
        )
        write_profile = ExecutionProfile(
            consistency_level=getattr(cassandra.ConsistencyLevel, config.connection_config.write_consistency),
            request_timeout=config.connection_config.write_timeout,
            load_balancing_policy=loadBalancePolicy,
        )
        # To replace default DCAwareRoundRobinPolicy
        default_profile = ExecutionProfile(
            load_balancing_policy=loadBalancePolicy,
        )
        return {
            "read_tuples": read_tuples_profile,
            "read_pandas": read_pandas_profile,
            "read_raw": read_raw_profile,
            "read_pandas_multi": read_pandas_multi_profile,
            "read_raw_multi": read_raw_multi_profile,
            "write": write_profile,
            EXEC_PROFILE_DEFAULT: default_profile,
        }
 
 

_make_session()

tuple[Cluster, Session] lsst.dax.apdb.cassandra.sessionFactory.SessionFactory._make_session ( self )

protected

Make Cassandra session.

Returns
-------
cluster : `cassandra.cluster.Cluster`
    Cassandra Cluster object
session : `cassandra.cluster.Session`
    Cassandra session object

Definition at line 111 of file sessionFactory.py.

    def _make_session(self) -> tuple[Cluster, Session]:
        """Make Cassandra session.
 
        Returns
        -------
        cluster : `cassandra.cluster.Cluster`
            Cassandra Cluster object
        session : `cassandra.cluster.Session`
            Cassandra session object
        """
        addressTranslator: AddressTranslator | None = None
        if self._config.connection_config.private_ips:
            addressTranslator = _AddressTranslator(
                self._config.contact_points, self._config.connection_config.private_ips
            )
 
        with Timer("cluster_connect", _MON):
            cluster = Cluster(
                execution_profiles=self._make_profiles(),
                contact_points=self._config.contact_points,
                port=self._config.connection_config.port,
                address_translator=addressTranslator,
                protocol_version=self._config.connection_config.protocol_version,
                auth_provider=self._make_auth_provider(),
                **self._config.connection_config.extra_parameters,
            )
            session = cluster.connect()
 
        # Dump queries if debug level is enabled.
        if _LOG.isEnabledFor(logging.DEBUG):
            session.add_request_init_listener(_dump_query)
 
        # Disable result paging
        session.default_fetch_size = None
 
        return cluster, session
 

session()

Session lsst.dax.apdb.cassandra.sessionFactory.SessionFactory.session

(

self

)

Return Cassandra Session, making new connection if necessary.

Returns
-------
session : `cassandra.cluster.Sesion`
    Cassandra session object.

Definition at line 99 of file sessionFactory.py.

    def session(self) -> Session:
        """Return Cassandra Session, making new connection if necessary.
 
        Returns
        -------
        session : `cassandra.cluster.Sesion`
            Cassandra session object.
        """
        if self._session is None:
            self._cluster, self._session = self._make_session()
        return self._session
 

Member Data Documentation

_cluster

lsst.dax.apdb.cassandra.sessionFactory.SessionFactory._cluster = None

protected

Definition at line 90 of file sessionFactory.py.

_config

lsst.dax.apdb.cassandra.sessionFactory.SessionFactory._config = config

protected

Definition at line 89 of file sessionFactory.py.

_session

Session | None lsst.dax.apdb.cassandra.sessionFactory.SessionFactory._session = None

protected

Definition at line 91 of file sessionFactory.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/sessionFactory.py