Database¶

Public API for async SQLite database operations with connection pooling and SQLCipher encryption. Provides a high-level interface for database operations with automatic connection management.

Tip

Pair this reference with Database for the feature guide.

Quick overview¶

AsyncDatabase provides the main entry point for database operations
ConnectionPool manages connection lifecycle with health checks and retry
PoolStats tracks pool usage statistics
is_sqlcipher_available() checks if SQLCipher encryption is available
Encryption keys can be sourced from passphrase, environment variable, or SOPS
All operations are async-first with context manager support

Configuration cascade¶

The module consults the loaded config for default values. A minimal config block:

db:
  pool:
    min_size: 2
    max_size: 20
    acquire_timeout: 30.0
  retry:
    max_attempts: 3
    delay: 0.5
  cipher:
    enabled: false        # Opt-in SQLCipher encryption
    key_source: env       # env | sops | passphrase
    key_env: "KSTLIB_DB_KEY"

Override any of these per instance:

from kstlib.db import AsyncDatabase

db = AsyncDatabase(
    "app.db",
    pool_min=5,
    pool_max=50,
    pool_timeout=60.0
)

Usage patterns¶

Basic database operations¶

from kstlib.db import AsyncDatabase

async with AsyncDatabase(":memory:") as db:
    await db.execute("CREATE TABLE test (id INTEGER)")
    await db.execute("INSERT INTO test VALUES (?)", (1,))
    row = await db.fetch_one("SELECT * FROM test")

Encrypted database with SOPS¶

from kstlib.db import AsyncDatabase

db = AsyncDatabase(
    "secure.db",
    cipher_sops="secrets.yml",
    cipher_sops_key="database_key"
)
async with db:
    await db.execute("SELECT * FROM sensitive_data")

Transaction with automatic rollback¶

async with db.transaction() as conn:
    await conn.execute("INSERT INTO accounts VALUES (?, ?)", (1, 1000))
    await conn.execute("INSERT INTO accounts VALUES (?, ?)", (2, 500))
    # Commits on success, rolls back on exception

Direct connection access¶

async with db.connection() as conn:
    await conn.execute("PRAGMA optimize")

Key resolution from various sources¶

from kstlib.db import resolve_cipher_key

# From passphrase
key = resolve_cipher_key(passphrase="direct-key")

# From environment variable
key = resolve_cipher_key(env_var="DB_KEY")

# From SOPS file
key = resolve_cipher_key(sops_path="secrets.yml", sops_key="db_key")

Module reference¶

Async database module for SQLite/SQLCipher.

Provides: - AsyncDatabase: High-level async interface - ConnectionPool: Connection pooling with retry - SQLCipher encryption via SOPS integration

Examples

Basic in-memory database:

>>> from kstlib.db import AsyncDatabase
>>> db = AsyncDatabase(":memory:")

Encrypted database with SOPS:

>>> db = AsyncDatabase(  
...     "app.db",
...     cipher_sops="secrets.yml",
...     cipher_sops_key="database_key"
... )

Usage as context manager:

>>> async with AsyncDatabase(":memory:") as db:  
...     await db.execute("CREATE TABLE test (id INTEGER)")
...     await db.execute("INSERT INTO test VALUES (?)", (1,))
...     row = await db.fetch_one("SELECT * FROM test")

class kstlib.db.AsyncDatabase(path, cipher_key=None, cipher_env=None, cipher_sops=None, cipher_sops_key='db_key', pool_min=1, pool_max=10, pool_timeout=30.0, max_retries=3, retry_delay=0.5, _pool=None, _resolved_key=None)[source]

Bases: object

Async database interface for SQLite/SQLCipher.

Provides connection pooling, encryption, and query helpers for async database operations.

Parameters:

path (str | Path) – Path to database file (or “:memory:” for in-memory).
cipher_key (str | None) – Direct encryption key for SQLCipher.
cipher_env (str | None) – Environment variable containing cipher key.
cipher_sops (str | Path | None) – Path to SOPS file containing cipher key.
cipher_sops_key (str) – Key name in SOPS file (default: “db_key”).
pool_min (int) – Minimum pool connections.
pool_max (int) – Maximum pool connections.
pool_timeout (float) – Acquire timeout in seconds.
max_retries (int) – Retry attempts on failure.
retry_delay (float) – Delay between retries.

