Moltres Package

Entry points and configuration.

Connection

moltres.connect(dsn: str | None = None, engine: Engine | None = None, session: Session | None = None, **options: EngineOptionValue) → Database[source]

Connect to a SQL database and return a Database handle.

Configuration can be provided via arguments or environment variables: - MOLTRES_DSN: Database connection string (if dsn is None) - MOLTRES_ECHO: Enable SQLAlchemy echo mode (true/false) - MOLTRES_FETCH_FORMAT: “records”, “pandas”, or “polars” - MOLTRES_DIALECT: Override SQL dialect detection - MOLTRES_POOL_SIZE: Connection pool size - MOLTRES_MAX_OVERFLOW: Maximum pool overflow connections - MOLTRES_POOL_TIMEOUT: Pool timeout in seconds - MOLTRES_POOL_RECYCLE: Connection recycle time in seconds - MOLTRES_POOL_PRE_PING: Enable connection health checks (true/false)

Parameters:

  • dsn – Database connection string. Examples: - SQLite: “sqlite:///path/to/database.db” - PostgreSQL: “postgresql://user:pass@host:port/dbname” - MySQL: “mysql://user:pass@host:port/dbname” If None, will use MOLTRES_DSN environment variable. Cannot be provided if engine or session is provided.

  • engine – SQLAlchemy Engine instance to use. If provided, dsn is ignored. This gives users more flexibility to configure the engine themselves. Pool configuration options (pool_size, max_overflow, etc.) are ignored when using an existing engine. Cannot be provided if session is provided.

  • session – SQLAlchemy Session or SQLModel Session instance to use. If provided, dsn and engine are ignored. The session’s bind (engine) will be used. This allows using Moltres with FastAPI’s dependency-injected sessions. Cannot be provided if dsn or engine is provided.

  • **options – Optional configuration parameters (can also be set via environment variables): - echo: Enable SQLAlchemy echo mode for debugging (default: False) - fetch_format: Result format - “records”, “pandas”, or “polars” (default: “records”) - dialect: Override SQL dialect detection (e.g., “postgresql”, “mysql”) - pool_size: Connection pool size (default: None, uses SQLAlchemy default). Ignored if engine is provided. - max_overflow: Maximum pool overflow connections (default: None). Ignored if engine is provided. - pool_timeout: Pool timeout in seconds (default: None). Ignored if engine is provided. - pool_recycle: Connection recycle time in seconds (default: None). Ignored if engine is provided. - pool_pre_ping: Enable connection health checks (default: False). Ignored if engine is provided. - future: Use SQLAlchemy 2.0 style (default: True)

Returns:

Database instance for querying and table operations

Return type:

Database

Raises:

  • ValueError – If neither dsn, engine, nor session is provided and MOLTRES_DSN is not set

  • ValueError – If multiple of dsn, engine, and session are provided

  • TypeError – If session is not a SQLAlchemy Session or SQLModel Session instance

Example

>>> # Using connection string with context manager (recommended)
>>> with connect("sqlite:///:memory:") as db:
...     from moltres.table.schema import column
...     _ = db.create_table("users", [column("id", "INTEGER"), column("active", "BOOLEAN")]).collect()
...     from moltres.io.records import Records
...     _ = Records.from_list([{"id": 1, "active": True}], database=db).insert_into("users")
...     df = db.table("users").select().where(col("active") == True)
...     results = df.collect()
...     # db.close() called automatically on exit

>>> # Using connection string (manual close)
>>> db = connect("sqlite:///:memory:")
>>> from moltres.table.schema import column
>>> _ = db.create_table("users", [column("id", "INTEGER"), column("active", "BOOLEAN")]).collect()
>>> from moltres.io.records import Records
>>> _ = Records.from_list([{"id": 1, "active": True}], database=db).insert_into("users")
>>> df = db.table("users").select().where(col("active") == True)
>>> results = df.collect()
>>> db.close()

>>> # Using SQLAlchemy Engine
>>> from sqlalchemy import create_engine
>>> engine = create_engine("sqlite:///:memory:")
>>> db2 = connect(engine=engine)
>>> _ = db2.create_table("test", [column("x", "INTEGER")]).collect()
>>> db2.close()

