Core

The core module contains the fundamental components for SQL processing, including statement handling, parameter binding, result processing, compilation, and caching.

Overview

Core components:

• SQL Statement (statement.py) - SQL wrapper with metadata and parameter binding

• Parameters (parameters.py) - Parameter style conversion and binding

• Results (result.py) - Result set handling and type mapping

• Compiler (compiler.py) - SQL compilation and validation using sqlglot

• Cache (cache.py) - Statement caching for performance

• Filters (filters.py) - SQL transformation filters

SQL Statement

class sqlspec.core.statement.SQL

Bases: object

SQL statement with parameter and filter support.

Represents a SQL statement that can be compiled with parameters and filters. Supports both positional and named parameters, statement filtering, and various execution modes including batch operations.

The main SQL statement class that wraps raw SQL with parameters.

Features:

• Parameter binding (positional and named)

• SQL compilation and validation

• Statement metadata

• Cache integration

• Filter application

Usage:

from sqlspec.core import SQL

# Simple SQL
stmt = SQL("SELECT * FROM users")

# With positional parameters
stmt = SQL("SELECT * FROM users WHERE id = ?", 123)

# With named parameters
stmt = SQL("SELECT * FROM users WHERE id = :user_id", user_id=123)

# With keyword parameters
stmt = SQL("SELECT * FROM users WHERE id = :id", id=123)

# Access SQL and parameters
print(stmt.sql)         # "SELECT * FROM users WHERE id = :id"
print(stmt.parameters)  # {"id": 123}

__init__(statement, *parameters, statement_config=None, is_many=None, **kwargs)

Initialize SQL statement.

Parameters:

• statement – SQL string, expression, or existing SQL object

• *parameters – Parameters and filters

• statement_config – Configuration

• is_many – Mark as execute_many operation

• **kwargs – Additional parameters

property sql: str

Get the raw SQL string.

property raw_sql: str

Get raw SQL string (public API).

Returns:

The raw SQL string

property parameters: Any

Get the original parameters.

property positional_parameters: list[TypeAliasForwardRef('typing.Any')]

Get positional parameters (public API).

property named_parameters: dict[str, TypeAliasForwardRef('typing.Any')]

Get named parameters (public API).

property original_parameters: Any

Get original parameters (public API).

property operation_type: Literal['SELECT', 'INSERT', 'UPDATE', 'DELETE', 'COPY', 'COPY_FROM', 'COPY_TO', 'EXECUTE', 'SCRIPT', 'DDL', 'PRAGMA', 'MERGE', 'UNKNOWN']

SQL operation type.

property statement_config: StatementConfig

Statement configuration.

property expression: Expression | None

SQLGlot expression.

property filters: list[StatementFilter]

Applied filters.

get_filters_view()

Get zero-copy filters view (public API).

Return type:

FiltersView

Returns:

Read-only view of filters without copying

property is_processed: bool

Check if SQL has been processed (public API).

get_processed_state()

Get processed state (public API).

Return type:

Any

property dialect: str | None

SQL dialect.

property statement_expression: Expression | None

Get parsed statement expression (public API).

Returns:

Parsed SQLGlot expression or None if not parsed

property is_many: bool

Check if this is execute_many.

property is_script: bool

Check if this is script execution.

property validation_errors: list[str]

Validation errors.

property has_errors: bool

Check if there are validation errors.

returns_rows()

Check if statement returns rows.

Return type:

bool

Returns:

True if the SQL statement returns result rows

is_modifying_operation()

Check if the SQL statement is a modifying operation.

Return type:

bool

Returns:

True if the operation modifies data (INSERT/UPDATE/DELETE)

compile()

Compile SQL statement with parameters.

Return type:

tuple[str, Any]

Returns:

Tuple of compiled SQL string and execution parameters

as_script()

Create copy marked for script execution.

Return type:

SQL

Returns:

New SQL instance configured for script execution

copy(statement=None, parameters=None, **kwargs)

Create copy with modifications.

Parameters:

• statement (str | Expression | None) – New SQL statement to use

• parameters (Any | None) – New parameters to use

• **kwargs (Any) – Additional modifications

Return type:

SQL

Returns:

New SQL instance with modifications applied

add_named_parameter(name, value)

Add a named parameter and return a new SQL instance.

Parameters:

• name (str) – Parameter name

• value (Any) – Parameter value

Return type:

SQL

Returns:

New SQL instance with the added parameter

where(condition)

Add WHERE condition to the SQL statement.

Parameters:

condition (str | Expression) – WHERE condition as string or SQLGlot expression