Examples

Basic usage:

>>> db = AsyncDatabase(":memory:")
>>> db.path
':memory:'

With encryption:

>>> db = AsyncDatabase("app.db", cipher_key="secret")  

With SOPS:

>>> db = AsyncDatabase("app.db", cipher_sops="secrets.yml")  

path: str | Path

cipher_key: str | None = None

cipher_env: str | None = None

cipher_sops: str | Path | None = None

cipher_sops_key: str = 'db_key'

pool_min: int = 1

pool_max: int = 10

pool_timeout: float = 30.0

max_retries: int = 3

retry_delay: float = 0.5

__post_init__(self) → 'None' -> None[source]

Resolve cipher key and apply config defaults with hard limits.

async connect(self) → 'None' -> None[source]

Initialize the connection pool.

Called automatically on first operation, but can be called explicitly for eager initialization.

async close(self) → 'None' -> None[source]

Close all connections and shutdown the pool.

async __aenter__(self) → 'Self' -> Self[source]

Async context manager entry.

async __aexit__(self, exc_type: 'type[BaseException] | None', exc_val: 'BaseException | None', exc_tb: 'object') → 'None' -> None[source]

Async context manager exit.

connection(self) → 'AsyncGenerator[aiosqlite.Connection, None]' -> AsyncGenerator[aiosqlite.Connection, None][source]

Get a connection from the pool.

Yields:: Database connection.

Examples

>>> async with db.connection() as conn:  
...     await conn.execute("SELECT 1")

transaction(self) → 'AsyncGenerator[aiosqlite.Connection, None]' -> AsyncGenerator[aiosqlite.Connection, None][source]

Execute operations within a transaction.

Automatically commits on success, rolls back on error.

Yields:: Database connection within transaction.
Raises:: TransactionError – If transaction fails.

Examples

>>> async with db.transaction() as conn:  
...     await conn.execute("INSERT INTO users VALUES (?)", ("alice",))
...     await conn.execute("INSERT INTO users VALUES (?)", ("bob",))

async execute(self, sql: 'str', parameters: 'Sequence[Any] | None' = None) → 'aiosqlite.Cursor' -> aiosqlite.Cursor[source]

Execute a single SQL statement.

Parameters:

sql (str) – SQL statement to execute.
parameters (Sequence[Any] | None) – Query parameters.

Returns:

Cursor with results.

Return type:

aiosqlite.Cursor

Examples

>>> await db.execute("CREATE TABLE test (id INTEGER)")  

async executemany(self, sql: 'str', parameters: 'Sequence[Sequence[Any]]') → 'aiosqlite.Cursor' -> aiosqlite.Cursor[source]

Execute SQL statement for multiple parameter sets.

Parameters:

sql (str) – SQL statement to execute.
parameters (Sequence[Sequence[Any]]) – Sequence of parameter tuples.

Returns:

Cursor with results.

Return type:

aiosqlite.Cursor

Examples

>>> await db.executemany(  
...     "INSERT INTO test VALUES (?)",
...     [(1,), (2,), (3,)]
... )

async fetch_one(self, sql: 'str', parameters: 'Sequence[Any] | None' = None) → 'tuple[Any, ...] | None' -> tuple[Any, ...] | None[source]

Fetch a single row.

Parameters:

sql (str) – SQL query.
parameters (Sequence[Any] | None) – Query parameters.

Returns:

Row tuple or None if no results.

Return type:

tuple[Any, …] | None

Examples

>>> row = await db.fetch_one("SELECT * FROM test WHERE id=?", (1,))  

async fetch_all(self, sql: 'str', parameters: 'Sequence[Any] | None' = None) → 'list[tuple[Any, ...]]' -> list[tuple[Any, ...]][source]

Fetch all rows.

Parameters:

sql (str) – SQL query.
parameters (Sequence[Any] | None) – Query parameters.

Returns:

List of row tuples.

Return type:

list[tuple[Any, …]]

Examples

>>> rows = await db.fetch_all("SELECT * FROM test")  