>>> # Using SQLAlchemy Session (e.g., from FastAPI dependency injection)
>>> from sqlalchemy.orm import Session, sessionmaker
>>> SessionLocal = sessionmaker(bind=create_engine("sqlite:///:memory:"))
>>> with SessionLocal() as session:
...     db3 = connect(session=session)
...     _ = db3.create_table("test2", [column("x", "INTEGER")]).collect()
moltres.async_connect(dsn: str | None = None, engine: AsyncEngine | None = None, session: AsyncSession | None = None, **options: EngineOptionValue) → AsyncDatabase[source]

Connect to a SQL database asynchronously and return an AsyncDatabase handle.

This function requires async dependencies. Install with: - pip install moltres[async] - for core async support (aiofiles) - pip install moltres[async-postgresql] - for PostgreSQL async support (includes async + asyncpg) - pip install moltres[async-mysql] - for MySQL async support (includes async + aiomysql) - pip install moltres[async-sqlite] - for SQLite async support (includes async + aiosqlite)

Configuration can be provided via arguments or environment variables: - MOLTRES_DSN: Database connection string (if dsn is None) - MOLTRES_ECHO: Enable SQLAlchemy echo mode (true/false) - MOLTRES_FETCH_FORMAT: “records”, “pandas”, or “polars” - MOLTRES_DIALECT: Override SQL dialect detection - MOLTRES_POOL_SIZE: Connection pool size - MOLTRES_MAX_OVERFLOW: Maximum pool overflow connections - MOLTRES_POOL_TIMEOUT: Pool timeout in seconds - MOLTRES_POOL_RECYCLE: Connection recycle time in seconds - MOLTRES_POOL_PRE_PING: Enable connection health checks (true/false)

Parameters:

  • dsn – Database connection string. Examples: - SQLite: “sqlite+aiosqlite:///path/to/database.db” - PostgreSQL: “postgresql+asyncpg://user:pass@host:port/dbname” - MySQL: “mysql+aiomysql://user:pass@host:port/dbname” If None, will use MOLTRES_DSN environment variable. Note: DSN should include async driver (e.g., +asyncpg, +aiomysql, +aiosqlite) Cannot be provided if engine or session is provided.

  • engine – SQLAlchemy async Engine instance to use. If provided, dsn is ignored. This gives users more flexibility to configure the engine themselves. Pool configuration options (pool_size, max_overflow, etc.) are ignored when using an existing engine. Cannot be provided if session is provided.

  • session – SQLAlchemy AsyncSession or SQLModel AsyncSession instance to use. If provided, dsn and engine are ignored. The session’s bind (async engine) will be used. This allows using Moltres with FastAPI’s dependency-injected async sessions. Cannot be provided if dsn or engine is provided.

  • **options – Optional configuration parameters (can also be set via environment variables): - echo: Enable SQLAlchemy echo mode for debugging (default: False) - fetch_format: Result format - “records”, “pandas”, or “polars” (default: “records”) - dialect: Override SQL dialect detection (e.g., “postgresql”, “mysql”) - pool_size: Connection pool size (default: None, uses SQLAlchemy default). Ignored if engine is provided. - max_overflow: Maximum pool overflow connections (default: None). Ignored if engine is provided. - pool_timeout: Pool timeout in seconds (default: None). Ignored if engine is provided. - pool_recycle: Connection recycle time in seconds (default: None). Ignored if engine is provided. - pool_pre_ping: Enable connection health checks (default: False). Ignored if engine is provided.

Returns:

AsyncDatabase instance for async querying and table operations

Return type:

AsyncDatabase

Raises:

  • ImportError – If async dependencies are not installed

  • ValueError – If neither dsn, engine, nor session is provided and MOLTRES_DSN is not set

  • ValueError – If multiple of dsn, engine, and session are provided

  • TypeError – If session is not a SQLAlchemy AsyncSession or SQLModel AsyncSession instance

Example

