API Reference

Protocols

ConnectionPoolProtocol

source

Protocol for connection pools.

max_connections

property max_connections() -> int

source

connection_timeout

property connection_timeout() -> float

source

initialize

async def initialize() -> None

source

Initialize the connection pool.

shutdown

async def shutdown() -> None

source

Shutdown the connection pool.

get_connection

def get_connection() -> AbstractAsyncContextManager[Any]

source

Get a connection from the pool.

get_pool_stats

async def get_pool_stats() -> dict[str, Any]

source

Get pool statistics.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Check pool health.

get_query_stats

async def get_query_stats(time_range_seconds: int = 3600) -> dict[str, Any]

source

Get query statistics.

warm

async def warm(count: int | None = None) -> None

source

Pre-create count connections to avoid cold-start latency.

Parameters

Parameter	Type	Description
count	int | None	Number of connections to open. Defaults to `min_connections` (or the pool minimum) if not specified.

validate_connections

async def validate_connections() -> int

source

Validate all idle connections in the pool, evicting dead ones.

Returns

Type	Description
int	Number of valid connections remaining after validation.

---

DatabaseProviderProtocol

source

Protocol for database providers.

This defines the interface that all database providers must implement, regardless of the underlying database technology.

Example

class PostgresProvider:
    async def connect(self) -> None:
        self._pool = await asyncpg.create_pool(self._dsn)

    async def execute_query(
        self,
        sql: str,
        params: list[Any] | None = None,
    ) -> QueryResult:
        async with self._pool.acquire() as conn:
            rows = await conn.fetch(sql, *params or [])
            return QueryResult(rows=list(map(dict, rows)), ...)

connect

async def connect() -> None

source

Establish connection to the database.

disconnect

async def disconnect() -> None

source

Close connection to the database.

is_connected

async def is_connected() -> bool

source

Check if database is connected.

get_primary_pool

async def get_primary_pool() -> ConnectionPoolProtocol

source

Return the primary connection pool.

For multi-backend setups, returns the backend marked primary: true. Raises NoPrimaryBackendError if no primary is marked or zero backends.

Returns

Type	Description
ConnectionPoolProtocol	The primary connection pool.

Raises

Exception	Description
NoPrimaryBackendError	If no primary backend is configured.

execute_query

async def execute_query(
    sql: str,
    params: list[Any] | None = None,
    **kwargs: Any
) -> QueryResult

source

Execute a SELECT query.

A statement that returns no rows (DDL, PRAGMA, a data-modifying statement without RETURNING) is not an error: it succeeds with an empty row list. The framework’s stores issue CREATE TABLE IF NOT EXISTS through this method when they provision their own table, and an implementation that treats “no result set” as failure makes that provisioning unusable.

Parameters

Parameter	Type	Description
sql	str	SQL query string.
params	list[Any] | None	Query parameters. JSON-shaped values (`dict/list`) are encoded for binding — see encode_bind_value. **kwargs: Additional options.

Returns

Type	Description
QueryResult	QueryResult with rows and execution metadata.

execute_insert

async def execute_insert(
    table: str,
    data: dict[str, Any],
    **kwargs: Any
) -> InsertResult

source

Execute an INSERT operation.

JSON-shaped values in data (dict/list) are encoded to JSON text before binding, matching the JSONB/TEXT columns the framework’s stores declare. Relational drivers cannot bind a Python container, so an implementation that passes one through fails at the driver — silently, when the caller does not inspect success.

Parameters

Parameter	Type	Description
table	str	Table name.
data	dict[str, Any]	Column-value mapping. **kwargs: Additional options.

Returns

Type	Description
InsertResult	InsertResult with inserted ID and influenced rows.

Raises

Exception	Description
IntegrityError	Constraint violations (foreign key, unique, NOT NULL, CHECK) propagate as the typed `quadkit.sql.exceptions.IntegrityError` family rather than folding into a failed result, so callers can answer for them by type.

execute_update

async def execute_update(
    table: str,
    data: dict[str, Any],
    where_clause: str,
    where_params: list[Any] | None = None,
    **kwargs: Any
) -> UpdateResult

source

Execute an UPDATE operation.

JSON-shaped values in data and where_params are encoded for binding, as in execute_insert.

A value may be a Delta instead of a plain value. It renders col = col + ? — a relative move, so a counter changes inside the statement rather than through a read the caller then writes back. An implementation that binds it as if it were a value writes the Delta object itself into the column, so this is part of the contract, not an optional extension.

Parameters

Parameter	Type	Description
table	str	Table name.
data	dict[str, Any]	Column-value updates, or `Delta` moves.
where_clause	str	WHERE condition.
where_params	list[Any] | None	Parameters for WHERE clause. **kwargs: Additional options.

Returns

Type	Description
UpdateResult	UpdateResult with affected rows.

Raises

Exception	Description
IntegrityError	As in execute_insert.

execute_delete

async def execute_delete(
    table: str,
    where_clause: str,
    where_params: list[Any] | None = None,
    **kwargs: Any
) -> DeleteResult

source

Execute a DELETE operation.

JSON-shaped values in where_params are encoded for binding, as in execute_insert.

Parameters

Parameter	Type	Description
table	str	Table name.
where_clause	str	WHERE condition.
where_params	list[Any] | None	Parameters for WHERE clause. **kwargs: Additional options.

Returns

Type	Description
DeleteResult	DeleteResult with affected rows.

Raises

Exception	Description
IntegrityError	As in execute_insert — a delete restricted by referencing rows raises the typed foreign-key error.

execute

async def execute(
    sql: str,
    params: Any = None
) -> QueryResult

source

Execute a raw SQL query with parameters.

Parameters

Parameter	Type	Description
sql	str	SQL query string.
params	Any	Query parameters.

Returns

Type	Description
QueryResult	QueryResult with execution results.

transaction

def transaction(isolation_level: IsolationLevel | None = None) -> AbstractAsyncContextManager[Any]

source

Context manager for transactions.

Parameters

Parameter	Type	Description
isolation_level	IsolationLevel | None	Optional ANSI SQL isolation level. When `None` the driver's default isolation level is used.

Example

async with db.transaction(isolation_level=IsolationLevel.SERIALIZABLE):
    await db.execute("INSERT INTO ...")
    await db.execute("UPDATE ...")

begin_transaction

async def begin_transaction() -> None

source

Begin a transaction.

commit_transaction

async def commit_transaction() -> None

source

Commit current transaction.

rollback_transaction

async def rollback_transaction() -> None

source

Rollback current transaction.

table_exists

async def table_exists(table_name: str) -> bool

source

Check if a table exists.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Perform health check on database connection.

scoped_context

def scoped_context() -> AbstractAsyncContextManager[Any]

source

Return an async context manager that establishes a scoped session.

The scoped context binds a database session to the current async context so that get_scoped_connection can retrieve it without passing the connection around explicitly.

Returns

Type	Description
AbstractAsyncContextManager[Any]	Async context manager that yields no value (or the session).

get_scoped_connection

async def get_scoped_connection() -> ConnectionProtocol

source

Return the connection bound to the current scoped context.

Must be called within an active scoped_context block.

Returns

Type	Description
ConnectionProtocol	Active ConnectionProtocol for the current scope.

acquire

async def acquire() -> ConnectionProtocol

source

Acquire a connection from the pool for manual management.

Use this when you need fine-grained control over connection lifecycle, but prefer scoped_context when possible for automatic cleanup.

Returns

Type	Description
ConnectionProtocol	An acquired connection that must be released via release.

Example

conn = await db.acquire()
try:
    result = await conn.execute("SELECT * FROM users")
finally:
    await db.release(conn)

release

async def release(connection: ConnectionProtocol) -> None

source

Release a connection back to the pool.

Parameters

Parameter	Type	Description
connection	ConnectionProtocol	Connection acquired via acquire.

---

QueryLoggerProtocol

source

Protocol for logging database queries.

Implement this protocol to enable query logging in database providers. The concrete quadkit.sql.logging.BaseQueryLogger implements this interface and adds helper methods for querying the stored entries.

log_query

async def log_query(entry: QueryLogEntry) -> None

source

Log a query execution.

get_recent_queries

async def get_recent_queries(limit: int = 100) -> list[QueryLogEntry]

source

Return the most recent limit entries.

get_slow_queries

async def get_slow_queries(
    threshold_seconds: float,
    limit: int = 50
) -> list[QueryLogEntry]

source

Return entries whose execution time exceeds threshold_seconds.

get_query_stats

async def get_query_stats(time_range_seconds: int = 3600) -> dict[str, Any]

source

Return aggregated statistics over a time window.

---

RepositoryProtocol

source

Protocol for full repository operations.

Extends ReadOnlyRepositoryProtocol with write operations.

Example

class UserRepository:
    async def save(self, entity: User) -> User:
        if entity.id:
            await self.db.update("users", entity.dict()).where(id=entity.id)
        else:
            entity.id = await self.db.insert("users", entity.dict())
        return entity

    async def delete(self, id: str) -> bool:
        result = await self.db.delete("users").where(id=id)
        return result.affected_rows > 0

save

async def save(entity: T) -> T

source

Save (create or update) an entity.

Parameters

Parameter	Type	Description
entity	T	Entity to save.

Returns

Type	Description
T	Saved entity with any generated fields populated.

delete

async def delete(item_id: str) -> bool

source

Delete entity by ID.

Parameters

Parameter	Type	Description
item_id	str	Entity identifier.

Returns

Type	Description
bool	True if deleted, False if not found.

save_many

async def save_many(entities: list[T]) -> list[T]

source

Save (create or update) multiple entities in a single operation.

Implementations SHOULD execute this as a batch for efficiency.

Parameters

Parameter	Type	Description
entities	list[T]	Entities to save.

Returns

Type	Description
list[T]	Saved entities with any generated fields populated.

delete_many

async def delete_many(item_ids: list[str]) -> int

source

Delete multiple entities by ID.

Implementations SHOULD execute this as a batch for efficiency.

Parameters

Parameter	Type	Description
item_ids	list[str]	Entity identifiers to delete.

Returns

Type	Description
int	Number of entities actually deleted.

---

Classes

AbstractConnectionPool

source

Base connection pool with common functionality

__init__

def __init__(
    min_connections: int = 1,
    max_connections: int = 10,
    connection_timeout: float = 30.0,
    max_idle_time: float = 300.0,
    poll_interval: float | None = None,
    **kwargs: Any
)

source

max_connections

property max_connections() -> int

source

Get maximum number of connections.

connection_timeout

property connection_timeout() -> float

source

Get connection timeout.

initialize

async def initialize() -> None

source

Initialize the connection pool

shutdown

async def shutdown() -> None

source

Shutdown the connection pool

get_connection

async def get_connection() -> AsyncGenerator[Any, None]

source

Get a connection from the pool

return_connection

async def return_connection(connection: Any) -> None

source

Return a connection to the pool.

This method is generally called automatically by get_connection(). Manual call is supported for specialized testing or legacy code.

get_pool_stats

async def get_pool_stats() -> dict[str, Any]

source

Get pool statistics

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Check pool health

---

AuditEntry

source

Immutable record of a security-relevant or compliance-relevant operation.

Canonical audit type used by all packages. Supersedes the former AuditEvent (simpler, no old/new values) and the AuditEntry from contracts/observability/audit.

Attributes: action: Dot-notation action identifier (e.g. "user.update"). actor_id: ID of the user or service that performed the action. resource_type: Kind of affected resource (e.g. "User"). resource_id: ID of the affected resource. outcome: "success" or "failure". severity: Severity level (defaults to MEDIUM). occurred_at: UTC datetime of the event (defaults to now). metadata: Arbitrary additional context. old_values: Field values before the action (optional). new_values: Field values after the action (optional). source: Originating subsystem (e.g. "sql", "admin", "ai"). tenant_id: Multi-tenant scoping (optional). correlation_id: Request correlation ID for tracing (optional). causation_id: Causing event ID for event-driven traces (optional). command_payload_hash: SHA-256 hash of the command payload (optional). payload_size_bytes: Size of the payload in bytes (optional). checksum: HMAC-SHA256 checksum of the persisted row, populated by stores on read-back (None for pre-checksum entries).

---

AuditRepositoryMixin

source

Mixin to add audit capabilities to repositories

Automatically logs all changes made through the repository.

__init__

def __init__(
    *args: Any,
    audit_logger: AuditLoggerProtocol | None = None,
    db_ctx: DbContext | None = None,
    **kwargs: Any
) -> None

source

create

async def create(entity: Any) -> Any

source

Create with audit logging

update

async def update(entity: Any) -> Any

source

Update with audit logging

delete

async def delete(key: Any) -> Any

source

Delete with audit logging

---

Column

source

A validated SQL column name.

Example

c = Column("email")
query = f"SELECT {c} FROM users"  # SELECT "email" FROM users

c = Column("email")
query = f"SELECT {c} FROM users"  # SELECT "email" FROM users

---

ConsoleQueryLogger

source

Query logger that outputs to console.

Useful for development and debugging.

__init__

def __init__(
    slow_query_threshold: float = 1.0,
    max_entries: int = 1000,
    log_level: str = 'INFO',
    sample_rate: float = 1.0
)

source

---

DatabaseConfig

source

Hierarchical root configuration for Quadkit Database.

Attributes: name: Configuration name (default: “database”) enabled: Whether the database module is enabled backend: Database backend/connection configuration pool: Connection pool settings operations: Database operation settings audit_hmac_key: Optional HMAC key (plain text or base64) for audit checksum signing. Set via QK_DB__AUDIT_HMAC_KEY. backends: List of named database backend configurations. When non-empty, enables multi-DB mode. The entry with primary=True (or the first entry) is also bound without a name for backward compatibility.

validate_production_security

def validate_production_security() -> DatabaseConfig

source

Block insecure database configurations in production.

from_url

def from_url(
    cls,
    url: str,
    name: str = 'database'
) -> DatabaseConfig

source

Create a DatabaseConfig from a URL string.

from_env

def from_env(
    cls,
    prefix: str = const.ENV_PREFIX
) -> DatabaseConfig

source

Create a DatabaseConfig by reading {const.ENV_PREFIX}* environment variables.

Environment variables are stripped of prefix, lowercased, and __ is treated as a nesting delimiter, so QK_SQL__BACKEND__URL maps to backend.url.

Parameters

Parameter	Type	Description
prefix	str	Environment variable prefix to strip. Defaults to `"{const.ENV_PREFIX}"`.