async fetch_value(self, sql: 'str', parameters: 'Sequence[Any] | None' = None) → 'Any' -> Any[source]

Fetch a single value (first column of first row).

Parameters:

sql (str) – SQL query.
parameters (Sequence[Any] | None) – Query parameters.

Returns:

Single value or None.

Return type:

Any

Examples

>>> count = await db.fetch_value("SELECT count(*) FROM test")  

async table_exists(self, table_name: 'str') → 'bool' -> bool[source]

Check if a table exists.

Parameters:: table_name (str) – Name of the table.
Returns:: True if table exists.
Return type:: bool

Examples

>>> await db.table_exists("users")  
False

property stats: PoolStats: Get connection pool statistics.

property is_encrypted: bool: Whether database is configured for encryption.

property pool_size: int: Current number of connections in pool.

__init__(self, path: 'str | Path', cipher_key: 'str | None' = None, cipher_env: 'str | None' = None, cipher_sops: 'str | Path | None' = None, cipher_sops_key: 'str' = 'db_key', pool_min: 'int' = 1, pool_max: 'int' = 10, pool_timeout: 'float' = 30.0, max_retries: 'int' = 3, retry_delay: 'float' = 0.5, _pool: 'ConnectionPool | None' = None, _resolved_key: 'str | None' = None) → None -> None

class kstlib.db.ConnectionPool(db_path, min_size=1, max_size=10, acquire_timeout=30.0, max_retries=3, retry_delay=0.5, cipher_key=None, on_connect=None, _pool=<factory>, _connections=<factory>, _lock=<factory>, _closed=False, _stats=<factory>)[source]

Bases: object

Async connection pool for SQLite/SQLCipher databases.

Manages a pool of database connections with health checks and automatic retry on failures.

Parameters:

db_path (str) – Path to database file.
min_size (int) – Minimum connections to maintain.
max_size (int) – Maximum connections allowed.
acquire_timeout (float) – Timeout for acquiring connection.
max_retries (int) – Retry attempts on failure.
retry_delay (float) – Delay between retries.
cipher_key (str | None) – Optional encryption key for SQLCipher.
on_connect (Any | None) – Callback after connection established.

Examples

>>> pool = ConnectionPool(":memory:", min_size=1, max_size=5)
>>> pool.max_size
5

db_path: str

min_size: int = 1

max_size: int = 10

acquire_timeout: float = 30.0

max_retries: int = 3

retry_delay: float = 0.5

cipher_key: str | None = None

on_connect: Any | None = None

__post_init__(self) → 'None' -> None[source]

Validate and clamp configuration values to hard limits.

async acquire(self) → 'aiosqlite.Connection' -> aiosqlite.Connection[source]

Acquire a connection from the pool.

Returns:

Database connection.

Raises:

PoolExhaustedError – If no connection available within timeout.
DatabaseConnectionError – If connection creation fails after retries.

Return type:

aiosqlite.Connection

async release(self, conn: 'aiosqlite.Connection') → 'None' -> None[source]

Release a connection back to the pool.

Parameters:: conn (aiosqlite.Connection) – Connection to release.

connection(self) → 'AsyncGenerator[aiosqlite.Connection, None]' -> AsyncGenerator[aiosqlite.Connection, None][source]

Context manager for acquiring and releasing connections.

Yields:: Database connection.

Examples

>>> async with pool.connection() as conn:  
...     await conn.execute("SELECT 1")

async close(self) → 'None' -> None[source]

Close all connections and shutdown the pool.

property stats: PoolStats: Get pool statistics.

property size: int: Current number of connections in pool.

property is_closed: bool: Whether the pool is closed.

__init__(self, db_path: 'str', min_size: 'int' = 1, max_size: 'int' = 10, acquire_timeout: 'float' = 30.0, max_retries: 'int' = 3, retry_delay: 'float' = 0.5, cipher_key: 'str | None' = None, on_connect: 'Any | None' = None, _pool: 'asyncio.Queue[aiosqlite.Connection]' = <factory>, _connections: 'set[aiosqlite.Connection]' = <factory>, _lock: 'asyncio.Lock' = <factory>, _closed: 'bool' = False, _stats: 'PoolStats' = <factory>) → None -> None