Return type:

SQL

Returns:

New SQL instance with the WHERE condition applied

__hash__()

Hash value computation.

Return type:

int

__eq__(other)

Equality comparison.

Return type:

bool

__repr__()

String representation.

Return type:

str

class sqlspec.core.statement.StatementConfig

Bases: object

Configuration for SQL statement processing.

Controls SQL parsing, validation, transformations, parameter handling, and other processing options for SQL statements.

Configuration for SQL statement processing.

Options:

• enable_validation - Validate SQL syntax

• enable_analysis - Analyze SQL for optimization hints

• enable_transformations - Apply SQL transformations

• cache_statements - Cache compiled statements

__init__(parameter_config=None, enable_parsing=True, enable_validation=True, enable_transformations=True, enable_analysis=False, enable_expression_simplification=False, enable_parameter_type_wrapping=True, enable_caching=True, parameter_converter=None, parameter_validator=None, dialect=None, pre_process_steps=None, post_process_steps=None, execution_mode=None, execution_args=None, output_transformer=None)

Initialize StatementConfig.

Parameters:

• parameter_config – Parameter style configuration

• enable_parsing – Enable SQL parsing

• enable_validation – Run SQL validators

• enable_transformations – Apply SQL transformers

• enable_analysis – Run SQL analyzers

• enable_expression_simplification – Apply expression simplification

• enable_parameter_type_wrapping – Wrap parameters with type information

• enable_caching – Cache processed SQL statements

• parameter_converter – Handles parameter style conversions

• parameter_validator – Validates parameter usage and styles

• dialect – SQL dialect

• pre_process_steps – Optional list of preprocessing steps

• post_process_steps – Optional list of postprocessing steps

• execution_mode – Special execution mode

• execution_args – Arguments for special execution modes

• output_transformer – Optional output transformation function

enable_parsing

enable_validation

enable_transformations

enable_analysis

enable_expression_simplification

enable_parameter_type_wrapping

enable_caching

parameter_converter

parameter_validator

parameter_config

dialect

pre_process_steps

post_process_steps

execution_mode

execution_args

output_transformer

replace(**kwargs)

Immutable update pattern.

Parameters:

**kwargs (Any) – Attributes to update

Return type:

StatementConfig

Returns:

New StatementConfig instance with updated attributes

__hash__()

Hash based on configuration settings.

Return type:

int

__repr__()

String representation of the StatementConfig instance.

Return type:

str

__eq__(other)

Equality comparison.

Return type:

bool

Parameter Handling

class sqlspec.core.parameters.ParameterProcessor

Bases: object

Parameter processing engine coordinating conversion phases.

Processes SQL parameters by converting between different parameter styles.

Supported parameter styles:

• ParameterStyle.QMARK - ? (SQLite/DuckDB positional)

• ParameterStyle.NUMERIC - $1, $2 (PostgreSQL positional for asyncpg, psqlpy)

• ParameterStyle.POSITIONAL_PYFORMAT - %s (PostgreSQL/MySQL format style)

• ParameterStyle.NAMED_COLON - :name (Named parameters for Oracle, SQLite)

• ParameterStyle.NAMED_AT - @name (BigQuery named parameters)

• ParameterStyle.NAMED_PYFORMAT - %(name)s (Python format style)

DEFAULT_CACHE_SIZE = 1000

__init__()

process(sql, parameters, config, dialect=None, is_many=False)

Return type:

ParameterProcessingResult

get_sqlglot_compatible_sql(sql, parameters, config, dialect=None)

Normalize SQL for parsing without altering execution format.

Parameters:

• sql (str) – Raw SQL text.

• parameters (Any) – Parameter payload supplied by the caller.

• config (ParameterStyleConfig) – Parameter style configuration.

• dialect (str | None) – Optional SQL dialect for compatibility checks.

Return type:

tuple[str, typing.Any]

Returns:

Tuple of normalized SQL and the original parameter payload.

class sqlspec.core.parameters.ParameterConverter

Bases: object

Parameter style conversion helper.

Converts parameters between different styles.

__init__()

validator

normalize_sql_for_parsing(sql, dialect=None)

Return type:

tuple[str, list[ParameterInfo]]

convert_placeholder_style(sql, parameters, target_style, is_many=False)

Return type:

tuple[str, Any]

class sqlspec.core.parameters.ParameterValidator

Bases: object

Extracts placeholder metadata and dialect compatibility information.

Validates parameter format and style.

__init__(cache_max_size=5000)

extract_parameters(sql)