Returns

Type	Description
DatabaseConfig	A `DatabaseConfig` populated from the current environment, falling back to field defaults for any unset variables.

Example

# With QK_SQL__BACKEND__URL=postgresql://user:pass@host/db set:
config = DatabaseConfig.from_env()
assert config.backend.url.get_secret_value().startswith("postgresql")

# With QK_SQL__BACKEND__URL=postgresql://user:pass@host/db set:
config = DatabaseConfig.from_env()
assert config.backend.url.get_secret_value().startswith("postgresql")

from_named

def from_named(
    cls,
    entry: NamedDatabaseConfig,
    base: DatabaseConfig | None = None
) -> DatabaseConfig

source

Build a DatabaseConfig from a NamedDatabaseConfig.

Parameters

Parameter	Type	Description
entry	NamedDatabaseConfig	The named backend entry to convert.
base	DatabaseConfig | None	Optional root DatabaseConfig to inherit operation settings from.

Returns

Type	Description
DatabaseConfig	A DatabaseConfig suitable for constructing a DatabaseService.

---

DatabaseLockStore

source

Persistent distributed lock store backed by a SQL table.

Implements LockStoreProtocol using the application’s shared connection pool. Expired locks are lazily reclaimed on the next acquire() attempt.

Parameters

Parameter	Type	Description
db_provider		Database provider injected from the DI container.
table_name		SQL table name used for lock storage.
auto_create_tables		Reserved — apply DDL via migrations in production.
clock		Clock for timestamps.

__init__

def __init__(
    db_provider: DatabaseProviderProtocol,
    table_name: str = 'locks',
    auto_create_tables: bool = True
) -> None

source

acquire

async def acquire(
    lock_name: str,
    owner: str,
    ttl: int
) -> bool

source

Attempt to acquire the named lock.

Returns True if the lock was acquired (or reclaimed after expiry); False if a non-expired lock already exists under a different owner.

release

async def release(
    lock_name: str,
    owner: str
) -> bool

source

Release the lock only if it is held by owner.

Returns True if the lock was released; False if it was not held by owner or did not exist.

extend

async def extend(
    lock_name: str,
    owner: str,
    ttl: int
) -> bool

source

Extend the TTL of a currently-held lock.

Returns True if the extension was applied; False if the lock is not held by owner.

---

DatabaseModule

source

Async SQLAlchemy ORM: repositories, unit-of-work, migrations, and connection pooling.

Registers DatabaseProviderProtocol and UnitOfWorkProtocol for constructor injection.

Call configure to configure and register the full database infrastructure, or stub for an isolated setup with no external service dependencies.

Usage

from quadkit.sql.config import DatabaseConfig

@module(
    imports=[
        DatabaseModule.configure(
            DatabaseConfig(url="postgresql+asyncpg://localhost/mydb")
        )
    ]
)
class AppModule(Module):
    pass

from quadkit.sql.config import DatabaseConfig

@module(
    imports=[
        DatabaseModule.configure(
            DatabaseConfig(url="postgresql+asyncpg://localhost/mydb")
        )
    ]
)
class AppModule(Module):
    pass

Shorthand with a URL string

@module(
    imports=[DatabaseModule.configure("postgresql+asyncpg://localhost/mydb")]
)
class AppModule(Module):
    pass

@module(
    imports=[DatabaseModule.configure("postgresql+asyncpg://localhost/mydb")]
)
class AppModule(Module):
    pass

configure

def configure(
    cls,
    config: DatabaseConfig | Any | None = None,
    migration_dir: str = 'migrations',
    enable_migrations: bool = False,
    enable_query_logging: bool = False,
    config_key: str = 'sql'
) -> DynamicModule

source

Create a DatabaseModule with explicit configuration.

Parameters

Parameter	Type	Description
config	DatabaseConfig | Any | None	A DatabaseConfig instance, a database URL string, or `None to resolve config from the container's ConfigProtocol` at boot time.
migration_dir	str	Path to the Alembic migrations directory relative to the project root. Defaults to `"migrations"`.
enable_migrations	bool	Run Alembic migrations automatically on boot. Defaults to `False` for safety in production deployments.
enable_query_logging	bool	Emit every SQL statement to the structured logger for debugging. Defaults to `False`; enable only in development environments.

Returns

Type	Description
DynamicModule	A DynamicModule descriptor.

scope

def scope(
    cls,
    *repositories: type
) -> DynamicModule

source

Scope repository classes into a feature module.

Registers the given repository classes as providers and exports them for constructor injection within the feature. The parent module graph must already include DatabaseModule.configure — this does not create a new database connection.

Uses the anonymous token pattern so both configure() and scope() can coexist in the same compiled graph without a ModuleDuplicateError.

Example

@module(
    imports=[
        DatabaseModule.configure(config),
        DatabaseModule.scope(UserRepository, OrderRepository),
    ]
)
class UserFeatureModule(Module):
    pass

@module(
    imports=[
        DatabaseModule.configure(config),
        DatabaseModule.scope(UserRepository, OrderRepository),
    ]
)
class UserFeatureModule(Module):
    pass

Parameters

Parameter	Type	Description

Returns

Type	Description
DynamicModule	A DynamicModule scoped to this feature.

stub

def stub(
    cls,
    config: DatabaseConfig | None = None
) -> DynamicModule

source

Create a DatabaseModule suitable for unit and integration testing.

Uses an in-memory or minimal backend with no external service dependencies.

Parameters

Parameter	Type	Description
config	DatabaseConfig | None	Optional DatabaseConfig override. Uses safe in-memory defaults when `None`.

Returns

Type	Description
DynamicModule	A DynamicModule descriptor.

---

DatabaseMonitor

source

Comprehensive database monitoring

__init__

def __init__(
    collector: DbMetricsCollector | None = None,
    slow_query_threshold: float = 1.0
)

source

start_pool_monitoring

async def start_pool_monitoring(
    pool: Any,
    provider: Any | None = None
) -> None

source

Start monitoring a connection pool.

Parameters

Parameter	Type	Description
pool	Any	The pool or provider object to monitor.
provider	Any | None	Optional provider used for active `SELECT 1 probing. When omitted, the pool itself is used as the provider only if it exposes scoped_context()`.

stop_pool_monitoring

async def stop_pool_monitoring() -> None

source

Stop pool monitoring

get_query_monitor

def get_query_monitor() -> QueryMonitor

source

Get the query monitor

get_transaction_monitor

def get_transaction_monitor() -> TransactionMonitor

source

Get the transaction monitor

get_health_checker

def get_health_checker() -> DatabaseHealthChecker

source

Get the health checker

get_stats

async def get_stats() -> dict[str, Any]

source

Get comprehensive monitoring statistics

perform_health_check

async def perform_health_check(
    connection_string: str,
    pool: Any = None
) -> dict[str, Any]

source

Perform comprehensive health check

---

DatabaseProvider

source

DI provider that registers database services into the container. config: DatabaseConfig | None _explicit_config: DatabaseConfig | None _db_provider: DatabaseService | None _migration_dir: str | Path | None

Single registration path for all database-related singletons and scoped services: DatabaseService, protocol bindings, connection pool, query logger, migration manager, audit logger, unit-of-work, and optional metrics/tracer.

Manages lifecycle: connect on boot, disconnect on shutdown. Multi-backend support: when config.backends is non-empty, each backend is registered under its name via container.singleton(name=entry.name). The primary backend (primary=True or first entry) also receives the unnamed bindings for backward compatibility.

__init__

def __init__(
    config: DatabaseConfig | str | None = None,
    migration_dir: str = 'migrations',
    enable_migrations: bool = False,
    config_key: str | None = None
) -> None

source

boot

async def boot(container: ContainerResolverProtocol) -> None

source

Boot all database backends and wire optional observability integrations.

In multi-backend mode all backends are connected in parallel. In single-backend mode the existing sequential boot is preserved.

get_primary_pool

async def get_primary_pool() -> ConnectionPoolProtocol

source

Return the primary connection pool.

For multi-backend setups, returns the backend marked primary: true. Raises NoPrimaryBackendError if no primary is marked or zero backends.

Returns

Type	Description
ConnectionPoolProtocol	The primary connection pool.

Raises

Exception	Description
NoPrimaryBackendError	If no primary backend is configured.

shutdown

async def shutdown() -> None

source

Disconnect all database backends in reverse registration order.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Check provider health across all registered backends.

In multi-backend mode the overall status is the worst individual status.

Parameters

Parameter	Type	Description
timeout	float	Maximum seconds to wait for health check response.

Returns

Type	Description
HealthCheckResult	HealthCheckResult with status and component details.

---

DatabaseResilienceHandler

source

Manages resilience pipelines for database operations.

Accepts an optional ResiliencePipelineFactory injected by the DI container. When a factory is present it builds real circuit-breaker / retry pipelines with database-specific defaults. When None is passed a lightweight no-op pipeline is used, so quadkit-sql has no hard runtime dependency on a resilience extension.

__init__

def __init__(pipeline_factory: ResiliencePipelineFactoryProtocol | None = None) -> None

source

Initialise the resilience handler.

Parameters

Parameter	Type	Description
pipeline_factory	ResiliencePipelineFactoryProtocol | None	Optional factory provided by `a resilience extension via the DI container. Pass None` for no-op behaviour.

get_pipeline

def get_pipeline(name: str) -> ResiliencePipelineProtocol | _NoOpPipeline

source

Get or create a resilience pipeline for a named database operation.

Parameters

Parameter	Type	Description
name	str	The name of the pipeline (usually the provider name).

Returns

Type	Description
ResiliencePipelineProtocol | _NoOpPipeline	A ResiliencePipelineProtocol when a factory was injected, otherwise a pass-through no-op pipeline.

---

DatabaseSecretStore

source

Async secret store backed by the application database.

Implements AsyncSecretStoreProtocol using a SQL table via the shared connection pool from DatabaseProviderProtocol.

Parameters

Parameter	Type	Description
db_provider		Database provider injected from the DI container.
table_name		SQL table name used for secret storage.
auto_create_tables		Reserved — apply DDL via migrations in production.

__init__

def __init__(
    db_provider: DatabaseProviderProtocol,
    table_name: str = 'secrets',
    auto_create_tables: bool = True
) -> None

source

get

async def get(name: str) -> str | None

source

Return the secret value for name, or None if absent.

get_bulk

async def get_bulk(*names: str) -> dict[str, str]

source

Return a mapping of name → value for all requested secrets.

Parameters

Parameter	Type	Description
names	str	One or more secret names.

Returns

Type	Description
dict[str, str]	Dict containing only the names that were found.

set

async def set(
    name: str,
    value: str
) -> None

source

Write or overwrite a secret value.

delete

async def delete(name: str) -> None

source

Remove a secret. No-op if absent.

---

DatabaseService

source

Database facade that manages a database driver and its lifecycle.

Wraps a DatabaseDriver (engine-specific driver) and provides connection pooling, transaction management, resilience patterns, and query logging. For registering this into the DI container, see DatabaseProvider in di_provider.py.

Delegates specialized logic to:

• _ConnectionMixin: Connection lifecycle, transactions, and scoped contexts.
• _QueryMixin: Query execution and result normalisation.
• _HealthMixin: Health checks and table existence.
• DatabaseManager: Scoped connections, UoW, and pooling.
• DatabaseResilienceHandler: Retry and circuit breaker pipelines.
• Driver-specific protocols: SQLite, Postgres, MySQL, etc.

Example

Basic usage

provider = DatabaseService(url="postgresql://localhost/mydb")
await provider.boot()
result = await provider.execute_query("SELECT * FROM users")
await provider.shutdown()

provider = DatabaseService(url="postgresql://localhost/mydb")
await provider.boot()
result = await provider.execute_query("SELECT * FROM users")
await provider.shutdown()

Using transactions

async with provider.transaction():
    await provider.execute_insert("users", {"name": "John"})
    await provider.execute_update("users", {"name": "Jane"}, "id = ?", (1,))

async with provider.transaction():
    await provider.execute_insert("users", {"name": "John"})
    await provider.execute_update("users", {"name": "Jane"}, "id = ?", (1,))

committed_read_state

def committed_read_state() -> CommittedReadState

source

Delegate safety inspection only to the verified local SQLite driver.

register_driver

def register_driver(
    cls,
    name: str,
    loader: ProviderFactory
) -> None

source

Register a database driver loader by name.

Parameters

Parameter	Type	Description
name	str	The driver name (e.g., "sqlite", "postgres", "mysql").
loader	ProviderFactory	A callable that returns a DatabaseProviderProtocol instance.

get_driver_loader

def get_driver_loader(
    cls,
    name: str
) -> ProviderFactory | None

source

Get a driver loader by name.

Parameters

Parameter	Type	Description
name	str	The driver name to look up.

Returns

Type	Description
ProviderFactory | None	The driver loader function or None if not found.

__init__

def __init__(
    config: DatabaseConfig | str | None = None,
    connection_pool: ConnectionPoolProtocol | None = None,
    query_logger: QueryLoggerProtocol | None = None,
    migration_manager: MigrationManagerProtocol | None = None,
    **kwargs: Any
) -> None

source

Initialize the DatabaseService.

Parameters

Parameter	Type	Description
config	DatabaseConfig | str | None	DatabaseConfig instance or connection URL string.
connection_pool	ConnectionPoolProtocol | None	Optional connection pool implementation.
query_logger	QueryLoggerProtocol | None	Optional query logger for debugging.
migration_manager	MigrationManagerProtocol | None	Optional migration manager for schema changes. **kwargs: Additional provider-specific options.

database_type

property database_type() -> str

source

Return the type of database (e.g., ‘postgres’, ‘sqlite’).

set_context

def set_context(context: Context | None) -> None

source

Attach an optional shared request context to query logging.

set_hook_registry

def set_hook_registry(hooks: HookRegistryProtocol | None) -> None

source

Attach an optional hook registry to the service runtime path.

from_config

def from_config(
    cls,
    config: DatabaseConfig,
    **context: Any
) -> DatabaseService

source

Create a DatabaseService from a DatabaseConfig.

Parameters

Parameter	Type	Description
config	DatabaseConfig	The database configuration. **context: Additional context options.

Returns

Type	Description
DatabaseService	A new DatabaseService instance.

connect

async def connect() -> None

source

Establish connection to the database.

disconnect

async def disconnect() -> None

source

Close connection to the database.