exception kstlib.db.DatabaseConnectionError[source]

Bases: DatabaseError

Failed to establish database connection.

exception kstlib.db.DatabaseError[source]

Bases: KstlibError

Base exception for database operations.

exception kstlib.db.EncryptionError[source]

Bases: DatabaseError

Failed to decrypt or access encrypted database.

exception kstlib.db.PoolExhaustedError[source]

Bases: DatabaseError

Connection pool exhausted, no connections available.

class kstlib.db.PoolStats(total_connections=0, active_connections=0, idle_connections=0, total_acquired=0, total_released=0, total_timeouts=0, total_errors=0)[source]

Bases: object

Statistics for connection pool monitoring.

total_connections

Total connections created.

Type:: int

active_connections

Currently in-use connections.

Type:: int

idle_connections

Available connections in pool.

Type:: int

total_acquired

Total acquire operations.

Type:: int

total_released

Total release operations.

Type:: int

total_timeouts

Acquire operations that timed out.

Type:: int

total_errors

Connection errors encountered.

Type:: int

Examples

>>> stats = PoolStats()
>>> stats.total_connections
0

total_connections: int = 0

active_connections: int = 0

idle_connections: int = 0

total_acquired: int = 0

total_released: int = 0

total_timeouts: int = 0

total_errors: int = 0

__init__(self, total_connections: 'int' = 0, active_connections: 'int' = 0, idle_connections: 'int' = 0, total_acquired: 'int' = 0, total_released: 'int' = 0, total_timeouts: 'int' = 0, total_errors: 'int' = 0) → None -> None

exception kstlib.db.TransactionError[source]

Bases: DatabaseError

Transaction operation failed.

kstlib.db.apply_cipher_key(conn: 'sqlite3.Connection', key: 'str') → 'None' -> None[source]

Apply SQLCipher key to a connection.

Parameters:

conn (sqlite3.Connection) – SQLite connection object.
key (str) – Encryption key to apply.

Raises:

EncryptionError – If key application fails.

kstlib.db.is_sqlcipher_available() → 'bool' -> bool[source]

Check if sqlcipher3 is installed and available.

Returns:: True if sqlcipher3 can be imported.
Return type:: bool

Examples

>>> is_sqlcipher_available()  
True

kstlib.db.resolve_cipher_key(*, passphrase: 'str | None' = None, env_var: 'str | None' = None, sops_path: 'str | Path | None' = None, sops_key: 'str' = 'db_key') → 'str' -> str[source]

Resolve encryption key from various sources.

Priority: passphrase > env_var > sops_path

Parameters:

passphrase (str | None) – Direct passphrase string.
env_var (str | None) – Environment variable name containing the key.
sops_path (str | Path | None) – Path to SOPS-encrypted file.
sops_key (str) – Key name within SOPS file (default: “db_key”).

Returns:

Resolved encryption key.

Raises:

EncryptionError – If no key source provided or resolution fails.

Return type:

str

Examples

>>> key = resolve_cipher_key(passphrase="my-secret-key")
>>> len(key) > 0
True

Database class¶

Async database wrapper for SQLite/SQLCipher.

Provides a high-level async interface for database operations with: - Connection pooling - Automatic retry on transient failures - SQLCipher encryption support - Transaction management - Query helpers

class kstlib.db.database.AsyncDatabase(path, cipher_key=None, cipher_env=None, cipher_sops=None, cipher_sops_key='db_key', pool_min=1, pool_max=10, pool_timeout=30.0, max_retries=3, retry_delay=0.5, _pool=None, _resolved_key=None)[source]

Bases: object

Async database interface for SQLite/SQLCipher.

Provides connection pooling, encryption, and query helpers for async database operations.

Parameters:

path (str | Path) – Path to database file (or “:memory:” for in-memory).
cipher_key (str | None) – Direct encryption key for SQLCipher.
cipher_env (str | None) – Environment variable containing cipher key.
cipher_sops (str | Path | None) – Path to SOPS file containing cipher key.
cipher_sops_key (str) – Key name in SOPS file (default: “db_key”).
pool_min (int) – Minimum pool connections.
pool_max (int) – Maximum pool connections.
pool_timeout (float) – Acquire timeout in seconds.
max_retries (int) – Retry attempts on failure.
retry_delay (float) – Delay between retries.