>>> import asyncio
>>> async def example():
...     # Using connection string with async context manager (recommended)
...     async with async_connect("sqlite+aiosqlite:///:memory:") as db:
...         from moltres.table.schema import column
...         await db.create_table("users", [column("id", "INTEGER")]).collect()
...         from moltres.io.records import AsyncRecords
...         records = AsyncRecords.from_list([{"id": 1}], database=db)
...         await records.insert_into("users")
...         table_handle = await db.table("users")
...         df = table_handle.select()
...         results = await df.collect()
...         assert len(results) == 1
...         assert results[0]["id"] == 1
...         # await db.close() called automatically on exit
...
...     # Using connection string (manual close)
...     db = async_connect("sqlite+aiosqlite:///:memory:")
...     await db.create_table("users", [column("id", "INTEGER")]).collect()
...     await db.close()
...     # Note: async examples require running in async context
...     # asyncio.run(example())

Expression helpers

moltres.col(name: str) → Column[source]

Create a Column expression from a column name.

This is the primary way to reference columns in Moltres queries. Column names can be simple (e.g., “age”) or qualified (e.g., “users.age”).

Parameters:

name – Column name as a string. Can be a simple name or qualified with table name (e.g., “table.column”).

Returns:

Column expression that can be used in DataFrame operations

Return type:

Column

Example

>>> from moltres import connect, col
>>> db = connect("sqlite:///:memory:")
>>> from moltres.table.schema import column
>>> _ = db.create_table("users", [column("id", "INTEGER"), column("name", "TEXT")]).collect()
>>> from moltres.io.records import :class:`Records`
>>> _ = :class:`Records`(_data=[{"id": 1, "name": "Alice"}], _database=db).insert_into("users")
>>> df = db.table("users").select(col("name"))
>>> results = df.collect()
>>> results[0]["name"]
'Alice'
>>> # Use in expressions
>>> df2 = db.table("users").select((col("id") * 2).alias("double_id"))
>>> results2 = df2.collect()
>>> results2[0]["double_id"]
2
>>> db.close()
moltres.lit(value: bool | int | float | str | None) → Column[source]

Create a literal column expression from a Python value.

Parameters:

value – The literal value (bool, int, float, str, or None)

Returns:

Column expression representing the literal value

Example

>>> from moltres import connect, col
>>> from moltres.expressions import functions as F
>>> db = connect("sqlite:///:memory:")
>>> from moltres.table.schema import column
>>> _ = db.create_table("test", [column("x", "INTEGER")]).collect()
>>> from moltres.io.records import :class:`Records`
>>> _ = :class:`Records`(_data=[{"x": 10}], _database=db).insert_into("test")
>>> # Use lit() to create literal values in expressions
>>> df = db.table("test").select((col("x") + F.lit(5)).alias("x_plus_5"))
>>> results = df.collect()
>>> results[0]["x_plus_5"]
15
>>> # String literals
>>> df2 = db.table("test").select(F.lit("constant").alias("constant_value"))
>>> results2 = df2.collect()
>>> results2[0]["constant_value"]
'constant'
>>> db.close()
moltres.column(name: str, type_name: str, nullable: bool = True, default: object | None = None, primary_key: bool = False, precision: int | None = None, scale: int | None = None) → ColumnDef[source]

Convenience helper for creating column definitions.

Parameters:

  • name – Column name

  • type_name – SQL type name (e.g., “INTEGER”, “TEXT”, “REAL”, “DECIMAL”)

  • nullable – Whether the column allows NULL values (default: True)

  • default – Default value for the column (default: None)

  • primary_key – Whether this column is a primary key (default: False)

  • precision – Precision for DECIMAL/NUMERIC types (default: None)

  • scale – Scale for DECIMAL/NUMERIC types (default: None)

Returns:

ColumnDef object for use in table creation

Return type:

ColumnDef

Example

>>> from moltres import connect
>>> from moltres.table.schema import column
>>> db = connect("sqlite:///:memory:")
>>> # Create table with column definitions
>>> _ = db.create_table(
...     "users",
...     [
...         column("id", "INTEGER", primary_key=True),
...         column("name", "TEXT", nullable=False),
...         column("age", "INTEGER"),
...         column("balance", "DECIMAL", precision=10, scale=2)
...     ]
... ).collect()
>>> from moltres.io.records import :class:`Records`
>>> _ = :class:`Records`(_data=[{"id": 1, "name": "Alice", "age": 30, "balance": 100.50}], _database=db).insert_into("users")
>>> df = db.table("users").select()
>>> results = df.collect()
>>> results[0]["name"]
'Alice'
>>> results[0]["age"]
30
>>> db.close()