get_primary_pool

async def get_primary_pool() -> ConnectionPoolProtocol

source

Return the primary connection pool.

---

DatabaseStateStore

source

State store that uses quadkit-sql's DatabaseService for connections.

Shares the application’s connection pool with the main application path, participates in DatabaseService health monitoring, and is driven by the same connection configuration as the rest of the application.

Parameters

Parameter	Type	Description
db_provider		Database provider injected from the DI container.
table_name		SQL table name used for state storage.
auto_create_tables		Reserved — table DDL should be applied via migrations in production.

__init__

def __init__(
    db_provider: DatabaseProviderProtocol,
    table_name: str = 'app_state',
    auto_create_tables: bool = True
) -> None

source

get

async def get(key: str) -> Any | None

source

Return the value stored under key, or None if absent.

set

async def set(
    key: str,
    value: Any,
    ttl: int | None = None
) -> None

source

Persist value under key with an optional TTL in seconds.

delete

async def delete(key: str) -> None

source

Remove the entry for key. No-op if absent.

get_many

async def get_many(keys: list[str]) -> dict[str, Any]

source

Fetch multiple keys in a single query.

Parameters

Parameter	Type	Description
keys	list[str]	List of storage keys.

Returns

Type	Description
dict[str, Any]	Dict of found key → deserialized value; absent keys are omitted.

set_many

async def set_many(
    items: dict[str, Any],
    ttl: int | None = None
) -> None

source

Persist multiple key-value pairs.

Parameters

Parameter	Type	Description
items	dict[str, Any]	Mapping of key → JSON-serializable value.
ttl	int | None	Optional TTL in seconds applied to every entry.

---

DbContext

source

Typed, injectable context for the database layer.

Wraps an injected ContextVarRegistry and exposes both generic get / set access via ContextKey and read-only convenience properties for well-known db keys.

An optional core_ctx reference lets RequestContextManager bridge values into the core framework context on enter and properly restore them on exit.

__init__

def __init__(
    registry: ContextVarRegistry,
    *,
    core_ctx: Context | None = None
) -> None

source

registry

property registry() -> ContextVarRegistry

source

The underlying registry (for advanced use / wiring).

core_ctx

property core_ctx() -> Context | None

source

Optional core Context used for bridging values.

get

def get(
    key: ContextKey[T],
    default: T | None = None
) -> T | None

source

Read the current value for key, returning default when unset.

set

def set(
    key: ContextKey[T],
    value: T
) -> contextvars.Token[T | None]

source

Write value for key and return a reset token.

reset

def reset(
    key: ContextKey[T],
    token: contextvars.Token[T | None]
) -> None

source

Restore the previous value of key using a token from set.

set_tenant_from_scope

def set_tenant_from_scope(scope: dict[str, Any]) -> contextvars.Token[str | None] | None

source

Set the tenant from the bridge payload or a full ASGI scope.

Accepts the current bridge payload {"tenant_id": "..."}, or a full ASGI scope carrying scope["state"]["tenant"] (a TenantInfo-like object). Returns None when no tenant is present; otherwise writes TENANT_ID and returns the reset token.

Parameters

Parameter	Type	Description
scope	dict[str, Any]	Bridge payload dict or ASGI scope dict.

Returns

Type	Description
contextvars.Token[str | None] | None	A `contextvars.Token for reset_tenant, or None`.

reset_tenant

def reset_tenant(token: contextvars.Token[str | None] | None) -> None

source

Reset the tenant context using a token from set_tenant_from_scope.

Parameters

Parameter	Type	Description
token	contextvars.Token[str | None] | None	Token returned by `set_tenant_from_scope`.

request_id

property request_id() -> str | None

source

Current request ID.

user_id

property user_id() -> str | None

source

Current user ID.

tenant_id

property tenant_id() -> str | None

source

Current tenant ID.

user

property user() -> Any | None

source

Current user object.

rate_limit_remaining

property rate_limit_remaining() -> int | None

source

Remaining rate-limit quota for this request.

rate_limit_limit

property rate_limit_limit() -> int | None

source

Maximum rate-limit quota per window.

rate_limit_reset

property rate_limit_reset() -> int | None

source

Unix timestamp when the rate-limit window resets.

snapshot

def snapshot() -> dict[str, Any]

source

Return a snapshot of all non-None db context values.

---

DeleteResult

source

Result of a delete operation.

---

Entity

source

Simple entity base class using Domain models from contracts.

This replaces the complex ORM Model metaclass with a clean, simple base class for domain entities.

get_table_name

def get_table_name(cls) -> str

source

Get the table name for this entity.

to_dict

def to_dict() -> dict[str, Any]

source

Convert entity to dictionary.

from_dict

def from_dict(
    cls,
    data: dict[str, Any]
) -> Entity

source

Create entity from dictionary.

---

F

source

Field reference used to build typed repository filters.

__init__

def __init__(reference: Any) -> None

source

in_

def in_(values: Iterable[Any]) -> Filter

source

Create an IN filter.

not_in

def not_in(values: Iterable[Any]) -> Filter

source

Create a NOT IN filter.

like

def like(pattern: str) -> Filter

source

Create a LIKE filter.

ilike

def ilike(pattern: str) -> Filter

source

Create a case-insensitive LIKE filter (portable across dialects).

contains

def contains(value: Any) -> Filter

source

Create a containment filter using wildcard matching.

is_null

def is_null() -> Filter

source

Create an IS NULL filter.

is_not_null

def is_not_null() -> Filter

source

Create an IS NOT NULL filter.

between

def between(
    low: Any,
    high: Any
) -> Filter

source

Create a BETWEEN filter.

---

FTSDialect

source

Supported full-text search dialects.

---

FTSResult

source

Thin wrapper around a list of entities that attaches FTS metadata.

Attributes: items: Matched entities in relevance order. total: Total number of results before limit/offset (if available). dialect: The SQL dialect used for the search.

---

FileQueryLogger

source

Query logger that writes to a file.

Useful for production logging and analysis.

__init__

def __init__(
    log_file: str,
    slow_query_threshold: float = 1.0,
    max_entries: int = 10000,
    sample_rate: float = 1.0,
    **_kwargs: Any
)

source

---

Filter

source

Composable SQL filter expression.

to_sql

def to_sql() -> tuple[str, list[Any]]

source

Render the filter expression to SQL and parameters.

---

GenericRepository

source

Generic repository with type safety.

Provides strongly-typed CRUD operations for entities.

__init__

def __init__(
    provider: DatabaseProviderProtocol,
    table_name: str,
    entity_class: type[TEntity],
    key_field: str = 'id',
    soft_delete_enabled: bool = False,
    multi_tenant: bool | None = None,
    rls_policy: RowLevelSecurityPolicy | None = None,
    *,
    db_ctx: DbContext | None = None
)

source

create

async def create(entity: Any) -> TEntity

source

Create entity and return a typed TEntity when entity_class is configured.

update

async def update(
    entity: Any,
    check_version: bool | None = None
) -> TEntity

source

Update entity and return a typed TEntity when entity_class is configured.

find

async def find(
    filters: dict[str, Any] | Filter | list[Filter] | tuple[Filter, Ellipsis] | None = None,
    limit: int | None = None,
    offset: int | None = 0,
    sort_by: str | None = None,
    sort_order: str = 'asc',
    columns: list[str] | None = None,
    include_deleted: bool = False
) -> list[TEntity]

source

Find entities with optional dict-style filters.

Convenience wrapper around find_many() that accepts filters as a plain dict instead of **kwargs.

count

async def count(
    filters: dict[str, Any] | Filter | list[Filter] | tuple[Filter, Ellipsis] | None = None,
    *filter_expressions: Any,
    include_deleted: bool = False,
    **kwargs: Any
) -> int

source

Count entities with optional dict-style filters.

Convenience wrapper around the mixin count() that accepts filters as a plain dict (or Filter objects) instead of **kwargs, mirroring find.

get

async def get(key: TKey) -> TEntity | None

source

Get a single entity by primary key.

Convenience alias for find_by_id().

save_many

async def save_many(entities: list[TEntity]) -> list[TEntity]

source

Persist entities, creating or updating each as appropriate.

RepositoryProtocol declares the batch primitives as ... stubs, so an implementation that does not override them answers None to every caller. This delegates to save in input order, which keeps the returned list aligned with entities and runs the post-save hooks once per entity.

Parameters

Parameter	Type	Description
entities	list[TEntity]	Entities to save.

Returns

Type	Description
list[TEntity]	The saved entities, in input order.

delete_many

async def delete_many(item_ids: list[TKey]) -> int

source

Delete every entity whose primary key appears in item_ids.

Delegates to delete so the post-delete hooks run for each entity actually removed.

Parameters

Parameter	Type	Description
item_ids	list[TKey]	Primary keys of the entities to delete.

Returns

Type	Description
int	Number of entities actually deleted.

---

HealthCheckResult

source

Result of a health check.

to_dict

def to_dict() -> dict[str, Any]

source

Convert to dictionary format.

is_healthy

def is_healthy() -> bool

source

Check if status is healthy.

is_degraded

def is_degraded() -> bool

source

Check if status is degraded.

---

HealthStatus

source

Unified health status.

---

Identifier

source

Validated, quoted, immutable SQL identifier.

Created once, validated once, safe to interpolate into SQL forever. __str__ returns the dialect-quoted form so f"SELECT * FROM {table}" is always safe when table is an Identifier.

Parameters

Parameter	Type	Description
name		The raw identifier name.
dialect		Target SQL dialect for quoting. Defaults to PostgreSQL.

Raises

Exception	Description
InvalidIdentifierError	If the name is empty, contains invalid characters, or exceeds the dialect's maximum length.

__init__

def __init__(
    name: str,
    *,
    dialect: SQLDialect = SQLDialect.POSTGRESQL
) -> None

source

name

property name() -> str

source

The raw, unquoted identifier name.

quoted

property quoted() -> str

source

The dialect-quoted identifier, safe for SQL interpolation.

dialect

property dialect() -> SQLDialect

source

The SQL dialect this identifier is quoted for.

is_reserved

property is_reserved() -> bool

source

Whether this identifier is a SQL reserved word.

---

InsertResult

source

Result of an insert operation.

---

MemoryQueryLogger

source

Query logger that stores entries in memory.

Useful for testing and development.

__init__

def __init__(
    slow_query_threshold: float = 1.0,
    max_entries: int = 1000,
    sample_rate: float = 1.0
)

source

log_query

async def log_query(entry: QueryLogEntry) -> None

source

Log a query execution to memory.

clear_logs

async def clear_logs() -> None

source

Clear all logged entries.

get_entries_by_sql

async def get_entries_by_sql(sql_pattern: str) -> list[QueryLogEntry]

source

Get entries matching SQL pattern.

get_failed_queries

async def get_failed_queries() -> list[QueryLogEntry]

source

Get all failed query entries.

---

MySQLFTSQuery

source

Builds MySQL ``MATCH … AGAINST`` full-text search queries.

Requires a FULLTEXT index on columns for non-keyword queries.

Attributes: table: Table name to search. columns: List of column names included in the FULLTEXT index. boolean_mode: When True uses IN BOOLEAN MODE; otherwise uses natural language mode.

__init__

def __init__(
    table: str,
    columns: list[str],
    boolean_mode: bool = False
) -> None

source

build

def build(
    query: str,
    extra_filters: dict[str, Any] | None = None,
    limit: int | None = 50,
    offset: int = 0
) -> tuple[str, list[Any]]

source

Return (sql, params) ready for provider.execute_query.

Parameters

Parameter	Type	Description
query	str	The search terms supplied by the user.
extra_filters	dict[str, Any] | None	Optional equality filters applied as `AND` clauses.
limit	int | None	Maximum rows to return.
offset	int	Row offset for pagination.

Returns

Type	Description
tuple[str, list[Any]]	A tuple of the SQL string and positional parameters.

---

PostgresFTSQuery

source

Builds PostgreSQL full-text search queries.

Uses to_tsvector (GIN index friendly) and to_tsquery / plainto_tsquery depending on plain mode.

Attributes: table: Table name to search. columns: List of column names to include in the tsvector. config: PostgreSQL text search configuration (e.g. 'english'). plain: When True uses plainto_tsquery which is safer for user-supplied input (no operator syntax needed).

__init__

def __init__(
    table: str,
    columns: list[str],
    config: str = 'english',
    plain: bool = True
) -> None

source

build

def build(
    query: str,
    extra_filters: dict[str, Any] | None = None,
    limit: int | None = 50,
    offset: int = 0
) -> tuple[str, list[Any]]

source

Return (sql, params) ready for provider.execute_query.

Parameters

Parameter	Type	Description
query	str	The search terms supplied by the user.
extra_filters	dict[str, Any] | None	Optional equality filters applied as `AND` clauses.
limit	int | None	Maximum rows to return.
offset	int	Row offset for pagination.

Returns

Type	Description
tuple[str, list[Any]]	A tuple of the SQL string and positional parameters.

---

QualifiedTable

source

A schema-qualified table name: ``schema.table``.

Example

qt = QualifiedTable(Schema("public"), Table("users"))
query = f"SELECT * FROM {qt}"  # SELECT * FROM "public"."users"

qt = QualifiedTable(Schema("public"), Table("users"))
query = f"SELECT * FROM {qt}"  # SELECT * FROM "public"."users"

__init__

def __init__(
    schema: Schema,
    table: Table
) -> None

source

schema

property schema() -> Schema

source

The schema component.

table

property table() -> Table

source

The table component.

quoted

property quoted() -> str

source

The fully-qualified, quoted identifier.

---

QueryLogEntry

source

Represents a database query log entry.

This is a concrete dataclass rather than a Protocol so that callers can instantiate it directly.

---

QueryLoggerBase

source

Base query logger with common functionality.

__init__

def __init__(
    slow_query_threshold: float = 1.0,
    max_entries: int = 1000,
    sample_rate: float = 1.0
)

source

Initialise the query logger.

Parameters

Parameter	Type	Description
slow_query_threshold	float	Queries exceeding this duration in seconds are always logged as warnings, regardless of *sample_rate*.
max_entries	int	Maximum number of log entries to keep in memory.
sample_rate	float	Fraction of normal (non-slow, non-error) queries to log, in the range `(0.0, 1.0]. Use 1.0 (default) to log all queries. Set to e.g. 0.1` to log only 10 %% of queries in high-QPS production deployments. Slow queries and error queries are always logged regardless of this setting.