Examples

Basic usage:

>>> db = AsyncDatabase(":memory:")
>>> db.path
':memory:'

With encryption:

>>> db = AsyncDatabase("app.db", cipher_key="secret")  

With SOPS:

>>> db = AsyncDatabase("app.db", cipher_sops="secrets.yml")  

path: str | Path

cipher_key: str | None = None

cipher_env: str | None = None

cipher_sops: str | Path | None = None

cipher_sops_key: str = 'db_key'

pool_min: int = 1

pool_max: int = 10

pool_timeout: float = 30.0

max_retries: int = 3

retry_delay: float = 0.5

__post_init__(self) → 'None' -> None[source]

Resolve cipher key and apply config defaults with hard limits.

async connect(self) → 'None' -> None[source]

Initialize the connection pool.

Called automatically on first operation, but can be called explicitly for eager initialization.

async close(self) → 'None' -> None[source]

Close all connections and shutdown the pool.

async __aenter__(self) → 'Self' -> Self[source]

Async context manager entry.

async __aexit__(self, exc_type: 'type[BaseException] | None', exc_val: 'BaseException | None', exc_tb: 'object') → 'None' -> None[source]

Async context manager exit.

connection(self) → 'AsyncGenerator[aiosqlite.Connection, None]' -> AsyncGenerator[aiosqlite.Connection, None][source]

Get a connection from the pool.

Yields:: Database connection.

Examples

>>> async with db.connection() as conn:  
...     await conn.execute("SELECT 1")

transaction(self) → 'AsyncGenerator[aiosqlite.Connection, None]' -> AsyncGenerator[aiosqlite.Connection, None][source]

Execute operations within a transaction.

Automatically commits on success, rolls back on error.

Yields:: Database connection within transaction.
Raises:: TransactionError – If transaction fails.

Examples

>>> async with db.transaction() as conn:  
...     await conn.execute("INSERT INTO users VALUES (?)", ("alice",))
...     await conn.execute("INSERT INTO users VALUES (?)", ("bob",))

async execute(self, sql: 'str', parameters: 'Sequence[Any] | None' = None) → 'aiosqlite.Cursor' -> aiosqlite.Cursor[source]

Execute a single SQL statement.

Parameters:

sql (str) – SQL statement to execute.
parameters (Sequence[Any] | None) – Query parameters.

Returns:

Cursor with results.

Return type:

aiosqlite.Cursor

Examples

>>> await db.execute("CREATE TABLE test (id INTEGER)")  

async executemany(self, sql: 'str', parameters: 'Sequence[Sequence[Any]]') → 'aiosqlite.Cursor' -> aiosqlite.Cursor[source]

Execute SQL statement for multiple parameter sets.

Parameters:

sql (str) – SQL statement to execute.
parameters (Sequence[Sequence[Any]]) – Sequence of parameter tuples.

Returns:

Cursor with results.

Return type:

aiosqlite.Cursor

Examples

>>> await db.executemany(  
...     "INSERT INTO test VALUES (?)",
...     [(1,), (2,), (3,)]
... )

async fetch_one(self, sql: 'str', parameters: 'Sequence[Any] | None' = None) → 'tuple[Any, ...] | None' -> tuple[Any, ...] | None[source]

Fetch a single row.

Parameters:

sql (str) – SQL query.
parameters (Sequence[Any] | None) – Query parameters.

Returns:

Row tuple or None if no results.

Return type:

tuple[Any, …] | None

Examples

>>> row = await db.fetch_one("SELECT * FROM test WHERE id=?", (1,))  

async fetch_all(self, sql: 'str', parameters: 'Sequence[Any] | None' = None) → 'list[tuple[Any, ...]]' -> list[tuple[Any, ...]][source]

Fetch all rows.

Parameters:

sql (str) – SQL query.
parameters (Sequence[Any] | None) – Query parameters.

Returns:

List of row tuples.

Return type:

list[tuple[Any, …]]

Examples

>>> rows = await db.fetch_all("SELECT * FROM test")  

async fetch_value(self, sql: 'str', parameters: 'Sequence[Any] | None' = None) → 'Any' -> Any[source]