Extract ordered parameter metadata from SQL text.

Return type:

list[ParameterInfo]

get_sqlglot_incompatible_styles(dialect=None)

Return placeholder styles incompatible with SQLGlot for the dialect.

Return type:

set[ParameterStyle]

class sqlspec.core.parameters.ParameterStyleConfig

Bases: object

Configuration describing parameter behaviour for a statement.

Configuration for parameter style handling.

__init__(default_parameter_style, supported_parameter_styles=None, supported_execution_parameter_styles=None, default_execution_parameter_style=None, type_coercion_map=None, has_native_list_expansion=False, needs_static_script_compilation=False, allow_mixed_parameter_styles=False, preserve_parameter_format=True, preserve_original_params_for_many=False, output_transformer=None, ast_transformer=None, json_serializer=None, json_deserializer=None)

default_parameter_style

supported_parameter_styles

supported_execution_parameter_styles

default_execution_parameter_style

type_coercion_map

has_native_list_expansion

output_transformer

ast_transformer

needs_static_script_compilation

allow_mixed_parameter_styles

preserve_parameter_format

preserve_original_params_for_many

json_serializer

json_deserializer

hash()

Return the hash value for caching compatibility.

Return type:

int

Returns:

Hash value matching hash() output for this config.

replace(**overrides)

Return type:

ParameterStyleConfig

with_json_serializers(serializer, *, tuple_strategy='list', deserializer=None)

Return a copy configured with JSON serializers for complex parameters.

class sqlspec.core.parameters.ParameterStyle

Bases: str, Enum

Enumeration of supported SQL parameter placeholder styles.

Enum defining supported parameter styles.

NONE = 'none'

STATIC = 'static'

QMARK = 'qmark'

NUMERIC = 'numeric'

NAMED_COLON = 'named_colon'

POSITIONAL_COLON = 'positional_colon'

NAMED_AT = 'named_at'

NAMED_DOLLAR = 'named_dollar'

NAMED_PYFORMAT = 'pyformat_named'

POSITIONAL_PYFORMAT = 'pyformat_positional'

__new__(value)

Result Processing

class sqlspec.core.result.SQLResult

Bases: StatementResult

Result class for SQL operations that return rows or affect rows.

Handles SELECT, INSERT, UPDATE, DELETE operations. For DML operations with RETURNING clauses, the returned data is stored in the data attribute. The operation_type attribute indicates the nature of the operation.

For script execution, tracks multiple statement results and errors.

Result container for executed SQL queries.

Attributes:

• data - List of result rows (dicts) - prefer using helper methods instead

• rows_affected - Number of rows affected by INSERT/UPDATE/DELETE

• columns - Column names

• metadata - Query metadata

Recommended Helper Methods:

result = await session.execute("SELECT * FROM users")

# Get all rows (replaces result.data)
all_rows = result.all()

# Get exactly one row (raises if not exactly one)
user = result.one()

# Get one row or None (raises if multiple)
user = result.one_or_none()

# Get first row without validation
first_row = result.get_first()

# Get scalar value (first column of first row)
count = result.scalar()

# Map to typed models
from pydantic import BaseModel

class User(BaseModel):
    id: int
    name: str
    email: str

# Get all rows as typed models
users: list[User] = result.all(schema_type=User)

# Get exactly one row as typed model
user: User = result.one(schema_type=User)

# Get one or none as typed model
user: User | None = result.one_or_none(schema_type=User)

__init__(statement, data=None, rows_affected=0, last_inserted_id=None, execution_time=None, metadata=None, error=None, operation_type='SELECT', operation_index=None, parameters=None, column_names=None, total_count=None, has_more=False, inserted_ids=None, statement_results=None, errors=None, total_statements=0, successful_statements=0)

Initialize SQL result.

Parameters:

• statement (SQL) – The original SQL statement that was executed.

• data (list[dict[str, Any]] | None) – The result data from the operation.

• rows_affected (int) – Number of rows affected by the operation.

• last_inserted_id (int | str | None) – Last inserted ID from the operation.

• execution_time (float | None) – Time taken to execute the statement in seconds.

• metadata (Optional[dict[str, typing.Any]]) – Additional metadata about the operation.

• error (Exception | None) – Exception that occurred during execution.

• operation_type (Literal['SELECT', 'INSERT', 'UPDATE', 'DELETE', 'COPY', 'COPY_FROM', 'COPY_TO', 'EXECUTE', 'SCRIPT', 'DDL', 'PRAGMA', 'MERGE', 'UNKNOWN']) – Type of SQL operation performed.