log_query

async def log_query(entry: QueryLogEntry) -> None

source

Log a query execution, honouring the configured sample rate.

Slow queries (exceeding slow_query_threshold) and failed queries are always recorded regardless of sample_rate. Normal successful queries are subject to sampling when sample_rate < 1.0.

Subclasses should override _on_log to add backend-specific output (e.g. to console or a file) rather than overriding this method, so that the sampling decision is computed exactly once per entry.

Parameters

Parameter	Type	Description
entry	QueryLogEntry	The query log entry to record.

get_recent_queries

async def get_recent_queries(limit: int = 100) -> list[QueryLogEntry]

source

Get recent query logs.

get_slow_queries

async def get_slow_queries(
    threshold_seconds: float,
    limit: int = 50
) -> list[QueryLogEntry]

source

Get slow query logs.

get_query_stats

async def get_query_stats(time_range_seconds: int = 3600) -> dict[str, Any]

source

Get query statistics.

---

QueryResult

source

Result of a database query.

Implements the iterator and sequence protocols so that code expecting a plain list[dict] from a query result continues to work after the return type is normalised to QueryResult.

Example

result = await provider.execute_query("SELECT * FROM users")
for row in result:          # iterate directly
    print(row["email"])
if not result:              # bool coercion
    raise LookupError("no rows")
first = result[0]           # index access
rows = list(result)         # convert to plain list

result = await provider.execute_query("SELECT * FROM users")
for row in result:          # iterate directly
    print(row["email"])
if not result:              # bool coercion
    raise LookupError("no rows")
first = result[0]           # index access
rows = list(result)         # convert to plain list

---

RawSQL

source

Raw SQL string that bypasses validation.

This is a marker type for SQL that should not be validated or quoted.

Warning

Never pass user-supplied values directly into this query string. Always use parameterized queries or the provided binding mechanisms to prevent SQL injection vulnerabilities.

Attributes: sql: The raw SQL string.

__init__

def __init__(sql: str) -> None

source

---

RequestContextManager

source

Async context manager that sets request-scoped db context variables and restores them on exit.

All dependencies are injected via a DbContext instance

db_ctx = create_db_context(core_ctx=core_ctx)

async with RequestContextManager(db_ctx, request_id="abc-123"):
    assert db_ctx.request_id == "abc-123"
# values are restored here

db_ctx = create_db_context(core_ctx=core_ctx)

async with RequestContextManager(db_ctx, request_id="abc-123"):
    assert db_ctx.request_id == "abc-123"
# values are restored here

Core-context bridging

When ``db_ctx.core_ctx`` is not ``None``, the manager will
automatically propagate ``request_id``, ``user_id`` and
``tenant_id`` into the core context on enter and reset them
on exit.