Fetch a single value (first column of first row).

Parameters:

sql (str) – SQL query.
parameters (Sequence[Any] | None) – Query parameters.

Returns:

Single value or None.

Return type:

Any

Examples

>>> count = await db.fetch_value("SELECT count(*) FROM test")  

async table_exists(self, table_name: 'str') → 'bool' -> bool[source]

Check if a table exists.

Parameters:: table_name (str) – Name of the table.
Returns:: True if table exists.
Return type:: bool

Examples

>>> await db.table_exists("users")  
False

property stats: PoolStats: Get connection pool statistics.

property is_encrypted: bool: Whether database is configured for encryption.

property pool_size: int: Current number of connections in pool.

Connection pool¶

Connection pool with retry support for async SQLite.

Provides connection pooling with: - Configurable pool size - Connection health checks - Automatic retry on transient failures - Graceful shutdown

class kstlib.db.pool.ConnectionPool(db_path, min_size=1, max_size=10, acquire_timeout=30.0, max_retries=3, retry_delay=0.5, cipher_key=None, on_connect=None, _pool=<factory>, _connections=<factory>, _lock=<factory>, _closed=False, _stats=<factory>)[source]

Bases: object

Async connection pool for SQLite/SQLCipher databases.

Manages a pool of database connections with health checks and automatic retry on failures.

Parameters:

db_path (str) – Path to database file.
min_size (int) – Minimum connections to maintain.
max_size (int) – Maximum connections allowed.
acquire_timeout (float) – Timeout for acquiring connection.
max_retries (int) – Retry attempts on failure.
retry_delay (float) – Delay between retries.
cipher_key (str | None) – Optional encryption key for SQLCipher.
on_connect (Any | None) – Callback after connection established.

Examples

>>> pool = ConnectionPool(":memory:", min_size=1, max_size=5)
>>> pool.max_size
5

db_path: str

min_size: int = 1

max_size: int = 10

acquire_timeout: float = 30.0

max_retries: int = 3

retry_delay: float = 0.5

cipher_key: str | None = None

on_connect: Any | None = None

__post_init__(self) → 'None' -> None[source]

Validate and clamp configuration values to hard limits.

async acquire(self) → 'aiosqlite.Connection' -> aiosqlite.Connection[source]

Acquire a connection from the pool.

Returns:

Database connection.

Raises:

PoolExhaustedError – If no connection available within timeout.
DatabaseConnectionError – If connection creation fails after retries.

Return type:

aiosqlite.Connection

async release(self, conn: 'aiosqlite.Connection') → 'None' -> None[source]

Release a connection back to the pool.

Parameters:: conn (aiosqlite.Connection) – Connection to release.

connection(self) → 'AsyncGenerator[aiosqlite.Connection, None]' -> AsyncGenerator[aiosqlite.Connection, None][source]

Context manager for acquiring and releasing connections.

Yields:: Database connection.

Examples

>>> async with pool.connection() as conn:  
...     await conn.execute("SELECT 1")

async close(self) → 'None' -> None[source]

Close all connections and shutdown the pool.

property stats: PoolStats: Get pool statistics.

property size: int: Current number of connections in pool.

property is_closed: bool: Whether the pool is closed.

class kstlib.db.pool.PoolStats(total_connections=0, active_connections=0, idle_connections=0, total_acquired=0, total_released=0, total_timeouts=0, total_errors=0)[source]

Bases: object

Statistics for connection pool monitoring.

total_connections

Total connections created.

Type:: int

active_connections

Currently in-use connections.

Type:: int

idle_connections

Available connections in pool.

Type:: int

total_acquired

Total acquire operations.

Type:: int

total_released

Total release operations.

Type:: int

total_timeouts

Acquire operations that timed out.

Type:: int

total_errors

Connection errors encountered.

Type:: int

Examples

>>> stats = PoolStats()
>>> stats.total_connections
0

total_connections: int = 0

active_connections: int = 0

idle_connections: int = 0

total_acquired: int = 0

total_released: int = 0

total_timeouts: int = 0

total_errors: int = 0

Cipher utilities¶

SQLCipher integration with SOPS secret resolution.