• operation_index (int | None) – Index of operation in a script.

• parameters (Any | None) – Parameters used for the query.

• column_names (Optional[list[str]]) – Names of columns in the result set.

• total_count (int | None) – Total number of rows in the complete result set.

• has_more (bool) – Whether there are additional result pages available.

• inserted_ids (Optional[list[int | str]]) – List of IDs from INSERT operations.

• statement_results (Optional[list[SQLResult]]) – Results from individual statements in a script.

• errors (Optional[list[str]]) – List of error messages for script execution.

• total_statements (int) – Total number of statements in a script.

• successful_statements (int) – Count of successful statements in a script.

error

operation_index

parameters

has_more

inserted_ids

statement_results

errors

total_statements

successful_statements

column_names

total_count

property operation_type: Literal['SELECT', 'INSERT', 'UPDATE', 'DELETE', 'COPY', 'COPY_FROM', 'COPY_TO', 'EXECUTE', 'SCRIPT', 'DDL', 'PRAGMA', 'MERGE', 'UNKNOWN']

Get operation type for this result.

Returns:

The type of SQL operation that produced this result.

get_metadata(key, default=None)

Get metadata value by key.

Parameters:

• key (str) – The metadata key to retrieve.

• default (Any) – Default value if key is not found.

Return type:

Any

Returns:

The metadata value or default.

set_metadata(key, value)

Set metadata value by key.

Parameters:

• key (str) – The metadata key to set.

• value (Any) – The value to set.

Return type:

None

is_success()

Check if the operation was successful.

Return type:

bool

Returns:

True if operation was successful, False otherwise.

get_data(*, schema_type=None)

Get the data from the result.

For regular operations, returns the list of rows. For script operations, returns a summary dictionary containing execution statistics and results.

Parameters:

schema_type (type[TypeVar(SchemaT)] | None) – Optional schema type to transform the data into. Supports Pydantic models, dataclasses, msgspec structs, attrs classes, and TypedDict.

Return type:

list[TypeVar(SchemaT)] | list[dict[str, typing.Any]]

Returns:

List of result rows (optionally transformed to schema_type) or script summary.

add_statement_result(result)

Add a statement result to the script execution results.

Parameters:

result (SQLResult) – Statement result to add.

Return type:

None

get_total_rows_affected()

Get the total number of rows affected across all statements.

Return type:

int

Returns:

Total rows affected.

property num_rows: int

Get the number of rows affected (alias for get_total_rows_affected).

Returns:

Total rows affected.

property num_columns: int

Get the number of columns in the result data.

Returns:

Number of columns.

get_first(*, schema_type=None)

Get the first row from the result, if any.

Parameters:

schema_type (type[TypeVar(SchemaT)] | None) – Optional schema type to transform the data into. Supports Pydantic models, dataclasses, msgspec structs, attrs classes, and TypedDict.

Return type:

Union[TypeVar(SchemaT), dict[str, typing.Any], None]

Returns:

First row (optionally transformed to schema_type) or None if no data.

get_count()

Get the number of rows in the current result set (e.g., a page of data).

Return type:

int

Returns:

Number of rows in current result set.

is_empty()

Check if the result set (self.data) is empty.

Return type:

bool

Returns:

True if result set is empty.

get_affected_count()

Get the number of rows affected by a DML operation.

Return type:

int

Returns:

Number of affected rows.

was_inserted()

Check if this was an INSERT operation.

Return type:

bool

Returns:

True if INSERT operation.

was_updated()

Check if this was an UPDATE operation.

Return type:

bool

Returns:

True if UPDATE operation.

was_deleted()

Check if this was a DELETE operation.

Return type:

bool

Returns:

True if DELETE operation.

__len__()

Get the number of rows in the result set.

Return type:

int

Returns:

Number of rows in the data.

__getitem__(index)

Get a row by index.

Parameters:

index (int) – Row index

Return type:

dict[str, typing.Any]

Returns:

The row at the specified index

__iter__()

Iterate over the rows in the result.

Return type:

Iterator[dict[str, typing.Any]]

Returns:

Iterator that yields each row as a dictionary

all(*, schema_type=None)

Return all rows as a list.

Parameters:

schema_type (type[TypeVar(SchemaT)] | None) – Optional schema type to transform the data into. Supports Pydantic models, dataclasses, msgspec structs, attrs classes, and TypedDict.

Return type:

list[TypeVar(SchemaT)] | list[dict[str, typing.Any]]

Returns:

List of all rows (optionally transformed to schema_type)

one(*, schema_type=None)

