Configuration¶
SQLSpec provides flexible configuration options for database connections, connection pooling, and statement processing. This guide covers everything you need to configure SQLSpec for production use.
Overview¶
SQLSpec configuration is organized into three main areas:
Database Configuration: Connection parameters and pool settings
Statement Configuration: SQL processing pipeline behavior
Cache Configuration: Multi-tier caching system settings
Basic Configuration¶
The simplest way to use SQLSpec is with default configuration:
from sqlspec import SQLSpec
from sqlspec.adapters.sqlite import SqliteConfig
# Create SQLSpec instance
db_manager = SQLSpec()
# Add database configuration
db = db_manager.add_config(SqliteConfig(pool_config={"database": ":memory:"}))
# Use the database
with db_manager.provide_session(db) as session:
result = session.execute("SELECT 1")
Database Configurations¶
Each database adapter has its own configuration class with adapter-specific settings.
Note
Async PostgreSQL examples in this guide read their connection details from
SQLSPEC_USAGE_PG_* environment variables. The test suite populates these
variables (host, port, user, password, database, DSN) automatically so the
literalincluded snippets can stay identical to the documentation.
SQLite Configuration¶
from sqlspec.adapters.sqlite import SqliteConfig
config = SqliteConfig(
pool_config={
"database": "myapp.db", # Database file path
"timeout": 5.0, # Lock timeout in seconds
"check_same_thread": False, # Allow multi-thread access
"cached_statements": 100, # Statement cache size
"uri": False, # Enable URI mode
}
)
Memory Databases
from sqlspec.adapters.sqlite import SqliteConfig
# In-memory database (isolated per connection)
config = SqliteConfig(pool_config={"database": ":memory:"})
PostgreSQL Configuration (asyncpg)¶
import os
from sqlspec.adapters.asyncpg import AsyncpgConfig
host = os.getenv("SQLSPEC_USAGE_PG_HOST", "localhost")
port = int(os.getenv("SQLSPEC_USAGE_PG_PORT", "5432"))
user = os.getenv("SQLSPEC_USAGE_PG_USER", "user")
password = os.getenv("SQLSPEC_USAGE_PG_PASSWORD", "password")
database = os.getenv("SQLSPEC_USAGE_PG_DATABASE", "db")
dsn = os.getenv("SQLSPEC_USAGE_PG_DSN", f"postgresql://{user}:{password}@{host}:{port}/{database}")
config = AsyncpgConfig(
pool_config={
"dsn": dsn,
"min_size": 10,
"max_size": 20,
"host": host,
"port": port,
"user": user,
"password": password,
"database": database,
}
)
PostgreSQL Configuration (psycopg)¶
import os
from sqlspec.adapters.psycopg import PsycopgAsyncConfig
host = os.getenv("SQLSPEC_USAGE_PG_HOST", "localhost")
port = int(os.getenv("SQLSPEC_USAGE_PG_PORT", "5432"))
database = os.getenv("SQLSPEC_USAGE_PG_DATABASE", "db")
user = os.getenv("SQLSPEC_USAGE_PG_USER", "user")
password = os.getenv("SQLSPEC_USAGE_PG_PASSWORD", "password")
dsn = os.getenv("SQLSPEC_USAGE_PG_DSN", f"postgresql://{user}:{password}@{host}:{port}/{database}")
# Async version
config = PsycopgAsyncConfig(
pool_config={
"conninfo": dsn,
# Or keyword arguments:
"host": host,
"port": port,
"dbname": database,
"user": user,
"password": password,
# Pool settings
"min_size": 5,
"max_size": 10,
"timeout": 30.0,
}
)
MySQL Configuration (asyncmy)¶
from sqlspec.adapters.asyncmy import AsyncmyConfig
config = AsyncmyConfig(
pool_config={
"host": "localhost",
"port": 3306,
"user": "myuser",
"password": "mypassword",
"database": "mydb",
"charset": "utf8mb4",
"minsize": 1,
"maxsize": 10,
"pool_recycle": 3600,
}
)
DuckDB Configuration¶
from sqlspec.adapters.duckdb import DuckDBConfig
in_memory_config = DuckDBConfig()
Connection Pooling¶
Connection pooling improves performance by reusing database connections. SQLSpec provides built-in pooling for most adapters.
Pool Configuration¶
import os
from sqlspec.adapters.asyncpg import AsyncpgConfig
dsn = os.getenv("SQLSPEC_USAGE_PG_DSN", "postgresql://localhost/db")
config = AsyncpgConfig(
pool_config={
"dsn": dsn,
"min_size": 10,
"max_size": 20,
"max_queries": 50000,
"max_inactive_connection_lifetime": 300.0,
}
)
Pool Lifecycle Management
import os
from sqlspec import SQLSpec
from sqlspec.adapters.asyncpg import AsyncpgConfig
db_manager = SQLSpec()
dsn = os.getenv("SQLSPEC_USAGE_PG_DSN", "postgresql://localhost/db")
asyncpg_key = db_manager.add_config(AsyncpgConfig(pool_config={"dsn": dsn}))
asyncpg_config = db_manager.get_config(asyncpg_key)
Using Pre-Created Pools¶
import os
import asyncpg
from sqlspec.adapters.asyncpg import AsyncpgConfig
pool = asyncpg.create_pool(
dsn=os.getenv("SQLSPEC_USAGE_PG_DSN", "postgresql://localhost/db"), min_size=10, max_size=20
)
db = AsyncpgConfig(pool_instance=pool)
No-Pooling Configuration¶
from sqlspec.adapters.sqlite import SqliteConfig
config = SqliteConfig(pool_config={"database": "test.db"})
Statement Configuration¶
Statement configuration controls SQL processing pipeline behavior.
Basic Statement Config¶
import os
from sqlspec import StatementConfig
from sqlspec.adapters.asyncpg import AsyncpgConfig
statement_config = StatementConfig(
dialect="postgres", # SQLGlot dialect
enable_parsing=True, # Parse SQL into AST
enable_validation=True, # Run security/performance validators
enable_transformations=True, # Apply AST transformations
enable_caching=True, # Enable multi-tier caching
)
# Apply to adapter
dsn = os.getenv("SQLSPEC_USAGE_PG_DSN", "postgresql://localhost/db")
config = AsyncpgConfig(pool_config={"dsn": dsn}, statement_config=statement_config)
Parameter Style Configuration¶
from sqlspec import ParameterStyle, ParameterStyleConfig, StatementConfig
param_config = ParameterStyleConfig(
default_parameter_style=ParameterStyle.NUMERIC, # $1, $2, ...
supported_parameter_styles={
ParameterStyle.NUMERIC,
ParameterStyle.NAMED_COLON, # :name
},
has_native_list_expansion=False,
needs_static_script_compilation=False,
)
statement_config = StatementConfig(dialect="postgres", parameter_config=param_config)
Parameter Styles
SQLSpec supports multiple parameter placeholder styles:
from sqlspec import ParameterStyle
# Question mark (SQLite, DuckDB)
qmark = ParameterStyle.QMARK # WHERE id = ?
# Numeric (PostgreSQL, asyncpg)
numeric = ParameterStyle.NUMERIC # WHERE id = $1
# Named colon (Oracle, SQLite)
named_colon = ParameterStyle.NAMED_COLON # WHERE id = :id
# Named at (BigQuery)
# Format/pyformat (psycopg, MySQL)
Validation Configuration¶
Configure security and performance validation.
Disable validation for performance-critical paths where input is trusted:
from sqlspec.core.cache import CacheConfig, update_cache_config
cache_config = CacheConfig(
compiled_cache_enabled=True, # Cache compiled SQL
sql_cache_enabled=True, # Cache SQL strings
fragment_cache_enabled=True, # Cache SQL fragments
optimized_cache_enabled=True, # Cache optimized AST
sql_cache_size=1000, # Maximum cached SQL items
)
# Update global cache configuration
update_cache_config(cache_config)
# Verify config applied
Cache Configuration¶
SQLSpec uses multi-tier caching to avoid recompiling SQL statements.
Global Cache Configuration¶
from sqlspec.core.cache import CacheConfig, update_cache_config
cache_config = CacheConfig(
compiled_cache_enabled=True, # Cache compiled SQL
sql_cache_enabled=True, # Cache SQL strings
fragment_cache_enabled=True, # Cache SQL fragments
optimized_cache_enabled=True, # Cache optimized AST
sql_cache_size=1000, # Maximum cached SQL items
)
# Update global cache configuration
update_cache_config(cache_config)
# Verify config applied
Per-Instance Cache Configuration¶
import tempfile
from sqlspec import CacheConfig, SQLSpec
from sqlspec.adapters.sqlite import SqliteConfig
with tempfile.NamedTemporaryFile(suffix=".db", delete=True) as tmp:
# Configure cache for specific SQLSpec instance
db_manager = SQLSpec()
db_manager.update_cache_config(CacheConfig(sql_cache_enabled=True, sql_cache_size=500))
# Add database config
db = db_manager.add_config(SqliteConfig(pool_config={"database": tmp.name}))
# Use the configured spec
with db_manager.provide_session(db) as session:
result = session.execute("SELECT 1")
Cache Statistics¶
Monitor cache statistics:
import tempfile
from sqlspec import SQLSpec
from sqlspec.adapters.sqlite import SqliteConfig
from sqlspec.core import get_cache_statistics, log_cache_stats
with tempfile.NamedTemporaryFile(suffix=".db", delete=True) as tmp:
db_manager = SQLSpec()
db = db_manager.add_config(SqliteConfig(pool_config={"database": tmp.name}))
# Execute some queries to generate cache activity
with db_manager.provide_session(db) as session:
session.execute("SELECT 1")
session.execute("SELECT 1") # Should hit cache
# Get statistics
stats = get_cache_statistics()
Clear Cache¶
from sqlspec.core import clear_all_caches, get_cache_statistics
# Get initial statistics
stats_before = get_cache_statistics()
Multiple Database Configurations¶
SQLSpec supports multiple database configurations in a single application.
Binding Multiple Configs¶
import os
import tempfile
from sqlspec import SQLSpec
from sqlspec.adapters.asyncpg import AsyncpgConfig
from sqlspec.adapters.sqlite import SqliteConfig
with tempfile.NamedTemporaryFile(suffix=".db", delete=True) as tmp:
db_manager = SQLSpec()
# Add multiple configurations
sqlite_key = db_manager.add_config(SqliteConfig(pool_config={"database": tmp.name}))
dsn = os.getenv("SQLSPEC_USAGE_PG_DSN", "postgresql://localhost/db")
asyncpg_key = db_manager.add_config(AsyncpgConfig(pool_config={"dsn": dsn}))
# Use specific configuration
with db_manager.provide_session(sqlite_key) as session:
session.execute("SELECT 1")
sqlite_config = db_manager.get_config(sqlite_key)
pg_config = db_manager.get_config(asyncpg_key)
Named Bindings¶
Use bind keys for clearer configuration management:
import os
import tempfile
from sqlspec import SQLSpec
from sqlspec.adapters.asyncpg import AsyncpgConfig
from sqlspec.adapters.sqlite import SqliteConfig
with tempfile.NamedTemporaryFile(suffix=".db", delete=True) as tmp:
db_manager = SQLSpec()
# Add with bind keys
cache_key = db_manager.add_config(SqliteConfig(pool_config={"database": tmp.name}, bind_key="cache_db"))
dsn = os.getenv("SQLSPEC_USAGE_PG_DSN", "postgresql://localhost/db")
main_key = db_manager.add_config(AsyncpgConfig(pool_config={"dsn": dsn}, bind_key="main_db"))
# Access by bind key
with db_manager.provide_session(cache_key) as session:
session.execute("SELECT 1")
cache_config = db_manager.get_config(cache_key)
main_config = db_manager.get_config(main_key)
SQLSpec.add_config() returns a key for the registered configuration.
Store that value (as shown above) and reuse it when calling
provide_session() or get_config(). The config registry holds a single
instance per config type, so creating multiple variants of the same adapter
requires defining lightweight subclasses or binding unique config classes for
each database.
Migration Configuration¶
SQLSpec includes a migration system for schema management.
Basic Migration Config¶
import os
from sqlspec.adapters.asyncpg import AsyncpgConfig
dsn = os.getenv("SQLSPEC_USAGE_PG_DSN", "postgresql://localhost/db")
config = AsyncpgConfig(
pool_config={"dsn": dsn},
extension_config={
"litestar": {"session_table": "custom_sessions"} # Extension settings
},
migration_config={
"script_location": "migrations", # Migration directory
"version_table": "alembic_version", # Version tracking table
"include_extensions": ["litestar"], # Simple string list only
},
)
Migration CLI
# Create migration
sqlspec --config myapp.config create-migration -m "Add users table"
# Apply migrations
sqlspec --config myapp.config upgrade
# Rollback
sqlspec --config myapp.config downgrade -1
Extension Migration Versioning¶
Extension migrations are automatically prefixed to prevent version collisions with user migrations:
# User migrations
0001_initial.py → version: 0001
0002_add_users.py → version: 0002
# Extension migrations (automatic prefix)
ext_adk_0001 → ADK tables migration
ext_litestar_0001 → Litestar session table migration
This ensures extension migrations never conflict with your application migrations in the version tracking table.
Extension Configuration¶
Framework integrations can be configured via extension_config.
Litestar Plugin Configuration¶
import os
from sqlspec.adapters.asyncpg import AsyncpgConfig
dsn = os.getenv("SQLSPEC_USAGE_PG_DSN", "postgresql://localhost/db")
config = AsyncpgConfig(
pool_config={"dsn": dsn},
extension_config={
"litestar": {
"connection_key": "db_connection",
"session_key": "db_session",
"pool_key": "db_pool",
"commit_mode": "autocommit_include_redirect",
"extra_commit_statuses": {201},
"extra_rollback_statuses": {422},
"enable_correlation_middleware": True,
"correlation_header": "x-request-id",
"correlation_headers": ["traceparent", "x-correlation-id"],
"auto_trace_headers": False,
"disable_di": False,
}
},
)
Telemetry Snapshot¶
Call SQLSpec.telemetry_snapshot() to inspect lifecycle counters, serializer metrics, and recent storage jobs:
from sqlspec import SQLSpec
from sqlspec.adapters.sqlite import SqliteConfig
db_manager = SQLSpec()
db = db_manager.add_config(SqliteConfig(pool_config={"database": ":memory:"}))
with db_manager.provide_session(db) as session:
session.execute("SELECT 1")
snapshot = db_manager.telemetry_snapshot()
Environment-Based Configuration¶
Use environment variables for configuration:
import os
from unittest.mock import patch
# Mock environment variables
env_vars = {
"DB_HOST": "testhost",
"DB_PORT": "5433",
"DB_USER": "testuser",
"DB_PASSWORD": "testpass",
"DB_NAME": "testdb",
}
with patch.dict(os.environ, env_vars, clear=False):
from sqlspec.adapters.asyncpg import AsyncpgConfig
config = AsyncpgConfig(
pool_config={
"host": os.getenv("DB_HOST", "localhost"),
"port": int(os.getenv("DB_PORT", "5432")),
"user": os.getenv("DB_USER"),
"password": os.getenv("DB_PASSWORD"),
"database": os.getenv("DB_NAME"),
}
)
Configuration Best Practices¶
1. Use Connection Pooling
Always use pooling in production:
import os
from sqlspec.adapters.asyncpg import AsyncpgConfig
dsn = os.getenv("SQLSPEC_USAGE_PG_DSN", "postgresql://localhost/db")
config = AsyncpgConfig(pool_config={"dsn": dsn, "min_size": 10, "max_size": 20})
2. Enable Caching
Enable caching to avoid recompiling SQL statements:
from sqlspec import StatementConfig
statement_config = StatementConfig(dialect="postgres", enable_caching=True)
3. Tune Pool Sizes
Size pools based on your workload:
# CPU-bound workload - smaller pool
cpu_bound_pool_config = {"min_size": 5, "max_size": 10}
4. Disable Validation in Production
For trusted, performance-critical queries:
from sqlspec import StatementConfig
# Example: Disabling security checks for trusted internal queries
statement_config = StatementConfig(
dialect="postgres",
enable_validation=False, # Skip security checks
)
5. Clean Up Resources
Always close pools on shutdown:
import tempfile
from sqlspec import SQLSpec
from sqlspec.adapters.aiosqlite import AiosqliteConfig
with tempfile.NamedTemporaryFile(suffix=".db", delete=True) as tmp:
db_manager = SQLSpec()
db = db_manager.add_config(AiosqliteConfig(pool_config={"database": tmp.name}))
# Use the connection
async with db_manager.provide_session(db) as session:
await session.execute("CREATE TABLE test (id INTEGER)")
# Clean up resources - important for async adapters
await db_manager.close_all_pools()
# Verify pools are closed
Next Steps¶
Now that you understand configuration:
Drivers and Querying - Execute queries with your configured databases
Framework Integrations - Integrate with web frameworks
Adapters - Detailed adapter reference
See Also¶
Base - SQLSpec base class API
Core - Core configuration classes
Execution Flow - Understanding the execution pipeline