Provides secure key management for encrypted SQLite databases. Keys can be loaded from: - Direct passphrase - Environment variable - SOPS-encrypted file via kstlib.secrets

kstlib.db.cipher.apply_cipher_key(conn: 'sqlite3.Connection', key: 'str') → 'None' -> None[source]

Apply SQLCipher key to a connection.

Parameters:

conn (sqlite3.Connection) – SQLite connection object.
key (str) – Encryption key to apply.

Raises:

EncryptionError – If key application fails.

kstlib.db.cipher.resolve_cipher_key(*, passphrase: 'str | None' = None, env_var: 'str | None' = None, sops_path: 'str | Path | None' = None, sops_key: 'str' = 'db_key') → 'str' -> str[source]

Resolve encryption key from various sources.

Priority: passphrase > env_var > sops_path

Parameters:

passphrase (str | None) – Direct passphrase string.
env_var (str | None) – Environment variable name containing the key.
sops_path (str | Path | None) – Path to SOPS-encrypted file.
sops_key (str) – Key name within SOPS file (default: “db_key”).

Returns:

Resolved encryption key.

Raises:

EncryptionError – If no key source provided or resolution fails.

Return type:

str

Examples

>>> key = resolve_cipher_key(passphrase="my-secret-key")
>>> len(key) > 0
True

Async SQLCipher¶

Async SQLCipher wrapper built on top of aiosqlite.

Provides async database connections with SQLCipher AES-256 encryption. This module wraps aiosqlite to use sqlcipher3 instead of standard sqlite3.

Requirements:: pip install kstlib[db-crypto] # Installs sqlcipher3

Examples

Basic encrypted connection:

import asyncio
from kstlib.db.aiosqlcipher import connect

async def main():
    async with connect(":memory:", cipher_key="secret") as db:
        await db.execute("CREATE TABLE test (id INTEGER)")

asyncio.run(main())

kstlib.db.aiosqlcipher.connect(database: 'str | Path', *, cipher_key: 'str', iter_chunk_size: 'int' = 64, **kwargs: 'Any') → 'Connection' -> Connection[source]

Create an async connection to an encrypted SQLite database.

This function is a drop-in replacement for aiosqlite.connect() that uses SQLCipher for AES-256 encryption. The cipher key is applied immediately after connection using PRAGMA key.

Parameters:

database (str | Path) – Path to database file or “:memory:” for in-memory.
cipher_key (str) – Encryption key for SQLCipher (required).
iter_chunk_size (int) – Rows to fetch per iteration (default: 64).
**kwargs (Any) – Additional arguments passed to sqlcipher3.connect().

Returns:

Async Connection object (same interface as aiosqlite.Connection).

Raises:

EncryptionError – If sqlcipher3 is not installed or key is empty.
sqlite3.DatabaseError – If database exists but key is wrong.

Return type:

Connection

Examples

>>> async with connect("app.db", cipher_key="secret") as db:  
...     await db.execute("CREATE TABLE users (id INTEGER)")

>>> # With SOPS key resolution:
>>> from kstlib.db.cipher import resolve_cipher_key
>>> key = resolve_cipher_key(sops_path="secrets.yml")  
>>> async with connect("app.db", cipher_key=key) as db:  
...     pass

kstlib.db.aiosqlcipher.is_sqlcipher_available() → 'bool' -> bool[source]

Check if sqlcipher3 is installed and available.

Returns:: True if sqlcipher3 can be imported.
Return type:: bool

Examples

>>> is_sqlcipher_available()  
True

Exceptions¶

Database module exceptions.

exception kstlib.db.exceptions.DatabaseError[source]

Bases: KstlibError

Base exception for database operations.

exception kstlib.db.exceptions.DatabaseConnectionError[source]

Bases: DatabaseError

Failed to establish database connection.

exception kstlib.db.exceptions.EncryptionError[source]

Bases: DatabaseError

Failed to decrypt or access encrypted database.

exception kstlib.db.exceptions.PoolExhaustedError[source]

Bases: DatabaseError

Connection pool exhausted, no connections available.

exception kstlib.db.exceptions.TransactionError[source]

Bases: DatabaseError

Transaction operation failed.