Return exactly one row.

Parameters:

schema_type (type[TypeVar(SchemaT)] | None) – Optional schema type to transform the data into. Supports Pydantic models, dataclasses, msgspec structs, attrs classes, and TypedDict.

Return type:

Union[TypeVar(SchemaT), dict[str, typing.Any]]

Returns:

The single row (optionally transformed to schema_type)

Raises:

ValueError – If no results or more than one result

one_or_none(*, schema_type=None)

Return at most one row.

Parameters:

schema_type (type[TypeVar(SchemaT)] | None) – Optional schema type to transform the data into. Supports Pydantic models, dataclasses, msgspec structs, attrs classes, and TypedDict.

Return type:

Union[TypeVar(SchemaT), dict[str, typing.Any], None]

Returns:

The single row (optionally transformed to schema_type) or None if no results

Raises:

ValueError – If more than one result

scalar()

Return the first column of the first row.

Return type:

Any

Returns:

The scalar value from first column of first row

scalar_or_none()

Return the first column of the first row, or None if no results.

Return type:

Any

Returns:

The scalar value from first column of first row, or None

write_to_storage_sync(destination, *, format_hint=None, storage_options=None, pipeline=None)

Return type:

StorageTelemetry

async write_to_storage_async(destination, *, format_hint=None, storage_options=None, pipeline=None)

Return type:

StorageTelemetry

Type Mapping:

SQLResult supports mapping to various Python type systems:

• Pydantic models

• msgspec.Struct

• attrs classes

• dataclasses

• TypedDict

• Plain dicts

Type mapping is handled directly by methods on the SQLResult class such as as_type(), as_type_one(), and as_type_one_or_none().

SQL Compilation

class sqlspec.core.compiler.SQLProcessor

Bases: object

SQL processor with compilation and caching.

Processes SQL statements by compiling them into executable format with parameter substitution. Includes LRU-style caching for compilation results to avoid re-processing identical statements.

SQL processor with compilation and caching.

Processes SQL statements by compiling them into executable format with parameter substitution. Includes LRU-style caching for compilation results to avoid re-processing identical statements.

Usage:

from sqlspec.core import StatementConfig
from sqlspec.core import SQLProcessor

config = StatementConfig(dialect="postgres")
processor = SQLProcessor(config)

# Compile SQL with parameters
compiled = processor.compile(
    "SELECT * FROM users WHERE active = ?",
    parameters=[True]
)

__init__(config, max_cache_size=1000)

Initialize processor.

Parameters:

• config (StatementConfig) – Statement configuration

• max_cache_size (int) – Maximum number of compilation results to cache

compile(sql, parameters=None, is_many=False)

Compile SQL statement.

Parameters:

• sql (str) – SQL string for compilation

• parameters (Any) – Parameter values for substitution

• is_many (bool) – Whether this is for execute_many operation

Return type:

CompiledSQL

Returns:

CompiledSQL with execution information

clear_cache()

Clear compilation cache and reset statistics.

Return type:

None

property cache_stats: dict[str, int]

Get cache statistics.

Returns:

Dictionary with cache statistics

class sqlspec.core.compiler.CompiledSQL

Bases: object

Compiled SQL result.

Contains the result of SQL compilation with information needed for execution. Immutable container holding compiled SQL text, processed parameters, operation type, and execution metadata.

Compiled SQL result containing the compiled SQL text, processed parameters, operation type, and execution metadata.

__init__(compiled_sql, execution_parameters, operation_type, expression=None, parameter_style=None, supports_many=False, parameter_casts=None, parameter_profile=None, operation_profile=None)

Initialize compiled result.

Parameters:

• compiled_sql (str) – SQL string ready for execution

• execution_parameters (Any) – Parameters in driver-specific format

• operation_type (Literal['SELECT', 'INSERT', 'UPDATE', 'DELETE', 'COPY', 'COPY_FROM', 'COPY_TO', 'EXECUTE', 'SCRIPT', 'DDL', 'PRAGMA', 'MERGE', 'UNKNOWN']) – SQL operation type (SELECT, INSERT, etc.)

• expression (Optional[Expression]) – SQLGlot AST expression

• parameter_style (str | None) – Parameter style used in compilation

• supports_many (bool) – Whether this supports execute_many operations

• parameter_casts (Optional[dict[int, str]]) – Mapping of parameter positions to cast types

• parameter_profile (ParameterProfile | None) – Profile describing detected placeholders

• operation_profile (OperationProfile | None) – Profile describing semantic characteristics

compiled_sql

