DB Interface¶

Connection Management¶

lightcurvedb.core.connection.db_from_config(_filepath: str | pathlib.Path | None = None, _section: str | None = None, **overrides: Any) → T¶

Factory for creating configured function instances.

This class maintains the schema context when defining parameters and options. Using the maintained definition we are able to call the wrapped function, provide overrides, or emit template configurations.

lightcurvedb.core.connection.builder¶

The configuration schema for the wrapped function.

Type:: ConfigurationBuilder

lightcurvedb.core.connection.section¶

Default configuration file section to use.

Type:: str

lightcurvedb.core.connection.configuration_order¶

Defines the precedence order for configuration sources.

Type:: ResolutionDefinition

Examples

>>> # Created by @configurable decorator
>>> @configurable("Settings")
... @param("value", type=int)
... def process(value):
...     return value * 2
>>> # process is now a ConfigurationFactory
>>> result = process("config.ini")

lightcurvedb.core.connection.configure_engine(_filepath: str | pathlib.Path | None = None, _section: str | None = None, **overrides: Any) → T

Factory for creating configured function instances.

lightcurvedb.core.connection.builder

The configuration schema for the wrapped function.

Type:: ConfigurationBuilder

lightcurvedb.core.connection.section

Default configuration file section to use.

Type:: str

lightcurvedb.core.connection.configuration_order

Defines the precedence order for configuration sources.

Type:: ResolutionDefinition

Examples

>>> # Created by @configurable decorator
>>> @configurable("Settings")
... @param("value", type=int)
... def process(value):
...     return value * 2
>>> # process is now a ConfigurationFactory
>>> result = process("config.ini")

lightcurvedb.core.connection.LCDB_Session = sessionmaker(class_='Session', bind=None, autoflush=True, expire_on_commit=False)¶

A configurable Session factory.

The sessionmaker factory generates new Session objects when called, creating them given the configurational arguments established here.

e.g.:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# an Engine, which the Session will use for connection
# resources
engine = create_engine("postgresql+psycopg2://scott:tiger@localhost/")

Session = sessionmaker(engine)

with Session() as session:
    session.add(some_object)
    session.add(some_other_object)
    session.commit()

Context manager use is optional; otherwise, the returned _orm.Session object may be closed explicitly via the _orm.Session.close() method. Using a try:/finally: block is optional, however will ensure that the close takes place even if there are database errors:

session = Session()
try:
    session.add(some_object)
    session.add(some_other_object)
    session.commit()
finally:
    session.close()

sessionmaker acts as a factory for _orm.Session objects in the same way as an _engine.Engine acts as a factory for _engine.Connection objects. In this way it also includes a _orm.sessionmaker.begin() method, that provides a context manager which both begins and commits a transaction, as well as closes out the _orm.Session when complete, rolling back the transaction if any errors occur:

Session = sessionmaker(engine)

with Session.begin() as session:
    session.add(some_object)
    session.add(some_other_object)
# commits transaction, closes session

Added in version 1.4.

When calling upon _orm.sessionmaker to construct a _orm.Session, keyword arguments may also be passed to the method; these arguments will override that of the globally configured parameters. Below we use a _orm.sessionmaker bound to a certain _engine.Engine to produce a _orm.Session that is instead bound to a specific _engine.Connection procured from that engine:

Session = sessionmaker(engine)

# bind an individual session to a connection

with engine.connect() as connection:
    with Session(bind=connection) as session:
        ...  # work with session

The class also includes a method _orm.sessionmaker.configure(), which can be used to specify additional keyword arguments to the factory, which will take effect for subsequent Session objects generated. This is usually used to associate one or more _engine.Engine objects with an existing sessionmaker factory before it is first used:

# application starts, sessionmaker does not have
# an engine bound yet
Session = sessionmaker()

# ... later, when an engine URL is read from a configuration
# file or other events allow the engine to be created
engine = create_engine("sqlite:///foo.db")
Session.configure(bind=engine)

sess = Session()
# work with session

Session Decorators¶

lightcurvedb.io.pipeline.db_scope(session_factory: sessionmaker | None = None, application_name: str | None = None, **session_kwargs: Any) → Callable[[F], F]¶

Decorator that provides automatic database session management.

Wraps a function to automatically handle database session lifecycle. The decorated function receives an open database session as its first argument. The session is automatically closed when the function returns, with automatic rollback of uncommitted changes.

Parameters:

session_factory (sqlalchemy.orm.sessionmaker, optional) – A SQLAlchemy sessionmaker instance for creating database sessions. If not provided, defaults to the global LCDB_Session.
application_name (str, optional) – Name used for logging purposes to identify the calling function. If not provided, uses the wrapped function’s name.
**session_kwargs (dict) –
Additional keyword arguments passed to the session factory when creating new sessions. Common uses include:
- bind: Override the database engine
- info: Attach metadata to the session
- expire_on_commit: Control object expiration behavior

Returns:

A decorator function that wraps the target function with automatic session management.

Return type:

Callable

Notes

The session is created using the provided factory with any kwargs
The session is automatically closed after function execution
Any uncommitted changes are automatically rolled back
The session is properly closed even if exceptions occur
The wrapped function must accept a session as its first argument

Examples

Basic usage with default session factory:

>>> from lightcurvedb.io.pipeline import db_scope
>>> from lightcurvedb.models import Mission
>>>
>>> @db_scope()
... def count_missions(session):
...     return session.query(Mission).count()
>>>
>>> mission_count = count_missions()

Using a custom session factory:

>>> from sqlalchemy.orm import sessionmaker
>>> custom_factory = sessionmaker(bind=my_engine)
>>>
>>> @db_scope(session_factory=custom_factory)
... def custom_query(session):
...     return session.execute("SELECT 1").scalar()

Passing session configuration:

>>> @db_scope(info={"task": "data_export"})
... def export_data(session, table_name):
...     # session.info will contain {"task": "data_export"}
...     return session.query(table_name).all()