<div class='qk-methods'>
<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>__init__</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>__init__</span><span class='qk-c'>(</span>
    <span class='qk-n'>db_ctx</span><span class='qk-c'>: </span><span class='qk-t'><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#dbcontext'>DbContext</a></span><span class='qk-c'>,</span>
    <span class='qk-d'>*</span><span class='qk-n'></span><span class='qk-c'>,</span>
    <span class='qk-n'>request_id</span><span class='qk-c'>: </span><span class='qk-t'>str <span class='qk-c'><span class='qk-c'>|</span></span> <span class='qk-d'>None</span></span><span class='qk-c'> = </span><span class='qk-d'>None</span><span class='qk-c'>,</span>
    <span class='qk-n'>user_id</span><span class='qk-c'>: </span><span class='qk-t'>str <span class='qk-c'><span class='qk-c'>|</span></span> <span class='qk-d'>None</span></span><span class='qk-c'> = </span><span class='qk-d'>None</span><span class='qk-c'>,</span>
    <span class='qk-n'>tenant_id</span><span class='qk-c'>: </span><span class='qk-t'>str <span class='qk-c'><span class='qk-c'>|</span></span> <span class='qk-d'>None</span></span><span class='qk-c'> = </span><span class='qk-d'>None</span><span class='qk-c'>,</span>
    <span class='qk-n'>user</span><span class='qk-c'>: </span><span class='qk-t'>Any <span class='qk-c'><span class='qk-c'>|</span></span> <span class='qk-d'>None</span></span><span class='qk-c'> = </span><span class='qk-d'>None</span><span class='qk-c'>,</span>
    <span class='qk-n'>rate_limit_remaining</span><span class='qk-c'>: </span><span class='qk-t'>int <span class='qk-c'><span class='qk-c'>|</span></span> <span class='qk-d'>None</span></span><span class='qk-c'> = </span><span class='qk-d'>None</span><span class='qk-c'>,</span>
    <span class='qk-n'>rate_limit_limit</span><span class='qk-c'>: </span><span class='qk-t'>int <span class='qk-c'><span class='qk-c'>|</span></span> <span class='qk-d'>None</span></span><span class='qk-c'> = </span><span class='qk-d'>None</span><span class='qk-c'>,</span>
    <span class='qk-n'>rate_limit_reset</span><span class='qk-c'>: </span><span class='qk-t'>int <span class='qk-c'><span class='qk-c'>|</span></span> <span class='qk-d'>None</span></span><span class='qk-c'> = </span><span class='qk-d'>None</span>
<span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-d'>None</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/context/manager.py#L60' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>


</div>

<hr class='qk-rule' />

<div data-pagefind-weight='10'>

### `RowLevelSecurityPolicy`

</div>

<span class='qk-meta' data-api-type='Classes'></span>

<div class='qk-source-row'><a class='qk-source' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/row_level_security.py#L153' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>
Carries the column declarations and manages the context-var scope.



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tn'>`columns`</td><td class='qk-tt'></td><td class='qk-td'>One or more <a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#scopecolumn'>ScopeColumn</a> instances that define which columns are used as the scope predicate.</td></tr></tbody></table></div>


The policy instance is typically a **singleton per repository class**
registered in the DI container

```python
container.register(
    RowLevelSecurityPolicy,
    lambda: RowLevelSecurityPolicy(columns=[ScopeColumn("user_id")]),
    lifetime="singleton",
)
```


    container.register(
        RowLevelSecurityPolicy,
        lambda: RowLevelSecurityPolicy(columns=[ScopeColumn("user_id")]),
        lifetime="singleton",
    )

<div class='qk-methods'>
<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>__init__</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>__init__</span><span class='qk-c'>(</span><span class='qk-n'>columns</span><span class='qk-c'>: </span><span class='qk-t'>list<span class='qk-c'>[</span><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#scopecolumn'>ScopeColumn</a><span class='qk-c'>]</span></span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-d'>None</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/row_level_security.py#L170' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>



<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>columns</span></div><pre class='qk-sig-pre'><span class='qk-k'>property </span><span class='qk-f'>columns</span><span class='qk-c'>(</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'>list<span class='qk-c'>[</span><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#scopecolumn'>ScopeColumn</a><span class='qk-c'>]</span></span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/row_level_security.py#L189' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

The scope columns declared for this policy.

<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>activate</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>activate</span><span class='qk-c'>(</span><span class='qk-d'>**</span><span class='qk-n'>values</span><span class='qk-c'>: </span><span class='qk-t'>Any</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'>Generator<span class='qk-c'>[</span><span class='qk-d'>None</span><span class='qk-c'>,</span> <span class='qk-d'>None</span><span class='qk-c'>,</span> <span class='qk-d'>None</span><span class='qk-c'>]</span></span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/row_level_security.py#L198' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Activate a scope for the duration of the ``with`` block.

Pass one keyword argument per <a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#scopecolumn'>ScopeColumn</a>, mapping
the column name to the principal's value

```python
with policy.activate(user_id="abc-123"):
    rows = await repo.find_many()
```


    with policy.activate(user_id="abc-123"):
        rows = await repo.find_many()



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody></tbody></table></div>




<div class='qk-table-wrap'><span class='qk-label'>Raises</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-45'>Exception</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tt'>ValueError</td><td class='qk-td'>If a required column is missing from *values*.</td></tr></tbody></table></div>



<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>admin_bypass</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>admin_bypass</span><span class='qk-c'>(</span>
    <span class='qk-d'>*</span><span class='qk-n'></span><span class='qk-c'>,</span>
    <span class='qk-n'>reason</span><span class='qk-c'>: </span><span class='qk-t'>str</span><span class='qk-c'>,</span>
    <span class='qk-n'>table</span><span class='qk-c'>: </span><span class='qk-t'>str</span>
<span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'>Generator<span class='qk-c'>[</span><span class='qk-d'>None</span><span class='qk-c'>,</span> <span class='qk-d'>None</span><span class='qk-c'>,</span> <span class='qk-d'>None</span><span class='qk-c'>]</span></span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/row_level_security.py#L234' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Bypass the RLS predicate for the duration of the ``with`` block.

Emits a structured WARNING log event so the bypass is auditable.
The *reason* string is mandatory — it will appear in the log record.



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tn'>`reason`</td><td class='qk-tt'>str</td><td class='qk-td'>Human-readable justification for the bypass (e.g. ``"support-export-ticket-42"``).</td></tr><tr><td class='qk-tn'>`table`</td><td class='qk-tt'>str</td><td class='qk-td'>The table being accessed (used in the audit log).</td></tr></tbody></table></div>




<div class='qk-table-wrap'><span class='qk-label'>Yields</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-45'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tt'>Generator<span class='qk-c'>[</span><span class='qk-d'>None</span><span class='qk-c'>,</span> <span class='qk-d'>None</span><span class='qk-c'>,</span> <span class='qk-d'>None</span><span class='qk-c'>]</span></td><td class='qk-td'>``None`` — the body runs with RLS bypassed.</td></tr></tbody></table></div>



<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>is_admin_bypass_active</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>is_admin_bypass_active</span><span class='qk-c'>(</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'>bool</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/row_level_security.py#L268' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Return ``True`` if an admin bypass is currently in effect.

<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>current_scope</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>current_scope</span><span class='qk-c'>(</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'>ActiveScope <span class='qk-c'><span class='qk-c'>|</span></span> <span class='qk-d'>None</span></span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/row_level_security.py#L272' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Return the currently active ActiveScope, or ``None``.

<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>require_scope</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>require_scope</span><span class='qk-c'>(</span>
    <span class='qk-d'>*</span><span class='qk-n'></span><span class='qk-c'>,</span>
    <span class='qk-n'>table</span><span class='qk-c'>: </span><span class='qk-t'>str</span><span class='qk-c'>,</span>
    <span class='qk-n'>operation</span><span class='qk-c'>: </span><span class='qk-t'>str</span>
<span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'>ActiveScope</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/row_level_security.py#L276' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Return the active scope, or raise :exc:`NoSecurityPolicyError`.

Called by repository mixins on paths that apply this policy. Admin
bypass suppresses the check.



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tn'>`table`</td><td class='qk-tt'>str</td><td class='qk-td'>Table name (for the error message).</td></tr><tr><td class='qk-tn'>`operation`</td><td class='qk-tt'>str</td><td class='qk-td'>SQL operation name (``"SELECT"``, etc.).</td></tr></tbody></table></div>




<div class='qk-table-wrap'><span class='qk-label'>Returns</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-45'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tt'>ActiveScope</td><td class='qk-td'>The active ActiveScope.</td></tr></tbody></table></div>




<div class='qk-table-wrap'><span class='qk-label'>Raises</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-45'>Exception</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tt'><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#nosecuritypolicyerror'>NoSecurityPolicyError</a></td><td class='qk-td'>If no scope is active and no admin bypass is in effect.</td></tr></tbody></table></div>


</div>

<hr class='qk-rule' />

<div data-pagefind-weight='10'>

### `SQLConnectionReadyHook`

</div>

<span class='qk-meta' data-api-type='Classes'></span>

<div class='qk-source-row'><a class='qk-source' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/hooks.py#L20' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>
Payload fired when the SQL connection pool is initialised and ready.

Attributes:
    backend: Database backend identifier (e.g. ``"postgresql"``).

<hr class='qk-rule' />

<div data-pagefind-weight='10'>

### `SQLDialect`

</div>

<span class='qk-meta' data-api-type='Classes'></span>

<div class='qk-source-row'><a class='qk-source' href='https://github.com/dbtinoy-/quadkit/blob/main/core/quadkit-contracts/src/quadkit/contracts/data/sql/sql_dialect.py#L14' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>
Supported SQL database dialects.

<hr class='qk-rule' />

<div data-pagefind-weight='10'>

### `SQLTransactionBegunHook`

</div>

<span class='qk-meta' data-api-type='Classes'></span>

<div class='qk-source-row'><a class='qk-source' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/hooks.py#L31' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>
Payload fired when a database transaction begins.

<hr class='qk-rule' />

<div data-pagefind-weight='10'>

### `SQLTransactionEndedHook`

</div>

<span class='qk-meta' data-api-type='Classes'></span>

<div class='qk-source-row'><a class='qk-source' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/hooks.py#L36' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>
Payload fired when a database transaction completes (commit or rollback).

Attributes:
    committed: ``True`` if the transaction was committed; ``False`` if rolled back.

<hr class='qk-rule' />

<div data-pagefind-weight='10'>

### `SQLiteProvider`

</div>

<span class='qk-meta' data-api-type='Classes'></span>

<div class='qk-source-row'><a class='qk-source' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/providers/sqlite_provider.py#L40' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>
SQLite database provider implementation

Uses aiosqlite for async SQLite operations.

<div class='qk-methods'>
<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>committed_read_state</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>committed_read_state</span><span class='qk-c'>(</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'>CommittedReadState</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/providers/sqlite_provider.py#L54' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Verify this implementation, its pool and observable transaction state.

The execution/lease locks, not this snapshot alone, prevent another
task's uncommitted rows being exposed. Unknown subclasses/pools and
disabled implementation capabilities are deliberately refused.

<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>__init__</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>__init__</span><span class='qk-c'>(</span>
    <span class='qk-n'>database_path</span><span class='qk-c'>: </span><span class='qk-t'>str</span><span class='qk-c'> = </span><span class='qk-s'>':memory:'</span><span class='qk-c'>,</span>
    <span class='qk-d'>**</span><span class='qk-n'>kwargs</span><span class='qk-c'>: </span><span class='qk-t'>Any</span>
<span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-d'>None</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/providers/sqlite_provider.py#L82' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>



<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>table_exists</span></div><pre class='qk-sig-pre'><span class='qk-k'>async </span><span class='qk-k'>def </span><span class='qk-f'>table_exists</span><span class='qk-c'>(</span><span class='qk-n'>table_name</span><span class='qk-c'>: </span><span class='qk-t'>str</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'>bool</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/providers/sqlite_provider.py#L272' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Check if table exists in SQLite

<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>get_table_columns</span></div><pre class='qk-sig-pre'><span class='qk-k'>async </span><span class='qk-k'>def </span><span class='qk-f'>get_table_columns</span><span class='qk-c'>(</span><span class='qk-n'>table_name</span><span class='qk-c'>: </span><span class='qk-t'>str</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'>list<span class='qk-c'>[</span>dict<span class='qk-c'>[</span>str<span class='qk-c'>,</span> Any<span class='qk-c'>]</span><span class='qk-c'>]</span></span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/providers/sqlite_provider.py#L280' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Get column information for SQLite table
</div>

<hr class='qk-rule' />

<div data-pagefind-weight='10'>

### `Schema`

</div>

<span class='qk-meta' data-api-type='Classes'></span>

<div class='qk-source-row'><a class='qk-source' href='https://github.com/dbtinoy-/quadkit/blob/main/core/quadkit-contracts/src/quadkit/contracts/data/identifiers.py#L263' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>
A validated SQL schema name.

<hr class='qk-rule' />

<div data-pagefind-weight='10'>

### `ScopeColumn`

</div>

<span class='qk-meta' data-api-type='Classes'></span>

<div class='qk-source-row'><a class='qk-source' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/row_level_security.py#L96' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>
Declares a single column that participates in the RLS predicate.



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tn'>`name`</td><td class='qk-tt'></td><td class='qk-td'>Column name in the database (e.g. ``"user_id"``).</td></tr></tbody></table></div>



<div class='qk-methods'>
<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>where_fragment</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>where_fragment</span><span class='qk-c'>(</span><span class='qk-n'>param_index</span><span class='qk-c'>: </span><span class='qk-t'>int</span><span class='qk-c'> = </span><span class='qk-d'>0</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'>tuple<span class='qk-c'>[</span>str<span class='qk-c'>,</span> str<span class='qk-c'>]</span></span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/row_level_security.py#L105' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Return ``(sql_fragment, placeholder_name)`` for this column.

The fragment uses a named placeholder so it can be appended as a
raw SQL ``WHERE`` clause segment.



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tn'>`param_index`</td><td class='qk-tt'>int</td><td class='qk-td'>Numeric suffix for the placeholder name to avoid collisions when multiple scope columns are present.</td></tr></tbody></table></div>




<div class='qk-table-wrap'><span class='qk-label'>Returns</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-45'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tt'>tuple<span class='qk-c'>[</span>str<span class='qk-c'>,</span> str<span class='qk-c'>]</span></td><td class='qk-td'>A 2-tuple of (``"col_name = ?"``), placeholder_key).</td></tr></tbody></table></div>


</div>

<hr class='qk-rule' />

<div data-pagefind-weight='10'>

### `SessionFactory`

</div>

<span class='qk-meta' data-api-type='Classes'></span>

<div class='qk-source-row'><a class='qk-source' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/orm/session.py#L52' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>
Owns one AsyncEngine and hands out ORM sessions.



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tn'>`url`</td><td class='qk-tt'></td><td class='qk-td'>SQLAlchemy database URL, e.g. ``postgresql+asyncpg://user:pw@host/db`` or ``sqlite+aiosqlite:///./app.db``.</td></tr><tr><td class='qk-tn'>`echo`</td><td class='qk-tt'></td><td class='qk-td'>Log every statement. Development only.</td></tr><tr><td class='qk-tn'>`pool_size`</td><td class='qk-tt'></td><td class='qk-td'>Connections kept open. Ignored by SQLite.</td></tr><tr><td class='qk-tn'>`max_overflow`</td><td class='qk-tt'></td><td class='qk-td'>Extra connections allowed beyond ``pool_size``.</td></tr><tr><td class='qk-tn'>`pool_recycle`</td><td class='qk-tt'></td><td class='qk-td'>Seconds before a pooled connection is replaced. Guards against a proxy or database closing idle connections.</td></tr><tr><td class='qk-tn'>`expire_on_commit`</td><td class='qk-tt'></td><td class='qk-td'>Left ``False`` by default so attributes stay readable after commit — the usual source of surprise ``MissingGreenlet`` errors in async code.</td></tr><tr><td class='qk-tn'>`sqlite_journal_mode`</td><td class='qk-tt'></td><td class='qk-td'>Journal mode applied to SQLite engines. ``TRUNCATE`` is the safe default; ``WAL`` is faster but needs sidecar files the filesystem may refuse.</td></tr><tr><td class='qk-tn'>`engine_kwargs`</td><td class='qk-tt'></td><td class='qk-td'>Passed straight to ``create_async_engine``.</td></tr></tbody></table></div>



**Example**

```python
factory = await container.resolve(SessionFactory)

async with factory.session() as session:
    rows = (await session.execute(select(User))).scalars().all()

async with factory.begin() as session:   # commits on success
    session.add(User(name="Mei"))
```

<div class='qk-methods'>
<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>__init__</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>__init__</span><span class='qk-c'>(</span>
    <span class='qk-n'>url</span><span class='qk-c'>: </span><span class='qk-t'>str</span><span class='qk-c'>,</span>
    <span class='qk-d'>*</span><span class='qk-n'></span><span class='qk-c'>,</span>
    <span class='qk-n'>echo</span><span class='qk-c'>: </span><span class='qk-t'>bool</span><span class='qk-c'> = </span><span class='qk-d'>False</span><span class='qk-c'>,</span>
    <span class='qk-n'>pool_size</span><span class='qk-c'>: </span><span class='qk-t'>int</span><span class='qk-c'> = </span><span class='qk-d'>5</span><span class='qk-c'>,</span>
    <span class='qk-n'>max_overflow</span><span class='qk-c'>: </span><span class='qk-t'>int</span><span class='qk-c'> = </span><span class='qk-d'>10</span><span class='qk-c'>,</span>
    <span class='qk-n'>pool_recycle</span><span class='qk-c'>: </span><span class='qk-t'>int</span><span class='qk-c'> = </span><span class='qk-d'>3600</span><span class='qk-c'>,</span>
    <span class='qk-n'>expire_on_commit</span><span class='qk-c'>: </span><span class='qk-t'>bool</span><span class='qk-c'> = </span><span class='qk-d'>False</span><span class='qk-c'>,</span>
    <span class='qk-n'>sqlite_journal_mode</span><span class='qk-c'>: </span><span class='qk-t'>str</span><span class='qk-c'> = </span><span class='qk-s'>'TRUNCATE'</span><span class='qk-c'>,</span>
    <span class='qk-d'>**</span><span class='qk-n'>engine_kwargs</span><span class='qk-c'>: </span><span class='qk-t'>Any</span>
<span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-d'>None</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/orm/session.py#L84' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>



<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>engine</span></div><pre class='qk-sig-pre'><span class='qk-k'>property </span><span class='qk-f'>engine</span><span class='qk-c'>(</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'>AsyncEngine</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/orm/session.py#L138' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

The underlying engine — for DDL, migrations and introspection.

<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>sessionmaker</span></div><pre class='qk-sig-pre'><span class='qk-k'>property </span><span class='qk-f'>sessionmaker</span><span class='qk-c'>(</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'>async_sessionmaker<span class='qk-c'>[</span>AsyncSession<span class='qk-c'>]</span></span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/orm/session.py#L145' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

The raw sessionmaker, for callers that manage their own scope.

<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>session</span></div><pre class='qk-sig-pre'><span class='qk-k'>async </span><span class='qk-k'>def </span><span class='qk-f'>session</span><span class='qk-c'>(</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'>AsyncIterator<span class='qk-c'>[</span>AsyncSession<span class='qk-c'>]</span></span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/orm/session.py#L160' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

A session with no surrounding transaction. Closes on exit.

<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>begin</span></div><pre class='qk-sig-pre'><span class='qk-k'>async </span><span class='qk-k'>def </span><span class='qk-f'>begin</span><span class='qk-c'>(</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'>AsyncIterator<span class='qk-c'>[</span>AsyncSession<span class='qk-c'>]</span></span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/orm/session.py#L166' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

A session inside a transaction: commits on success, rolls back on error.

<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>dispose</span></div><pre class='qk-sig-pre'><span class='qk-k'>async </span><span class='qk-k'>def </span><span class='qk-f'>dispose</span><span class='qk-c'>(</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-d'>None</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/orm/session.py#L173' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Close every pooled connection. Safe if the engine was never built.
</div>

<hr class='qk-rule' />

<div data-pagefind-weight='10'>

### `SimpleConnectionPool`

</div>

<span class='qk-meta' data-api-type='Classes'></span>

<div class='qk-source-row'><a class='qk-source' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/pool/connection.py#L334' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>
Simple connection pool that creates connections on demand

This is a basic implementation that doesn't use any external pooling libraries.



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tn'>`provider`</td><td class='qk-tt'></td><td class='qk-td'>The database provider object used to satisfy connections.</td></tr><tr><td class='qk-tn'>`use_provider_proxy`</td><td class='qk-tt'></td><td class='qk-td'>If True, returns a lightweight proxy object instead of the raw provider when creating connections. Defaults to False to preserve backwards compatibility.</td></tr></tbody></table></div>



<div class='qk-methods'>
<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>__init__</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>__init__</span><span class='qk-c'>(</span>
    <span class='qk-n'>provider</span><span class='qk-c'>: </span><span class='qk-t'><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#databaseproviderprotocol'>DatabaseProviderProtocol</a></span><span class='qk-c'>,</span>
    <span class='qk-n'>use_provider_proxy</span><span class='qk-c'>: </span><span class='qk-t'>bool</span><span class='qk-c'> = </span><span class='qk-d'>False</span><span class='qk-c'>,</span>
    <span class='qk-d'>**</span><span class='qk-n'>kwargs</span><span class='qk-c'>: </span><span class='qk-t'>Any</span>
<span class='qk-c'>)</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/pool/connection.py#L347' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>



<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>warm</span></div><pre class='qk-sig-pre'><span class='qk-k'>async </span><span class='qk-k'>def </span><span class='qk-f'>warm</span><span class='qk-c'>(</span><span class='qk-n'>count</span><span class='qk-c'>: </span><span class='qk-t'>int <span class='qk-c'><span class='qk-c'>|</span></span> <span class='qk-d'>None</span></span><span class='qk-c'> = </span><span class='qk-d'>None</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-d'>None</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/pool/connection.py#L406' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Pre-create connections to avoid cold-start latency.



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tn'>`count`</td><td class='qk-tt'>int <span class='qk-c'><span class='qk-c'>|</span></span> <span class='qk-d'>None</span></td><td class='qk-td'>Number of connections to open. Defaults to ``min_connections`` if not specified. Will not exceed ``max_connections``.</td></tr></tbody></table></div>



<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>validate_connections</span></div><pre class='qk-sig-pre'><span class='qk-k'>async </span><span class='qk-k'>def </span><span class='qk-f'>validate_connections</span><span class='qk-c'>(</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'>int</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/pool/connection.py#L423' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Validate all idle connections in the pool, evicting dead ones.

Iterates through all idle connections, discards those that are expired
(older than ``max_idle_time``) or fail validation, then refills the
pool to ``min_connections`` via ``warm()``.



<div class='qk-table-wrap'><span class='qk-label'>Returns</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-45'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tt'>int</td><td class='qk-td'>Number of valid connections remaining after validation.</td></tr></tbody></table></div>


</div>

<hr class='qk-rule' />

<div data-pagefind-weight='10'>

### `SimpleMigrationManager`

</div>

<span class='qk-meta' data-api-type='Classes'></span>

<div class='qk-source-row'><a class='qk-source' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/migrations/manager/_simple.py#L61' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>
Simple migration manager for community edition.

Provides basic migration tracking and execution without advanced
features like dependency resolution or rollback scripts.


**Example**

Basic usage

```python
manager = SimpleMigrationManager(provider=db_provider)
await manager.initialize_migration_table()
await manager.apply_migration("001", "create_users", CREATE_SQL)
```


    manager = SimpleMigrationManager(provider=db_provider)
    await manager.initialize_migration_table()
    await manager.apply_migration("001", "create_users", CREATE_SQL)

<div class='qk-methods'>
<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>create</span></div><pre class='qk-sig-pre'><span class='qk-k'>async </span><span class='qk-k'>def </span><span class='qk-f'>create</span><span class='qk-c'>(</span>
    <span class='qk-n'>name</span><span class='qk-c'>: </span><span class='qk-t'>str</span><span class='qk-c'>,</span>
    <span class='qk-n'>message</span><span class='qk-c'>: </span><span class='qk-t'>str</span>
<span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'>str</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/migrations/manager/_simple.py#L75' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Create a new migration file (alias for create_migration_file).



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tn'>`name`</td><td class='qk-tt'>str</td><td class='qk-td'>Migration name.</td></tr><tr><td class='qk-tn'>`message`</td><td class='qk-tt'>str</td><td class='qk-td'>Migration description or SQL content.</td></tr></tbody></table></div>




<div class='qk-table-wrap'><span class='qk-label'>Returns</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-45'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tt'>str</td><td class='qk-td'>The version string of the created migration.</td></tr></tbody></table></div>



<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>get_current_version</span></div><pre class='qk-sig-pre'><span class='qk-k'>async </span><span class='qk-k'>def </span><span class='qk-f'>get_current_version</span><span class='qk-c'>(</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'>str <span class='qk-c'><span class='qk-c'>|</span></span> <span class='qk-d'>None</span></span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/migrations/manager/_simple.py#L87' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Get the version of the most recently applied migration.



<div class='qk-table-wrap'><span class='qk-label'>Returns</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-45'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tt'>str <span class='qk-c'><span class='qk-c'>|</span></span> <span class='qk-d'>None</span></td><td class='qk-td'>The version string, or None if no migrations have been applied.</td></tr></tbody></table></div>



<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>__init__</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>__init__</span><span class='qk-c'>(</span>
    <span class='qk-n'>provider</span><span class='qk-c'>: </span><span class='qk-t'><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#databaseproviderprotocol'>DatabaseProviderProtocol</a> <span class='qk-c'><span class='qk-c'>|</span></span> <span class='qk-d'>None</span></span><span class='qk-c'> = </span><span class='qk-d'>None</span><span class='qk-c'>,</span>
    <span class='qk-n'>migrations_dir</span><span class='qk-c'>: </span><span class='qk-t'>str <span class='qk-c'><span class='qk-c'>|</span></span> <span class='qk-d'>None</span></span><span class='qk-c'> = </span><span class='qk-d'>None</span>
<span class='qk-c'>)</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/migrations/manager/_simple.py#L98' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Initialize the simple migration manager.



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tn'>`provider`</td><td class='qk-tt'><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#databaseproviderprotocol'>DatabaseProviderProtocol</a> <span class='qk-c'><span class='qk-c'>|</span></span> <span class='qk-d'>None</span></td><td class='qk-td'>Database provider instance.</td></tr><tr><td class='qk-tn'>`migrations_dir`</td><td class='qk-tt'>str <span class='qk-c'><span class='qk-c'>|</span></span> <span class='qk-d'>None</span></td><td class='qk-td'>Directory for migration files. Defaults to "./migrations".</td></tr></tbody></table></div>



<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>initialize_migration_table</span></div><pre class='qk-sig-pre'><span class='qk-k'>async </span><span class='qk-k'>def </span><span class='qk-f'>initialize_migration_table</span><span class='qk-c'>(</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-d'>None</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/migrations/manager/_simple.py#L113' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Initialize the migration tracking table.

Creates the __migrations table if it doesn't exist.

<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>get_applied_migrations</span></div><pre class='qk-sig-pre'><span class='qk-k'>async </span><span class='qk-k'>def </span><span class='qk-f'>get_applied_migrations</span><span class='qk-c'>(</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'>list<span class='qk-c'>[</span>MigrationRecord<span class='qk-c'>]</span></span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/migrations/manager/_simple.py#L133' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Get list of applied migrations.



<div class='qk-table-wrap'><span class='qk-label'>Returns</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-45'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tt'>list<span class='qk-c'>[</span>MigrationRecord<span class='qk-c'>]</span></td><td class='qk-td'>List of MigrationRecord for all applied migrations.</td></tr></tbody></table></div>



<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>apply_migration</span></div><pre class='qk-sig-pre'><span class='qk-k'>async </span><span class='qk-k'>def </span><span class='qk-f'>apply_migration</span><span class='qk-c'>(</span>
    <span class='qk-n'>version</span><span class='qk-c'>: </span><span class='qk-t'>str</span><span class='qk-c'>,</span>
    <span class='qk-n'>name</span><span class='qk-c'>: </span><span class='qk-t'>str</span><span class='qk-c'>,</span>
    <span class='qk-n'>sql</span><span class='qk-c'>: </span><span class='qk-t'>str</span>
<span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'>bool</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/migrations/manager/_simple.py#L160' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Apply a migration to the database.

<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>rollback_migration</span></div><pre class='qk-sig-pre'><span class='qk-k'>async </span><span class='qk-k'>def </span><span class='qk-f'>rollback_migration</span><span class='qk-c'>(</span><span class='qk-n'>version</span><span class='qk-c'>: </span><span class='qk-t'>str</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'>bool</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/migrations/manager/_simple.py#L183' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Rollback a migration by removing it from the tracking table.


**Note**

This only removes the migration record. Actual rollback
SQL must be executed separately.



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tn'>`version`</td><td class='qk-tt'>str</td><td class='qk-td'>Migration version to rollback.</td></tr></tbody></table></div>



<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>get_pending_migrations</span></div><pre class='qk-sig-pre'><span class='qk-k'>async </span><span class='qk-k'>def </span><span class='qk-f'>get_pending_migrations</span><span class='qk-c'>(</span><span class='qk-n'>available_migrations</span><span class='qk-c'>: </span><span class='qk-t'>list<span class='qk-c'>[</span>str<span class='qk-c'>]</span></span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'>list<span class='qk-c'>[</span>str<span class='qk-c'>]</span></span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/migrations/manager/_simple.py#L203' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Get migrations that haven't been applied yet.

<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>apply_pending_migrations</span></div><pre class='qk-sig-pre'><span class='qk-k'>async </span><span class='qk-k'>def </span><span class='qk-f'>apply_pending_migrations</span><span class='qk-c'>(</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'>list<span class='qk-c'>[</span>str<span class='qk-c'>]</span></span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/migrations/manager/_simple.py#L212' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Apply all pending migrations in the migrations directory.

<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>create_migration_file</span></div><pre class='qk-sig-pre'><span class='qk-k'>async </span><span class='qk-k'>def </span><span class='qk-f'>create_migration_file</span><span class='qk-c'>(</span>
    <span class='qk-n'>name</span><span class='qk-c'>: </span><span class='qk-t'>str</span><span class='qk-c'>,</span>
    <span class='qk-n'>sql</span><span class='qk-c'>: </span><span class='qk-t'>str</span>
<span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'>str</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/migrations/manager/_simple.py#L243' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Create a new migration file.
</div>

<hr class='qk-rule' />

<div data-pagefind-weight='10'>

### `SimpleTransactionManager`

</div>

<span class='qk-meta' data-api-type='Classes'></span>

<div class='qk-source-row'><a class='qk-source' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/unit_of_work/manager.py#L24' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>
Simple transaction manager for application-level use.

Not to be confused with ``providers.transaction_manager.TransactionManager``
which is the internal provider-level transaction handler.

<div class='qk-methods'>
<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>__init__</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>__init__</span><span class='qk-c'>(</span><span class='qk-n'>connection_pool</span><span class='qk-c'>: </span><span class='qk-t'><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#connectionpoolprotocol'>ConnectionPoolProtocol</a></span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-d'>None</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/unit_of_work/manager.py#L31' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>



<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>get_current_connection</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>get_current_connection</span><span class='qk-c'>(</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'>Any</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/unit_of_work/manager.py#L36' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Get the active transaction connection for the current async task.

<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>transaction</span></div><pre class='qk-sig-pre'><span class='qk-k'>async </span><span class='qk-k'>def </span><span class='qk-f'>transaction</span><span class='qk-c'>(</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'>AsyncGenerator<span class='qk-c'>[</span>Any<span class='qk-c'>,</span> <span class='qk-d'>None</span><span class='qk-c'>]</span></span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/unit_of_work/manager.py#L41' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Context manager for database transactions.
</div>

<hr class='qk-rule' />

<div data-pagefind-weight='10'>

### `SimpleUnitOfWork`

</div>

<span class='qk-meta' data-api-type='Classes'></span>

<div class='qk-source-row'><a class='qk-source' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/unit_of_work/simple/_uow.py#L27' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>
SQL-backed Unit of Work.

Extends AbstractUnitOfWork with full DB
transaction management. On commit the queued EntityOperation
items are executed via :data:`_operation_handler_registry` and the
provider transaction is committed. All entity-tracking and event-
publishing logic is inherited from the base class.



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tn'>`provider`</td><td class='qk-tt'></td><td class='qk-td'>The database provider used to execute SQL operations.</td></tr><tr><td class='qk-tn'>`event_bus`</td><td class='qk-tt'></td><td class='qk-td'>Optional event bus forwarded to the base class for domain event publishing on commit.</td></tr></tbody></table></div>



<div class='qk-methods'>
<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>__init__</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>__init__</span><span class='qk-c'>(</span>
    <span class='qk-n'>provider</span><span class='qk-c'>: </span><span class='qk-t'><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#databaseproviderprotocol'>DatabaseProviderProtocol</a></span><span class='qk-c'>,</span>
    <span class='qk-n'>event_bus</span><span class='qk-c'>: </span><span class='qk-t'>Any</span><span class='qk-c'> = </span><span class='qk-d'>None</span>
<span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-d'>None</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/unit_of_work/simple/_uow.py#L42' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>



<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>commit</span></div><pre class='qk-sig-pre'><span class='qk-k'>async </span><span class='qk-k'>def </span><span class='qk-f'>commit</span><span class='qk-c'>(</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-d'>None</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/unit_of_work/simple/_uow.py#L88' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Flush all queued operations and commit the DB transaction.



<div class='qk-table-wrap'><span class='qk-label'>Raises</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-45'>Exception</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tt'>RuntimeError</td><td class='qk-td'>If not currently in a transaction, or already committed.</td></tr><tr><td class='qk-tt'><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#databaseerror'>DatabaseError</a></td><td class='qk-td'>Re-raised after automatic rollback if the flush or commit_transaction call fails.</td></tr></tbody></table></div>



<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>rollback</span></div><pre class='qk-sig-pre'><span class='qk-k'>async </span><span class='qk-k'>def </span><span class='qk-f'>rollback</span><span class='qk-c'>(</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-d'>None</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/unit_of_work/simple/_uow.py#L109' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Roll back the current transaction and discard all tracked state.



<div class='qk-table-wrap'><span class='qk-label'>Raises</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-45'>Exception</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tt'>RuntimeError</td><td class='qk-td'>If not currently in a transaction.</td></tr></tbody></table></div>



<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>register_new</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>register_new</span><span class='qk-c'>(</span><span class='qk-n'>entity</span><span class='qk-c'>: </span><span class='qk-t'>Any</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-d'>None</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/unit_of_work/simple/_uow.py#L149' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Track *entity* for insertion and queue an insert operation.



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tn'>`entity`</td><td class='qk-tt'>Any</td><td class='qk-td'>The new entity to insert on commit.</td></tr></tbody></table></div>




<div class='qk-table-wrap'><span class='qk-label'>Raises</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-45'>Exception</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tt'>RuntimeError</td><td class='qk-td'>If the unit of work is already committed or rolled back.</td></tr></tbody></table></div>



<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>register_dirty</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>register_dirty</span><span class='qk-c'>(</span><span class='qk-n'>entity</span><span class='qk-c'>: </span><span class='qk-t'>Any</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-d'>None</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/unit_of_work/simple/_uow.py#L169' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Track *entity* as modified and queue an update operation.



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tn'>`entity`</td><td class='qk-tt'>Any</td><td class='qk-td'>The modified entity to update on commit.</td></tr></tbody></table></div>




<div class='qk-table-wrap'><span class='qk-label'>Raises</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-45'>Exception</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tt'>RuntimeError</td><td class='qk-td'>If the unit of work is already committed or rolled back.</td></tr></tbody></table></div>



<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>register_deleted</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>register_deleted</span><span class='qk-c'>(</span><span class='qk-n'>entity</span><span class='qk-c'>: </span><span class='qk-t'>Any</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-d'>None</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/unit_of_work/simple/_uow.py#L189' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Track *entity* for deletion and queue a delete operation.



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tn'>`entity`</td><td class='qk-tt'>Any</td><td class='qk-td'>The entity to delete on commit.</td></tr></tbody></table></div>




<div class='qk-table-wrap'><span class='qk-label'>Raises</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-45'>Exception</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tt'>RuntimeError</td><td class='qk-td'>If the unit of work is already committed or rolled back.</td></tr></tbody></table></div>



<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>register_event</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>register_event</span><span class='qk-c'>(</span><span class='qk-n'>event</span><span class='qk-c'>: </span><span class='qk-t'>Any</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-d'>None</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/unit_of_work/simple/_uow.py#L209' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Manually register a domain event for publication on commit.



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tn'>`event`</td><td class='qk-tt'>Any</td><td class='qk-td'>Any domain event (or plain object) to buffer.</td></tr></tbody></table></div>




<div class='qk-table-wrap'><span class='qk-label'>Raises</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-45'>Exception</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tt'>RuntimeError</td><td class='qk-td'>If the unit of work is already committed or rolled back.</td></tr></tbody></table></div>



<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>collect_events</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>collect_events</span><span class='qk-c'>(</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'>list<span class='qk-c'>[</span>Any<span class='qk-c'>]</span></span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/unit_of_work/simple/_uow.py#L222' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Return and clear all buffered events.

Events are buffered eagerly during register_new,
register_dirty, and register_deleted via
_collect_entity_events. This method drains the internal list.



<div class='qk-table-wrap'><span class='qk-label'>Returns</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-45'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tt'>list<span class='qk-c'>[</span>Any<span class='qk-c'>]</span></td><td class='qk-td'>All pending events; the internal buffer is cleared.</td></tr></tbody></table></div>



<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>reset</span></div><pre class='qk-sig-pre'><span class='qk-k'>async </span><span class='qk-k'>def </span><span class='qk-f'>reset</span><span class='qk-c'>(</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-d'>None</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/unit_of_work/simple/_uow.py#L314' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Reset unit of work state without interacting with the DB transaction.

Useful for reusing a UoW instance between logical operations within
a single session.

<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>savepoint</span></div><pre class='qk-sig-pre'><span class='qk-k'>async </span><span class='qk-k'>def </span><span class='qk-f'>savepoint</span><span class='qk-c'>(</span><span class='qk-n'>name</span><span class='qk-c'>: </span><span class='qk-t'>str <span class='qk-c'><span class='qk-c'>|</span></span> <span class='qk-d'>None</span></span><span class='qk-c'> = </span><span class='qk-d'>None</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'>str</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/unit_of_work/simple/_uow.py#L331' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Create a savepoint within the current transaction.



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tn'>`name`</td><td class='qk-tt'>str <span class='qk-c'><span class='qk-c'>|</span></span> <span class='qk-d'>None</span></td><td class='qk-td'>Optional savepoint name. Auto-generated if not provided.</td></tr></tbody></table></div>




<div class='qk-table-wrap'><span class='qk-label'>Returns</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-45'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tt'>str</td><td class='qk-td'>The savepoint name.</td></tr></tbody></table></div>




<div class='qk-table-wrap'><span class='qk-label'>Raises</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-45'>Exception</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tt'>RuntimeError</td><td class='qk-td'>If not currently in a transaction.</td></tr></tbody></table></div>



<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>rollback_to</span></div><pre class='qk-sig-pre'><span class='qk-k'>async </span><span class='qk-k'>def </span><span class='qk-f'>rollback_to</span><span class='qk-c'>(</span><span class='qk-n'>savepoint_name</span><span class='qk-c'>: </span><span class='qk-t'>str</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-d'>None</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/unit_of_work/simple/_uow.py#L352' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Roll back to a named savepoint.



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tn'>`savepoint_name`</td><td class='qk-tt'>str</td><td class='qk-td'>Name of the savepoint to restore.</td></tr></tbody></table></div>




<div class='qk-table-wrap'><span class='qk-label'>Raises</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-45'>Exception</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tt'>RuntimeError</td><td class='qk-td'>If *savepoint_name* is not registered.</td></tr></tbody></table></div>



<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>release_savepoint</span></div><pre class='qk-sig-pre'><span class='qk-k'>async </span><span class='qk-k'>def </span><span class='qk-f'>release_savepoint</span><span class='qk-c'>(</span><span class='qk-n'>savepoint_name</span><span class='qk-c'>: </span><span class='qk-t'>str</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-d'>None</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/unit_of_work/simple/_uow.py#L367' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Release (destroy) a savepoint.



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tn'>`savepoint_name`</td><td class='qk-tt'>str</td><td class='qk-td'>Name of the savepoint to release.</td></tr></tbody></table></div>




<div class='qk-table-wrap'><span class='qk-label'>Raises</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-45'>Exception</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tt'>RuntimeError</td><td class='qk-td'>If *savepoint_name* is not registered.</td></tr></tbody></table></div>



<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>before_commit</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>before_commit</span><span class='qk-c'>(</span><span class='qk-n'>hook</span><span class='qk-c'>: </span><span class='qk-t'>Callable</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-d'>None</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/unit_of_work/simple/_uow.py#L385' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Register a callable to run just before the SQL commit.



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tn'>`hook`</td><td class='qk-tt'>Callable</td><td class='qk-td'>A sync or async callable with no arguments.</td></tr></tbody></table></div>



<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>after_commit</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>after_commit</span><span class='qk-c'>(</span><span class='qk-n'>hook</span><span class='qk-c'>: </span><span class='qk-t'>Callable</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-d'>None</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/unit_of_work/simple/_uow.py#L393' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Register a callable to run just after the SQL commit.



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tn'>`hook`</td><td class='qk-tt'>Callable</td><td class='qk-td'>A sync or async callable with no arguments.</td></tr></tbody></table></div>


</div>

<hr class='qk-rule' />

<div data-pagefind-weight='10'>

### `Table`

</div>

<span class='qk-meta' data-api-type='Classes'></span>

<div class='qk-source-row'><a class='qk-source' href='https://github.com/dbtinoy-/quadkit/blob/main/core/quadkit-contracts/src/quadkit/contracts/data/identifiers.py#L243' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>
A validated SQL table name.

Example

```python
t = Table("users")
query = f"SELECT * FROM {t}"  # SELECT * FROM "users"
```


    t = Table("users")
    query = f"SELECT * FROM {t}"  # SELECT * FROM "users"

<hr class='qk-rule' />

<div data-pagefind-weight='10'>

### `UpdateResult`

</div>

<span class='qk-meta' data-api-type='Classes'></span>

<div class='qk-source-row'><a class='qk-source' href='https://github.com/dbtinoy-/quadkit/blob/main/core/quadkit-contracts/src/quadkit/contracts/data/sql/database/_results.py#L94' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>
Result of an update operation.

<hr class='qk-rule' />

## Functions


<div data-pagefind-weight='10'>

### `column`

</div>

<span class='qk-meta' data-api-type='Functions'></span>

<div class='qk-sig'><div class='qk-sig-head'><span class='qk-sig-label'>column</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>column</span><span class='qk-c'>(</span>
    <span class='qk-n'>name</span><span class='qk-c'>: </span><span class='qk-t'>str</span><span class='qk-c'>,</span>
    <span class='qk-d'>*</span><span class='qk-n'></span><span class='qk-c'>,</span>
    <span class='qk-n'>dialect</span><span class='qk-c'>: </span><span class='qk-t'><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#sqldialect'>SQLDialect</a></span><span class='qk-c'> = </span><span class='qk-d'>SQLDialect.POSTGRESQL</span>
<span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#column'>Column</a></span></pre></div>

<div class='qk-source-row'><a class='qk-source' href='https://github.com/dbtinoy-/quadkit/blob/main/core/quadkit-contracts/src/quadkit/contracts/data/identifiers.py#L391' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>
Create a cached, validated <a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#column'>Column</a> identifier.



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tn'>`name`</td><td class='qk-tt'>str</td><td class='qk-td'>Raw column name.</td></tr><tr><td class='qk-tn'>`dialect`</td><td class='qk-tt'><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#sqldialect'>SQLDialect</a></td><td class='qk-td'>SQL dialect for quoting.</td></tr></tbody></table></div>




<div class='qk-table-wrap'><span class='qk-label'>Returns</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-45'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tt'><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#column'>Column</a></td><td class='qk-td'>A validated, dialect-quoted <a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#column'>Column</a> identifier.</td></tr></tbody></table></div>



<hr class='qk-rule' />

<div data-pagefind-weight='10'>

### `create_db_context`

</div>

<span class='qk-meta' data-api-type='Functions'></span>

<div class='qk-sig'><div class='qk-sig-head'><span class='qk-sig-label'>create_db_context</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>create_db_context</span><span class='qk-c'>(</span>
    <span class='qk-d'>*</span><span class='qk-n'></span><span class='qk-c'>,</span>
    <span class='qk-n'>core_ctx</span><span class='qk-c'>: </span><span class='qk-t'>Context <span class='qk-c'><span class='qk-c'>|</span></span> <span class='qk-d'>None</span></span><span class='qk-c'> = </span><span class='qk-d'>None</span><span class='qk-c'>,</span>
    <span class='qk-n'>extra_keys</span><span class='qk-c'>: </span><span class='qk-t'>tuple<span class='qk-c'>[</span>ContextKey<span class='qk-c'>[</span>Any<span class='qk-c'>]</span><span class='qk-c'>,</span> Ellipsis<span class='qk-c'>]</span></span><span class='qk-c'> = </span><span class='qk-d'>()</span>
<span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#dbcontext'>DbContext</a></span></pre></div>

<div class='qk-source-row'><a class='qk-source' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/context/db_context.py#L184' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>
Create a ``DbContext`` with all standard db keys registered.

Call **once** in your composition root

```python
db_ctx = create_db_context(core_ctx=app_ctx)
handler = MyHandler(db_ctx=db_ctx)
```


    db_ctx = create_db_context(core_ctx=app_ctx)
    handler = MyHandler(db_ctx=db_ctx)



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tn'>`core_ctx`</td><td class='qk-tt'>Context <span class='qk-c'><span class='qk-c'>|</span></span> <span class='qk-d'>None</span></td><td class='qk-td'>Optional core ``Context`` instance.  When provided, ``RequestContextManager`` will bridge ``request_id``, ``user_id`` and ``tenant_id`` into it on enter and restore them on exit.</td></tr><tr><td class='qk-tn'>`extra_keys`</td><td class='qk-tt'>tuple<span class='qk-c'>[</span>ContextKey<span class='qk-c'>[</span>Any<span class='qk-c'>]</span><span class='qk-c'>,</span> Ellipsis<span class='qk-c'>]</span></td><td class='qk-td'>Additional ``ContextKey`` instances to register beyond the standard ``DB_KEYS``.</td></tr></tbody></table></div>



<hr class='qk-rule' />

<div data-pagefind-weight='10'>

### `create_task_with_context`

</div>

<span class='qk-meta' data-api-type='Functions'></span>

<div class='qk-sig'><div class='qk-sig-head'><span class='qk-sig-label'>create_task_with_context</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>create_task_with_context</span><span class='qk-c'>(</span>
    <span class='qk-n'>coro</span><span class='qk-c'>: </span><span class='qk-t'>Coroutine<span class='qk-c'>[</span>Any<span class='qk-c'>,</span> Any<span class='qk-c'>,</span> T<span class='qk-c'>]</span></span><span class='qk-c'>,</span>
    <span class='qk-d'>*</span><span class='qk-n'></span><span class='qk-c'>,</span>
    <span class='qk-n'>name</span><span class='qk-c'>: </span><span class='qk-t'>str <span class='qk-c'><span class='qk-c'>|</span></span> <span class='qk-d'>None</span></span><span class='qk-c'> = </span><span class='qk-d'>None</span>
<span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'>asyncio.Task<span class='qk-c'>[</span>T<span class='qk-c'>]</span></span></pre></div>

<div class='qk-source-row'><a class='qk-source' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/context/tasks.py#L20' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>
Create an ``asyncio.Task`` with the current context propagated.

In Python 3.11+ ``asyncio.create_task`` already copies context
automatically.  This wrapper exists for clarity and documentation.

<hr class='qk-rule' />

<div data-pagefind-weight='10'>

### `field`

</div>

<span class='qk-meta' data-api-type='Functions'></span>

<div class='qk-sig'><div class='qk-sig-head'><span class='qk-sig-label'>field</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>field</span><span class='qk-c'>(</span><span class='qk-n'>reference</span><span class='qk-c'>: </span><span class='qk-t'>Any</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#f'>F</a></span></pre></div>

<div class='qk-source-row'><a class='qk-source' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/repositories/filter_objects.py#L346' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>
Create an ``F`` builder from a field reference.

<hr class='qk-rule' />

<div data-pagefind-weight='10'>

### `full_text_search`

</div>

<span class='qk-meta' data-api-type='Functions'></span>

<div class='qk-sig'><div class='qk-sig-head'><span class='qk-sig-label'>full_text_search</span></div><pre class='qk-sig-pre'><span class='qk-k'>async </span><span class='qk-k'>def </span><span class='qk-f'>full_text_search</span><span class='qk-c'>(</span>
    <span class='qk-d'>*</span><span class='qk-n'></span><span class='qk-c'>,</span>
    <span class='qk-n'>provider</span><span class='qk-c'>: </span><span class='qk-t'><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#databaseproviderprotocol'>DatabaseProviderProtocol</a></span><span class='qk-c'>,</span>
    <span class='qk-n'>table</span><span class='qk-c'>: </span><span class='qk-t'>str</span><span class='qk-c'>,</span>
    <span class='qk-n'>columns</span><span class='qk-c'>: </span><span class='qk-t'>list<span class='qk-c'>[</span>str<span class='qk-c'>]</span></span><span class='qk-c'>,</span>
    <span class='qk-n'>query</span><span class='qk-c'>: </span><span class='qk-t'>str</span><span class='qk-c'>,</span>
    <span class='qk-n'>dialect</span><span class='qk-c'>: </span><span class='qk-t'><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#ftsdialect'>FTSDialect</a> <span class='qk-c'><span class='qk-c'>|</span></span> str</span><span class='qk-c'> = </span><span class='qk-d'>FTSDialect.POSTGRES</span><span class='qk-c'>,</span>
    <span class='qk-n'>entity_class</span><span class='qk-c'>: </span><span class='qk-t'>type<span class='qk-c'>[</span>TEntity<span class='qk-c'>]</span> <span class='qk-c'><span class='qk-c'>|</span></span> <span class='qk-d'>None</span></span><span class='qk-c'> = </span><span class='qk-d'>None</span><span class='qk-c'>,</span>
    <span class='qk-n'>extra_filters</span><span class='qk-c'>: </span><span class='qk-t'>dict<span class='qk-c'>[</span>str<span class='qk-c'>,</span> Any<span class='qk-c'>]</span> <span class='qk-c'><span class='qk-c'>|</span></span> <span class='qk-d'>None</span></span><span class='qk-c'> = </span><span class='qk-d'>None</span><span class='qk-c'>,</span>
    <span class='qk-n'>limit</span><span class='qk-c'>: </span><span class='qk-t'>int <span class='qk-c'><span class='qk-c'>|</span></span> <span class='qk-d'>None</span></span><span class='qk-c'> = </span><span class='qk-d'>50</span><span class='qk-c'>,</span>
    <span class='qk-n'>offset</span><span class='qk-c'>: </span><span class='qk-t'>int</span><span class='qk-c'> = </span><span class='qk-d'>0</span><span class='qk-c'>,</span>
    <span class='qk-n'>postgres_config</span><span class='qk-c'>: </span><span class='qk-t'>str</span><span class='qk-c'> = </span><span class='qk-s'>'english'</span><span class='qk-c'>,</span>
    <span class='qk-n'>postgres_plain</span><span class='qk-c'>: </span><span class='qk-t'>bool</span><span class='qk-c'> = </span><span class='qk-d'>True</span><span class='qk-c'>,</span>
    <span class='qk-n'>mysql_boolean_mode</span><span class='qk-c'>: </span><span class='qk-t'>bool</span><span class='qk-c'> = </span><span class='qk-d'>False</span>
<span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#ftsresult'>FTSResult</a></span></pre></div>

<div class='qk-source-row'><a class='qk-source' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/search/full_text.py#L261' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>
Dialect-agnostic full-text search helper.

Builds and executes the appropriate FTS query using *provider*.



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tn'>`provider`</td><td class='qk-tt'><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#databaseproviderprotocol'>DatabaseProviderProtocol</a></td><td class='qk-td'>The active database provider used to execute queries.</td></tr><tr><td class='qk-tn'>`table`</td><td class='qk-tt'>str</td><td class='qk-td'>Table name to search.</td></tr><tr><td class='qk-tn'>`columns`</td><td class='qk-tt'>list<span class='qk-c'>[</span>str<span class='qk-c'>]</span></td><td class='qk-td'>Columns to include in the full-text index / tsvector.</td></tr><tr><td class='qk-tn'>`query`</td><td class='qk-tt'>str</td><td class='qk-td'>User-supplied search string.</td></tr><tr><td class='qk-tn'>`dialect`</td><td class='qk-tt'><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#ftsdialect'>FTSDialect</a> <span class='qk-c'><span class='qk-c'>|</span></span> str</td><td class='qk-td'>Which SQL dialect to use (``"postgres"`` or ``"mysql"``).</td></tr><tr><td class='qk-tn'>`entity_class`</td><td class='qk-tt'>type<span class='qk-c'>[</span>TEntity<span class='qk-c'>]</span> <span class='qk-c'><span class='qk-c'>|</span></span> <span class='qk-d'>None</span></td><td class='qk-td'>Optional class to deserialise rows into.  When ``None`` raw ``dict`` rows are returned.</td></tr><tr><td class='qk-tn'>`extra_filters`</td><td class='qk-tt'>dict<span class='qk-c'>[</span>str<span class='qk-c'>,</span> Any<span class='qk-c'>]</span> <span class='qk-c'><span class='qk-c'>|</span></span> <span class='qk-d'>None</span></td><td class='qk-td'>Additional equality filters (``AND`` clauses).</td></tr><tr><td class='qk-tn'>`limit`</td><td class='qk-tt'>int <span class='qk-c'><span class='qk-c'>|</span></span> <span class='qk-d'>None</span></td><td class='qk-td'>Maximum rows to return.</td></tr><tr><td class='qk-tn'>`offset`</td><td class='qk-tt'>int</td><td class='qk-td'>Row offset for pagination.</td></tr><tr><td class='qk-tn'>`postgres_config`</td><td class='qk-tt'>str</td><td class='qk-td'>PostgreSQL text search configuration.</td></tr><tr><td class='qk-tn'>`postgres_plain`</td><td class='qk-tt'>bool</td><td class='qk-td'>Use ``plainto_tsquery`` when ``True``.</td></tr><tr><td class='qk-tn'>`mysql_boolean_mode`</td><td class='qk-tt'>bool</td><td class='qk-td'>Use MySQL boolean mode when ``True``.</td></tr></tbody></table></div>




<div class='qk-table-wrap'><span class='qk-label'>Returns</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-45'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tt'><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#ftsresult'>FTSResult</a></td><td class='qk-td'><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#ftsresult'>FTSResult</a> containing matched items (deserialised when *entity_class* is provided) and the result count.</td></tr></tbody></table></div>




<div class='qk-table-wrap'><span class='qk-label'>Raises</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-45'>Exception</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tt'>ValueError</td><td class='qk-td'>When an unsupported *dialect* is given.</td></tr></tbody></table></div>



<hr class='qk-rule' />

<div data-pagefind-weight='10'>

### `run_in_threadpool_with_context`

</div>

<span class='qk-meta' data-api-type='Functions'></span>

<div class='qk-sig'><div class='qk-sig-head'><span class='qk-sig-label'>run_in_threadpool_with_context</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>run_in_threadpool_with_context</span><span class='qk-c'>(</span>
    <span class='qk-n'>func</span><span class='qk-c'>: </span><span class='qk-t'>Callable<span class='qk-c'>[</span>Ellipsis<span class='qk-c'>,</span> T<span class='qk-c'>]</span></span><span class='qk-c'>,</span>
    <span class='qk-d'>*</span><span class='qk-n'>args</span><span class='qk-c'>: </span><span class='qk-t'>Any</span><span class='qk-c'>,</span>
    <span class='qk-d'>**</span><span class='qk-n'>kwargs</span><span class='qk-c'>: </span><span class='qk-t'>Any</span>
<span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'>asyncio.Future<span class='qk-c'>[</span>T<span class='qk-c'>]</span></span></pre></div>

<div class='qk-source-row'><a class='qk-source' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/context/tasks.py#L33' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>
Run a synchronous callable in the default thread-pool executor
with the current ``contextvars`` context snapshot propagated.

Uses ``get_running_loop()`` (not the deprecated
``get_event_loop()``).

<hr class='qk-rule' />

<div data-pagefind-weight='10'>

### `schema`

</div>

<span class='qk-meta' data-api-type='Functions'></span>

<div class='qk-sig'><div class='qk-sig-head'><span class='qk-sig-label'>schema</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>schema</span><span class='qk-c'>(</span>
    <span class='qk-n'>name</span><span class='qk-c'>: </span><span class='qk-t'>str</span><span class='qk-c'>,</span>
    <span class='qk-d'>*</span><span class='qk-n'></span><span class='qk-c'>,</span>
    <span class='qk-n'>dialect</span><span class='qk-c'>: </span><span class='qk-t'><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#sqldialect'>SQLDialect</a></span><span class='qk-c'> = </span><span class='qk-d'>SQLDialect.POSTGRESQL</span>
<span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#schema'>Schema</a></span></pre></div>

<div class='qk-source-row'><a class='qk-source' href='https://github.com/dbtinoy-/quadkit/blob/main/core/quadkit-contracts/src/quadkit/contracts/data/identifiers.py#L405' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>
Create a cached, validated <a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#schema'>Schema</a> identifier.



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tn'>`name`</td><td class='qk-tt'>str</td><td class='qk-td'>Raw schema name.</td></tr><tr><td class='qk-tn'>`dialect`</td><td class='qk-tt'><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#sqldialect'>SQLDialect</a></td><td class='qk-td'>SQL dialect for quoting.</td></tr></tbody></table></div>




<div class='qk-table-wrap'><span class='qk-label'>Returns</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-45'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tt'><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#schema'>Schema</a></td><td class='qk-td'>A validated, dialect-quoted <a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#schema'>Schema</a> identifier.</td></tr></tbody></table></div>



<hr class='qk-rule' />

<div data-pagefind-weight='10'>

### `table`

</div>

<span class='qk-meta' data-api-type='Functions'></span>

<div class='qk-sig'><div class='qk-sig-head'><span class='qk-sig-label'>table</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>table</span><span class='qk-c'>(</span>
    <span class='qk-n'>name</span><span class='qk-c'>: </span><span class='qk-t'>str</span><span class='qk-c'>,</span>
    <span class='qk-d'>*</span><span class='qk-n'></span><span class='qk-c'>,</span>
    <span class='qk-n'>dialect</span><span class='qk-c'>: </span><span class='qk-t'><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#sqldialect'>SQLDialect</a></span><span class='qk-c'> = </span><span class='qk-d'>SQLDialect.POSTGRESQL</span>
<span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#table'>Table</a></span></pre></div>

<div class='qk-source-row'><a class='qk-source' href='https://github.com/dbtinoy-/quadkit/blob/main/core/quadkit-contracts/src/quadkit/contracts/data/identifiers.py#L375' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>
Create a cached, validated <a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#table'>Table</a> identifier.

Repeated calls with the same arguments return the **same object**.



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tn'>`name`</td><td class='qk-tt'>str</td><td class='qk-td'>Raw table name.</td></tr><tr><td class='qk-tn'>`dialect`</td><td class='qk-tt'><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#sqldialect'>SQLDialect</a></td><td class='qk-td'>SQL dialect for quoting.</td></tr></tbody></table></div>




<div class='qk-table-wrap'><span class='qk-label'>Returns</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-45'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tt'><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#table'>Table</a></td><td class='qk-td'>A validated, dialect-quoted <a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#table'>Table</a> identifier.</td></tr></tbody></table></div>



<hr class='qk-rule' />

<div data-pagefind-weight='10'>

### `transaction`

</div>

<span class='qk-meta' data-api-type='Functions'></span>

<div class='qk-sig'><div class='qk-sig-head'><span class='qk-sig-label'>transaction</span></div><pre class='qk-sig-pre'><span class='qk-k'>async </span><span class='qk-k'>def </span><span class='qk-f'>transaction</span><span class='qk-c'>(</span><span class='qk-n'>manager</span><span class='qk-c'>: </span><span class='qk-t'><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#simpletransactionmanager'>SimpleTransactionManager</a></span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'>AsyncGenerator<span class='qk-c'>[</span><span class='qk-d'>None</span><span class='qk-c'>,</span> <span class='qk-d'>None</span><span class='qk-c'>]</span></span></pre></div>

<div class='qk-source-row'><a class='qk-source' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/unit_of_work/manager.py#L81' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>
Transaction context manager.



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tn'>`manager`</td><td class='qk-tt'><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#simpletransactionmanager'>SimpleTransactionManager</a></td><td class='qk-td'>SimpleTransactionManager instance from DI container.</td></tr></tbody></table></div>



<hr class='qk-rule' />

<div data-pagefind-weight='10'>

### `unit_of_work`

</div>

<span class='qk-meta' data-api-type='Functions'></span>

<div class='qk-sig'><div class='qk-sig-head'><span class='qk-sig-label'>unit_of_work</span></div><pre class='qk-sig-pre'><span class='qk-k'>async </span><span class='qk-k'>def </span><span class='qk-f'>unit_of_work</span><span class='qk-c'>(</span><span class='qk-n'>provider</span><span class='qk-c'>: </span><span class='qk-t'><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#databaseproviderprotocol'>DatabaseProviderProtocol</a></span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'>AsyncGenerator<span class='qk-c'>[</span><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#simpleunitofwork'>SimpleUnitOfWork</a><span class='qk-c'>,</span> <span class='qk-d'>None</span><span class='qk-c'>]</span></span></pre></div>

<div class='qk-source-row'><a class='qk-source' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/unit_of_work/simple/_uow.py#L408' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>
Async context manager that yields a committed <a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#simpleunitofwork'>SimpleUnitOfWork</a>.

Commits automatically on clean exit; rolls back on exception.



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tn'>`provider`</td><td class='qk-tt'><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#databaseproviderprotocol'>DatabaseProviderProtocol</a></td><td class='qk-td'>Database provider to use for the transaction.</td></tr></tbody></table></div>




<div class='qk-table-wrap'><span class='qk-label'>Yields</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-45'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tt'>AsyncGenerator<span class='qk-c'>[</span><a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#simpleunitofwork'>SimpleUnitOfWork</a><span class='qk-c'>,</span> <span class='qk-d'>None</span><span class='qk-c'>]</span></td><td class='qk-td'>A <a class='qk-typelink' href='/quadkit/packages/data/quadkit-sql/api/#simpleunitofwork'>SimpleUnitOfWork</a> instance within an open transaction.</td></tr></tbody></table></div>


Example

```python
async with unit_of_work(provider) as uow:
    uow.register_new(entity)
    # committed automatically on clean exit
```


    async with unit_of_work(provider) as uow:
        uow.register_new(entity)
        # committed automatically on clean exit

<hr class='qk-rule' />

## Exceptions


<div data-pagefind-weight='10'>

### `DatabaseError`

</div>

<span class='qk-meta' data-api-type='Exceptions'></span>

<div class='qk-source-row'><a class='qk-source' href='https://github.com/dbtinoy-/quadkit/blob/main/core/quadkit-contracts/src/quadkit/contracts/exceptions/infra.py#L30' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>
Database-related error (infrastructure-level).

<div class='qk-methods'>
<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>__init__</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>__init__</span><span class='qk-c'>(</span>
    <span class='qk-n'>message</span><span class='qk-c'>: </span><span class='qk-t'>str</span><span class='qk-c'> = </span><span class='qk-s'>'Database error'</span><span class='qk-c'>,</span>
    <span class='qk-d'>**</span><span class='qk-n'>kwargs</span><span class='qk-c'>: </span><span class='qk-t'>Any</span>
<span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-d'>None</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/core/quadkit-contracts/src/quadkit/contracts/exceptions/infra.py#L35' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>


</div>

<hr class='qk-rule' />

<div data-pagefind-weight='10'>

### `InvalidIdentifierError`

</div>

<span class='qk-meta' data-api-type='Exceptions'></span>

<div class='qk-source-row'><a class='qk-source' href='https://github.com/dbtinoy-/quadkit/blob/main/core/quadkit-contracts/src/quadkit/contracts/data/sql/sql.py#L14' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>
Raised when an SQL identifier fails validation.

Attributes:
    identifier: The offending identifier string.

<div class='qk-methods'>
<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>__init__</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>__init__</span><span class='qk-c'>(</span>
    <span class='qk-n'>message</span><span class='qk-c'>: </span><span class='qk-t'>str</span><span class='qk-c'>,</span>
    <span class='qk-n'>identifier</span><span class='qk-c'>: </span><span class='qk-t'>str <span class='qk-c'><span class='qk-c'>|</span></span> <span class='qk-d'>None</span></span><span class='qk-c'> = </span><span class='qk-d'>None</span><span class='qk-c'>,</span>
    <span class='qk-d'>**</span><span class='qk-n'>kwargs</span><span class='qk-c'>: </span><span class='qk-t'>Any</span>
<span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-d'>None</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/core/quadkit-contracts/src/quadkit/contracts/data/sql/sql.py#L23' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>


</div>

<hr class='qk-rule' />

<div data-pagefind-weight='10'>

### `NoSecurityPolicyError`

</div>

<span class='qk-meta' data-api-type='Exceptions'></span>

<div class='qk-source-row'><a class='qk-source' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/row_level_security.py#L70' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>
Raised when a repository with an RLS policy is queried without an
active scope *and* no admin bypass is in effect.

Attributes:
    table: The table that was queried.
    operation: The SQL operation attempted (``SELECT``, ``UPDATE``, etc.).

<div class='qk-methods'>
<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>__init__</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>__init__</span><span class='qk-c'>(</span>
    <span class='qk-n'>table</span><span class='qk-c'>: </span><span class='qk-t'>str</span><span class='qk-c'>,</span>
    <span class='qk-n'>operation</span><span class='qk-c'>: </span><span class='qk-t'>str</span>
<span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-d'>None</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/row_level_security.py#L79' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>


</div>

<hr class='qk-rule' />

<div data-pagefind-weight='10'>

### `OptimisticLockError`

</div>

<span class='qk-meta' data-api-type='Exceptions'></span>

<div class='qk-source-row'><a class='qk-source' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/exceptions.py#L211' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>
Raised when an optimistic lock check fails on UPDATE.

Indicates a concurrent modification was detected: the row's version
column did not match the expected value, so the UPDATE affected 0 rows.



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tn'>`entity_type`</td><td class='qk-tt'></td><td class='qk-td'>Class name of the entity that failed the version check.</td></tr><tr><td class='qk-tn'>`entity_id`</td><td class='qk-tt'></td><td class='qk-td'>Primary key of the entity.</td></tr><tr><td class='qk-tn'>`expected_version`</td><td class='qk-tt'></td><td class='qk-td'>The version the caller expected to update from.</td></tr></tbody></table></div>



<div class='qk-methods'>
<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>__init__</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>__init__</span><span class='qk-c'>(</span>
    <span class='qk-n'>entity_type</span><span class='qk-c'>: </span><span class='qk-t'>str</span><span class='qk-c'>,</span>
    <span class='qk-n'>entity_id</span><span class='qk-c'>: </span><span class='qk-t'>str <span class='qk-c'><span class='qk-c'>|</span></span> int</span><span class='qk-c'>,</span>
    <span class='qk-n'>expected_version</span><span class='qk-c'>: </span><span class='qk-t'>int</span>
<span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-d'>None</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/exceptions.py#L225' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>


</div>

<hr class='qk-rule' />

<div data-pagefind-weight='10'>

### `RepositoryError`

</div>

<span class='qk-meta' data-api-type='Exceptions'></span>

<div class='qk-source-row'><a class='qk-source' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-sql/src/quadkit/sql/exceptions.py#L303' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>
RepositoryProtocol-level operation failure.

<hr class='qk-rule' />