execution_parameters

operation_type: Literal['SELECT', 'INSERT', 'UPDATE', 'DELETE', 'COPY', 'COPY_FROM', 'COPY_TO', 'EXECUTE', 'SCRIPT', 'DDL', 'PRAGMA', 'MERGE', 'UNKNOWN']

expression

parameter_style

supports_many

parameter_casts

parameter_profile

operation_profile

__hash__()

Cached hash value.

Return type:

int

__eq__(other)

Equality comparison.

Return type:

bool

__repr__()

String representation.

Return type:

str

sqlspec.core.compiler.OperationType

alias of Literal[‘SELECT’, ‘INSERT’, ‘UPDATE’, ‘DELETE’, ‘COPY’, ‘COPY_FROM’, ‘COPY_TO’, ‘EXECUTE’, ‘SCRIPT’, ‘DDL’, ‘PRAGMA’, ‘MERGE’, ‘UNKNOWN’]

Statement Caching

class sqlspec.core.cache.UnifiedCache

Bases: object

Cache with LRU eviction and TTL support.

Parameters:

• max_size (int) – Maximum number of items to cache (LRU eviction when exceeded)

• ttl_seconds (int | None) – Time-to-live in seconds (None for no expiration)

Unified LRU cache for SQL statements and compilation results.

__init__(max_size=10000, ttl_seconds=3600)

Initialize unified cache.

Parameters:

• max_size (int) – Maximum number of cache entries

• ttl_seconds (int | None) – Time-to-live in seconds (None for no expiration)

get(key)

Get value from cache.

Parameters:

key (CacheKey) – Cache key to lookup

Return type:

Any | None

Returns:

Cached value or None if not found or expired

put(key, value)

Put value in cache.

Parameters:

• key (CacheKey) – Cache key

• value (Any) – Value to cache

Return type:

None

delete(key)

Delete entry from cache.

Parameters:

key (CacheKey) – Cache key to delete

Return type:

bool

Returns:

True if key was found and deleted, False otherwise

clear()

Clear all cache entries.

Return type:

None

size()

Get current cache size.

Return type:

int

is_empty()

Check if cache is empty.

Return type:

bool

get_stats()

Get cache statistics.

Return type:

CacheStats

__len__()

Get current cache size.

Return type:

int

__contains__(key)

Check if key exists in cache.

Return type:

bool

class sqlspec.core.cache.MultiLevelCache

Bases: object

Single cache with namespace isolation - no connection pool complexity.

Multi-level cache system for different caching strategies.

__init__(max_size=10000, ttl_seconds=3600)

Initialize multi-level cache.

Parameters:

• max_size (int) – Maximum number of cache entries

• ttl_seconds (int | None) – Time-to-live in seconds (None for no expiration)

get(level, key, dialect=None)

Get value from cache with level and dialect namespace.

Parameters:

• level (str) – Cache level (e.g., “statement”, “expression”, “parameter”)

• key (str) – Cache key

• dialect (str | None) – SQL dialect (optional)

Return type:

Any | None

Returns:

Cached value or None if not found

put(level, key, value, dialect=None)

Put value in cache with level and dialect namespace.

Parameters:

• level (str) – Cache level (e.g., “statement”, “expression”, “parameter”)

• key (str) – Cache key

• value (Any) – Value to cache

• dialect (str | None) – SQL dialect (optional)

Return type:

None

delete(level, key, dialect=None)

Delete entry from cache.

Parameters:

• level (str) – Cache level

• key (str) – Cache key to delete

• dialect (str | None) – SQL dialect (optional)

Return type:

bool

Returns:

True if key was found and deleted, False otherwise

clear()

Clear all cache entries.

Return type:

None

get_stats()

Get cache statistics.

Return type:

CacheStats

class sqlspec.core.cache.CacheConfig

Bases: object

Global cache configuration for SQLSpec.

Configuration for caching behavior.

__init__(*, compiled_cache_enabled=True, sql_cache_enabled=True, fragment_cache_enabled=True, optimized_cache_enabled=True, sql_cache_size=1000, fragment_cache_size=5000, optimized_cache_size=2000)

Initialize cache configuration.

Parameters:

• compiled_cache_enabled (bool) – Enable compiled SQL caching

• sql_cache_enabled (bool) – Enable SQL statement caching

• fragment_cache_enabled (bool) – Enable AST fragment caching

• optimized_cache_enabled (bool) – Enable optimized expression caching

• sql_cache_size (int) – Maximum SQL cache entries

• fragment_cache_size (int) – Maximum fragment cache entries

