Core Database Functions#

✓

Documentation Verified Last checked: 2025-11-25 Reviewer: Christof Buchbender

This page documents the core database connection and initialization API. These are the entry points for working with ops-db programmatically.

Configuration#

Database connection parameters are managed via Dynaconf settings. The configuration file is located at ccat_ops_db/config/settings.toml.

Default settings:

[default]
database_type = "sqlite"
database_sqlite_database = "/tmp/dev_ccat_ops_db.sqlite"

database_postgresql_database = "ccat_ops_db"
database_postgresql_user = "ccat"
database_postgresql_password = "pw4ccat"
database_postgresql_host = "localhost"
database_postgresql_port = "5432"

Environment-specific settings (e.g., [production], [development]) override defaults.

Environment variables with prefix CCAT_OPS_DB_ can override any setting:

export CCAT_OPS_DB_DATABASE_TYPE=postgresql
export CCAT_OPS_DB_DATABASE_POSTGRESQL_HOST=db.prod.example.com

The active environment is set via ENV_FOR_DYNACONF environment variable:

export ENV_FOR_DYNACONF=production

Database Initialization#

This function is a helper function for development and staging purposes. It fills the database with the starting seed that we used for production it will generate a representative starting state with defined obs_units, sources, instruments, and the like. A seeded database can be immediately used to simulate observing and data transfer as well as data reduction and archive transfer in our development and staging environments.

Initialize a database connection based on the supplied configuration.

Parameters:#

database_type: str, optional: Can be ‘sqlite’, ‘mysql’, or ‘postgresql’. Defaults to config setting.
database: str, optional: URL string to connect to the database. Defaults to config setting.
drop: bool, default False: If True, drops all tables before creating them.
null_pool: bool, default False: If True, uses NullPool instead of the default connection pool.
host: str, optional: Database host override.
port: str, optional: Database port override.
user: str, optional: Database user override.
password: str, optional: Database password override.
db_name: str, optional: Database name override.

Returns:#

Tuple[scoped_session, Engine]: A tuple containing the database session and engine.

Usage Example:

from ccat_ops_db import init_ccat_ops_db

# Initialize with default settings (from environment)
session, engine = init_ccat_ops_db()

# Initialize with specific database
session, engine = init_ccat_ops_db(
    database_type="postgresql",
    host="db.example.com",
    port="5432",
    user="myuser",
    password="mypass",
    db_name="ops_db"
)

# Initialize and drop existing tables (for testing)
session, engine = init_ccat_ops_db(drop=True)

# Use NullPool for short-lived connections
session, engine = init_ccat_ops_db(null_pool=True)

Database URL Construction#

Generate database URL based on type and configuration.

Parameters:

database_type – Type of database (‘sqlite’, ‘mysql’, ‘postgresql’)
database – Optional database URL override
host – Optional host override
port – Optional port override
user – Optional user override
password – Optional password override
db_name – Optional database name override
async_driver – Whether to use async driver (e.g., asyncpg for PostgreSQL)

Usage Example:

from ccat_ops_db import get_database_url

# PostgreSQL URL
url = get_database_url(
    database_type="postgresql",
    host="localhost",
    port="5432",
    user="ccat",
    password="secret",
    db_name="ccat_ops_db"
)
# Returns: postgresql://ccat:secret@localhost:5432/ccat_ops_db

# Async PostgreSQL URL
url = get_database_url(
    database_type="postgresql",
    host="localhost",
    db_name="ccat_ops_db",
    async_driver=True
)
# Returns: postgresql+asyncpg://...

# SQLite URL
url = get_database_url(
    database_type="sqlite",
    database="/tmp/test.db"
)
# Returns: sqlite:////tmp/test.db

Connection Pooling#

By default, SQLAlchemy uses connection pooling. For short-lived scripts or testing, you may want to disable pooling:

session, engine = init_ccat_ops_db(null_pool=True)

This uses NullPool, which creates a new connection for each operation and closes it immediately.

For long-running applications (like the API server), use the default pooling behavior.

Working with the Session#

The returned session is a SQLAlchemy scoped_session:

from ccat_ops_db import init_ccat_ops_db
from ccat_ops_db.models import Observatory

session, engine = init_ccat_ops_db()

# Query
obs = session.query(Observatory).filter_by(name="CCAT").first()

# Add
new_obs = Observatory(name="Test", description="Test observatory")
session.add(new_obs)
session.commit()

# Close when done
session.close()

For context manager usage:

from contextlib import contextmanager

@contextmanager
def get_db_session():
    session, engine = init_ccat_ops_db()
    try:
        yield session
        session.commit()
    except:
        session.rollback()
        raise
    finally:
        session.close()

with get_db_session() as session:
    # Your database operations
    pass