Configuration

class moltres.config.MoltresConfig(engine: EngineConfig, default_schema: str | None = None, include_metadata: bool = False, allowed_paths: tuple[str, ...] | None=None, options: dict[str, object]=<factory>)[source]

Bases: object

Container for all runtime configuration knobs.

allowed_paths: tuple[str, ...] | None = None
default_schema: str | None = None
engine: EngineConfig
include_metadata: bool = False
options: dict[str, object]
moltres.config.create_config(dsn: str | None = None, engine: Engine | 'AsyncEngine | None' = None, session: object | None = None, **kwargs: EngineOptionValue) → MoltresConfig[source]

Convenience helper used by moltres.connect.

Supports environment variables for configuration: - MOLTRES_DSN: Database connection string - MOLTRES_ECHO: Enable SQLAlchemy echo mode (true/false) - MOLTRES_FETCH_FORMAT: “records”, “pandas”, or “polars” - MOLTRES_DIALECT: Override SQL dialect detection - MOLTRES_POOL_SIZE: Connection pool size - MOLTRES_MAX_OVERFLOW: Maximum pool overflow connections - MOLTRES_POOL_TIMEOUT: Pool timeout in seconds - MOLTRES_POOL_RECYCLE: Connection recycle time in seconds - MOLTRES_POOL_PRE_PING: Enable connection health checks (true/false) - MOLTRES_QUERY_TIMEOUT: Query execution timeout in seconds - MOLTRES_ALLOWED_PATHS: OS-separated list of allowed filesystem roots for file I/O

Parameters:

  • dsn – Database connection string (e.g., “sqlite:///example.db”). If None, will try MOLTRES_DSN environment variable. Cannot be provided if engine or session is provided.

  • engine – SQLAlchemy Engine instance to use. If provided, dsn is ignored. This gives users more flexibility to configure the engine themselves. Cannot be provided if session is provided.

  • session – SQLAlchemy Session or AsyncSession instance to use. If provided, dsn and engine are ignored. The session’s bind (engine) will be used. Cannot be provided if dsn or engine is provided.

  • **kwargs – Additional configuration options. Valid keys include: - echo: Enable SQLAlchemy echo mode - fetch_format: “records”, “pandas”, or “polars” - dialect: Override SQL dialect detection - pool_size: Connection pool size (ignored if engine is provided) - max_overflow: Maximum pool overflow connections (ignored if engine is provided) - pool_timeout: Pool timeout in seconds (ignored if engine is provided) - pool_recycle: Connection recycle time in seconds (ignored if engine is provided) - pool_pre_ping: Enable connection health checks (ignored if engine is provided) - query_timeout: Query execution timeout in seconds - allowed_paths: Tuple of directory roots permitted for file read/write paths - future: Use SQLAlchemy 2.0 style (default: True) - Other options are stored in config.options

Returns:

MoltresConfig instance with parsed configuration

Return type:

MoltresConfig

Raises:

  • ValueError – If neither dsn, engine, nor session is provided and MOLTRES_DSN is not set

  • ValueError – If multiple of dsn, engine, and session are provided

Performance hooks

moltres.register_performance_hook(event: str, callback: Callable[[str, float, dict[str, Any]], None]) → None[source]

Register a performance monitoring hook.

Parameters:

  • event – Event type - “query_start” or “query_end”

  • callback – Callback function that receives (sql, elapsed_time, metadata)

Example

>>> def log_slow_queries(sql: str, elapsed: float, metadata: dict):
...     if elapsed > 1.0:
...         print(f"Slow query ({elapsed:.2f}s): {sql[:100]}")
>>> register_performance_hook("query_end", log_slow_queries)
moltres.unregister_performance_hook(event: str, callback: Callable[[str, float, dict[str, Any]], None]) → None[source]

Unregister a performance monitoring hook.

Parameters:

  • event – Event type - “query_start” or “query_end”

  • callback – Callback function to remove