• optimized_cache_size (int) – Maximum optimized cache entries

class sqlspec.core.cache.CacheStats

Bases: object

Cache statistics tracking.

Tracks cache metrics including hit rates, evictions, and memory usage.

Cache statistics tracking hits, misses, and evictions.

__init__()

Initialize cache statistics.

hits

misses

evictions

total_operations

memory_usage

property hit_rate: float

Calculate cache hit rate as percentage.

property miss_rate: float

Calculate cache miss rate as percentage.

record_hit()

Record a cache hit.

Return type:

None

record_miss()

Record a cache miss.

Return type:

None

record_eviction()

Record a cache eviction.

Return type:

None

reset()

Reset all statistics.

Return type:

None

__repr__()

String representation of cache statistics.

Return type:

str

class sqlspec.core.cache.CacheKey

Bases: object

Immutable cache key.

Parameters:

key_data (tuple[Any, ...]) – Tuple of hashable values that uniquely identify the cached item

Cache key for tracking cached items.

__init__(key_data)

Initialize cache key.

Parameters:

key_data (tuple[Any, ...]) – Tuple of hashable values for the cache key

property key_data: tuple[Any, ...]

Get the key data tuple.

__hash__()

Return cached hash value.

Return type:

int

__eq__(other)

Equality comparison.

Return type:

bool

__repr__()

String representation of the cache key.

Return type:

str

SQL Filters

Filters transform SQL statements by adding or modifying clauses.

class sqlspec.core.filters.StatementFilter

Bases: ABC

Abstract base class for filters that can be appended to a statement.

Base class for SQL statement filters.

All filters implement the append_to_statement() method to modify SQL statements.

abstractmethod append_to_statement(statement)

Append the filter to the statement.

This method should modify the SQL expression only, not the parameters. Parameters should be provided via extract_parameters().

Return type:

SQL

extract_parameters()

Extract parameters that this filter contributes.

Returns:

• positional_parameters: List of positional parameter values

• named_parameters: Dict of parameter name to value

Return type:

Tuple of (positional_parameters, named_parameters) where

abstractmethod get_cache_key()

Return a cache key for this filter’s configuration.

Return type:

tuple[Any, ...]

Returns:

Tuple of hashable values representing the filter’s configuration

Built-in Filters

class sqlspec.core.filters.LimitOffsetFilter

Bases: PaginationFilter

Filter for LIMIT and OFFSET clauses.

Adds pagination support through LIMIT/OFFSET SQL clauses.

Adds LIMIT and OFFSET clauses.

from sqlspec.core import LimitOffsetFilter

filter = LimitOffsetFilter(limit=10, offset=20)
filtered_sql = filter.append_to_statement(base_sql)
# Adds: LIMIT 10 OFFSET 20

__init__(limit, offset)

property limit: int

property offset: int

get_param_names()

Get parameter names without storing them.

Return type:

list[str]

extract_parameters()

Extract filter parameters.

Return type:

tuple[list[Any], dict[str, Any]]

append_to_statement(statement)

Append the filter to the statement.

This method should modify the SQL expression only, not the parameters. Parameters should be provided via extract_parameters().

Return type:

SQL

get_cache_key()

Return cache key for this filter configuration.

Return type:

tuple[Any, ...]

class sqlspec.core.filters.OrderByFilter

Bases: StatementFilter

Filter for ORDER BY clauses.

Adds sorting capability to SQL queries.

Adds ORDER BY clause.

from sqlspec.core import OrderByFilter

filter = OrderByFilter(field_name="created_at", sort_order="desc")
filtered_sql = filter.append_to_statement(base_sql)
# Adds: ORDER BY created_at DESC

__init__(field_name, sort_order='asc')

property field_name: str

property sort_order: Literal['asc', 'desc']

extract_parameters()

Extract filter parameters.

Return type:

tuple[list[Any], dict[str, Any]]

append_to_statement(statement)

Append the filter to the statement.

This method should modify the SQL expression only, not the parameters. Parameters should be provided via extract_parameters().

Return type:

SQL

get_cache_key()

Return cache key for this filter configuration.

Return type:

tuple[Any, ...]

class sqlspec.core.filters.SearchFilter

Bases: StatementFilter

Filter for text search queries.

Constructs WHERE field_name LIKE ‘%value%’ clauses.

Adds WHERE search condition.

from sqlspec.core import SearchFilter

filter = SearchFilter(field_name="name", value="John", operator="ILIKE")
filtered_sql = filter.append_to_statement(base_sql)
# Adds: WHERE name ILIKE '%John%'

__init__(field_name, value, ignore_case=False)

property field_name: str | set[str]

property value: str

property ignore_case: bool | None

get_param_name()

Get parameter name without storing it.

Return type:

str | None

extract_parameters()

Extract filter parameters.

Return type:

tuple[list[Any], dict[str, Any]]

append_to_statement(statement)

Append the filter to the statement.

This method should modify the SQL expression only, not the parameters. Parameters should be provided via extract_parameters().

Return type:

SQL

get_cache_key()

Return cache key for this filter configuration.

Return type:

tuple[Any, ...]

Filter Composition

Filters can be composed and chained:

from sqlspec.core import (
    LimitOffsetFilter,
    OrderByFilter,
    SearchFilter
)
from sqlspec.core import SQL

base_sql = SQL("SELECT * FROM users")

# Apply multiple filters
filtered = base_sql
for filter_obj in [
    SearchFilter(field_name="name", value="Alice"),
    OrderByFilter(field_name="created_at", sort_order="desc"),
    LimitOffsetFilter(limit=10, offset=0)
]:
    filtered = filter_obj.append_to_statement(filtered)

# Result: SELECT * FROM users
#         WHERE name ILIKE '%Alice%'
#         ORDER BY created_at DESC
#         LIMIT 10 OFFSET 0

Type Conversions

class sqlspec.core.type_conversion.BaseTypeConverter

Bases: object

Universal type detection and conversion for all adapters.

Provides centralized type detection and conversion functionality that can be used across all database adapters to ensure consistent behavior. Users can extend this class for custom type conversion needs.

Base class for type conversion between Python types and database types.

detect_type(value)

Detect special types from string values.

Parameters:

value (str) – String value to analyze.

Return type:

str | None

Returns:

Type name if detected, None otherwise.

convert_value(value, detected_type)

Convert string value to appropriate Python type.

Parameters:

• value (str) – String value to convert.

• detected_type (str) – Detected type name.

Return type:

Any

Returns:

Converted value in appropriate Python type.

convert_if_detected(value)

Convert value only if special type detected, else return original.

This method provides performance optimization by avoiding expensive regex operations on plain strings that don’t contain special characters.

Parameters:

value (Any) – Value to potentially convert.

Return type:

Any

Returns:

Converted value if special type detected, original value otherwise.

Advanced Features

Statement Analysis

from sqlspec.core import SQLProcessor

compiler = SQLProcessor(dialect="postgres")

analysis = compiler.analyze("""
    SELECT u.name, COUNT(o.id) as order_count
    FROM users u
    LEFT JOIN orders o ON u.id = o.user_id
    WHERE u.active = TRUE
    GROUP BY u.id, u.name
    HAVING COUNT(o.id) > 5
""")

# Analysis includes:
# - Tables accessed
# - Columns referenced
# - Join types
# - Aggregate functions
# - Filter conditions
# - Potential optimizations

Custom Filters

Create custom filters for specific needs:

from sqlspec.core import StatementFilter
from sqlspec.core import SQL

class TenantFilter(StatementFilter):
    def __init__(self, tenant_id: int):
        self.tenant_id = tenant_id

    def apply(self, sql: SQL) -> SQL:
        # Add tenant condition to WHERE clause
        condition = f"tenant_id = {self.tenant_id}"
        if "WHERE" in sql.sql.upper():
            new_sql = sql.sql.replace("WHERE", f"WHERE {condition} AND")
        else:
            new_sql = sql.sql + f" WHERE {condition}"

        return SQL(new_sql, sql.parameters)

# Usage
filter = TenantFilter(tenant_id=123)
filtered = filter.apply(SQL("SELECT * FROM users"))
# Result: SELECT * FROM users WHERE tenant_id = 123

Performance Tips

1. Enable Caching

   config = StatementConfig(cache_statements=True)
   sql = SQL("SELECT * FROM users", config=config)
   

2. Reuse Compiled Statements

   # Compile once
   stmt = SQL("SELECT * FROM users WHERE id = ?")
   
   # Execute many times
   for user_id in user_ids:
       result = await session.execute(stmt, user_id)
   

3. Use Positional Parameters

   # Positional parameters
   stmt = SQL("SELECT * FROM users WHERE id = ?", 123)
   
   # Named parameters (requires parsing)
   stmt = SQL("SELECT * FROM users WHERE id = :id", id=123)
   

See Also

• Drivers and Querying - Query execution

• Builder - SQL builder integration

• Driver - Driver implementation

• Base - SQLSpec configuration