Creating Adapters

This guide explains how to create a new database adapter for SQLSpec. An adapter integrates a specific database system into SQLSpec’s unified interface, providing configuration, driver implementation, and type conversion capabilities.

Overview of Adapter Architecture

SQLSpec adapters follow a standardized structure that separates concerns into distinct modules:

Adapter Components

Each adapter consists of the following key components:

  1. Configuration Module (config.py): Defines connection configuration classes

  2. Driver Module (driver.py): Implements the database driver that executes queries

  3. Type Converter Module (type_converter.py): Optional, for database-specific type handling

  4. Types Module (_types.py): Type definitions for database-specific objects (not compiled by mypyc)

  5. Package Initialization (__init__.py): Exports the public API

Directory Structure

sqlspec/adapters/your_adapter/

__init__.py # Public exports

config.py # Configuration classes

driver.py # Driver implementation

_types.py # Type definitions (optional)

type_converter.py # Type conversion (optional)

Design Principles

  • Separation of Concerns: Configuration is separate from execution logic

  • Protocol-Based Design: Use SQLSpec’s protocols for type safety

  • Minimal Abstraction: Stay close to the underlying database API

  • Type Safety: Leverage Python’s type system extensively

MANDATORY Parameter Parsing Flow

Danger

CRITICAL: This section describes the MANDATORY parameter parsing flow that ALL SQLSpec drivers MUST follow. This is not optional or advisoryit is a core architectural requirement. Deviation from this flow will cause parameter handling failures, security vulnerabilities, and incompatibility with SQLGlot transformations.

Understanding the Problem

Different databases use different parameter placeholder styles:

  • PostgreSQL (asyncpg): $1, $2, $3 (numeric)

  • PostgreSQL (psycopg): %s or %(name)s (pyformat)

  • MySQL: %s (positional pyformat)

  • SQLite: ? (qmark) or :name (named)

  • Oracle: :name or :1 (named/positional colon)

SQLGlot, which SQLSpec uses for SQL parsing and AST manipulation, doesn’t natively understand all these styles. More critically, when SQLGlot transforms the AST (e.g., parameterizing literals), it can change parameter positions, making simple string replacement strategies fail.

The Parameter Parsing Flow

SQLSpec solves this with a sophisticated multi-phase pipeline that all drivers MUST follow:

        graph TB
   A[Original SQL + Parameters] --> B[1. Parameter Extraction];
   B --> C[2. SQL Normalization];
   C --> D[3. SQLGlot Processing];
   D --> E[4. Parameter Style Restoration];
   E --> F[5. Final SQL + Parameters];

   B -.-> G[ParameterInfo extracted];
   C -.-> H[Normalized for SQLGlot];
   D -.-> I[AST transformations];
   E -.-> J[Original style restored];

   style A fill:#e1f5ff
   style F fill:#d4f1d4
   style B fill:#fff4e6
   style C fill:#fff4e6
   style D fill:#ffe6f0
   style E fill:#fff4e6

Step-by-Step Breakdown

Phase 1: Parameter Extraction (MANDATORY)

Before any SQL modifications, extract and preserve the original parameter information using ParameterValidator.

from sqlspec.core import ParameterValidator, ParameterStyle

validator = ParameterValidator(
    default_style=ParameterStyle.NUMERIC,  # or your adapter's style
    allowed_styles={ParameterStyle.NUMERIC}
)

# Extract parameter metadata
param_info = validator.extract_parameter_info(sql_string, parameters)

The ParameterInfo object captures:

  • Parameter style (e.g., $1 vs ? vs :name)

  • Parameter positions in the original SQL

  • Whether parameters are named or positional

  • Original parameter count and names

Why This Matters: This metadata is essential for correctly mapping parameters after AST transformations change the SQL structure.

Phase 2: SQL Normalization (CONDITIONAL)

Convert SQL to a format that SQLGlot can parse correctly. This is only required for parameter styles that SQLGlot doesn’t understand.

Normalization Requirements

Parameter Style

Needs Normalization?

Normalization Strategy

? (qmark)

 Yes

Replace with :param1, :param2, etc.

$1 (numeric)

 Yes

Replace with :param1, :param2, etc.

:name (named colon)

L No

SQLGlot understands natively

%s (pyformat positional)

 Yes

Replace with :param1, :param2, etc.

%(name)s (pyformat named)

 Yes

Replace with :name equivalents

Example: AsyncPG Driver (PostgreSQL with $1 style)

from sqlspec.core import ParameterConverter

converter = ParameterConverter()

# Original: SELECT * FROM users WHERE id = $1 AND status = $2
# Normalized: SELECT * FROM users WHERE id = :param1 AND status = :param2
normalized_sql = converter.convert_sql(
    sql=original_sql,
    from_style=ParameterStyle.NUMERIC,
    to_style=ParameterStyle.NAMED_COLON
)

Phase 3: SQLGlot Processing (MANDATORY)

Parse the normalized SQL into an Abstract Syntax Tree (AST) and apply transformations.

from sqlspec.core import SQLGlotCompiler

compiler = SQLGlotCompiler(dialect="postgres")

# Parse into AST
ast = compiler.parse(normalized_sql)

# Apply transformations (e.g., parameterize literals)
# This might add NEW parameters!
transformed_ast = compiler.apply_transformations(ast, transformers=[...])

# Generate SQL from transformed AST
transformed_sql = compiler.generate(transformed_ast)

Critical: At this point, the SQL may have:

  • Different parameter positions than the original

  • Additional parameters (from literal parameterization)

  • Modified structure (from AST optimizations)

Phase 4: Parameter Style Restoration (CONDITIONAL)

If normalization was applied in Phase 2, convert the SQL back to the database’s native parameter style.

# transformed_sql uses :param1, :param2, etc.
# Need to convert back to $1, $2 for asyncpg

final_sql = converter.convert_sql(
    sql=transformed_sql,
    from_style=ParameterStyle.NAMED_COLON,
    to_style=ParameterStyle.NUMERIC
)

# Result: SELECT * FROM users WHERE id = $1 AND status = $2 AND created > $3
# (assuming a literal was parameterized to $3)

Phase 5: Final SQL Delivery (MANDATORY)

Deliver the final SQL and correctly ordered parameters to the database driver.

# Execute with final SQL and parameter mapping
result = await connection.execute(final_sql, *final_parameters)

Implementation Guide

This section walks through creating a complete adapter from scratch.

Step 1: Create the Configuration Class

The configuration class defines how to connect to the database.

# sqlspec/adapters/your_adapter/config.py

from dataclasses import dataclass
from typing import TYPE_CHECKING, Any

from sqlspec.config import SQLConfig

if TYPE_CHECKING:
    from sqlspec.driver import DriverProtocol

@dataclass
class YourAdapterConfig(SQLConfig):
    """Configuration for Your Database.

    Args:
        host: Database host
        port: Database port
        database: Database name
        user: Username
        password: Password
        driver_args: Additional driver arguments
    """

    host: str = "localhost"
    port: int = 5432
    database: str = "mydb"
    user: str = "user"
    password: str = "password"
    driver_args: "dict[str, Any] | None" = None

    @property
    def driver_class(self) -> "type[DriverProtocol]":
        """Return the driver class for this configuration."""
        from sqlspec.adapters.your_adapter.driver import YourAdapterDriver
        return YourAdapterDriver

    @property
    def connection_string(self) -> str:
        """Generate connection string."""
        return f"{self.host}:{self.port}/{self.database}"

Key Points:

  • Inherit from SQLConfig

  • Use dataclass for clean configuration

  • Implement driver_class property (lazy import to avoid circular dependencies)

  • Optional: Implement connection_string for logging/debugging

Step 2: Create the Driver Class

The driver executes SQL statements and manages transactions.

# sqlspec/adapters/your_adapter/driver.py

from typing import TYPE_CHECKING, Any

from sqlspec.core import StatementConfig
from sqlspec.driver import AsyncDriverAdapterBase  # or SyncDriverAdapterBase

if TYPE_CHECKING:
    from contextlib import AbstractAsyncContextManager
    from sqlspec.core import SQLResult
    from sqlspec.core import SQL
    from sqlspec.driver import ExecutionResult

class YourAdapterDriver(AsyncDriverAdapterBase):
    """Driver implementation for Your Database."""

    __slots__ = ("_connection", "_config")

    def __init__(self, connection: Any, config: "YourAdapterConfig") -> None:
        self._connection = connection
        self._config = config

    @property
    def statement_config(self) -> "StatementConfig":
        """Return statement configuration for this driver."""
        return your_adapter_statement_config

    def handle_database_exceptions(self) -> "AbstractAsyncContextManager[None]":
        """Handle database-specific exceptions."""
        return YourAdapterExceptionHandler()

    def with_cursor(self, connection: Any) -> Any:
        """Create cursor context manager."""
        return YourAdapterCursor(connection)

    async def _execute_statement(self, cursor: Any, statement: "SQL") -> "ExecutionResult":
        """Execute a single SQL statement."""
        result = await cursor.execute(statement.sql, statement.parameters)
        return ExecutionResult(
            rows=await cursor.fetchall(),
            rowcount=cursor.rowcount,
            description=cursor.description
        )

    async def _execute_many(self, cursor: Any, statement: "SQL") -> "ExecutionResult":
        """Execute statement with multiple parameter sets."""
        await cursor.executemany(statement.sql, statement.parameters)
        return ExecutionResult(rows=[], rowcount=cursor.rowcount)

    async def _execute_script(self, cursor: Any, statement: "SQL") -> "ExecutionResult":
        """Execute multiple SQL statements."""
        await cursor.executescript(statement.sql)
        return ExecutionResult(rows=[], rowcount=-1)

    async def _try_special_handling(self, cursor: Any, statement: "SQL") -> "SQLResult | None":
        """Handle database-specific operations (e.g., COPY, bulk insert)."""
        return None

    async def begin(self) -> None:
        """Begin transaction."""
        await self._connection.begin()

    async def commit(self) -> None:
        """Commit transaction."""
        await self._connection.commit()

    async def rollback(self) -> None:
        """Rollback transaction."""
        await self._connection.rollback()

Statement Configuration

Define how the driver handles parameters and SQL processing:

from sqlspec.core import ParameterStyle, ParameterStyleConfig
from sqlspec.core import StatementConfig

your_adapter_statement_config = StatementConfig(
    dialect="postgres",  # SQLGlot dialect name
    parameter_config=ParameterStyleConfig(
        # What parameter styles can users write?
        default_parameter_style=ParameterStyle.NUMERIC,
        supported_parameter_styles={ParameterStyle.NUMERIC},

        # What style does the driver actually use?
        default_execution_parameter_style=ParameterStyle.NUMERIC,
        supported_execution_parameter_styles={ParameterStyle.NUMERIC},

        # Does the driver expand lists natively? (e.g., WHERE id IN ($1))
        has_native_list_expansion=True,

        # Does the driver require static SQL strings for scripts?
        needs_static_script_compilation=False,

        # Should parameter format be preserved through transformations?
        preserve_parameter_format=True,
    ),
    enable_parsing=True,       # Parse SQL with SQLGlot
    enable_validation=True,     # Run security/performance validators
    enable_caching=True,        # Cache parsed statements
    enable_parameter_type_wrapping=True,  # Wrap parameters with type info
)

Step 3: Exception Handling

Map database-specific exceptions to SQLSpec’s exception hierarchy:

from sqlspec.exceptions import (
    UniqueViolationError,
)

class YourAdapterExceptionHandler:
    """Map database exceptions to SQLSpec exceptions."""

    async def __aenter__(self) -> None:
        return None

    async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
        if exc_type is None:
            return

        # Map database-specific exceptions
        if issubclass(exc_type, YourDatabaseError):
            error_code = getattr(exc_val, "error_code", None)

            if error_code == "23505":  # Unique violation
                msg = f"Unique constraint violation: {exc_val}"
                raise UniqueViolationError(msg) from exc_val

            # Add more mappings...

Step 4: Testing Requirements

Create comprehensive tests for your adapter:

# tests/integration/test_adapters/test_your_adapter/test_basic.py

import pytest
from sqlspec import SQLSpec
from sqlspec.adapters.your_adapter import YourAdapterConfig

@pytest.fixture
def config() -> YourAdapterConfig:
    """Create test configuration."""
    return YourAdapterConfig(
        host="localhost",
        database="test_db"
    )

@pytest.fixture
async def sqlspec(config: YourAdapterConfig) -> SQLSpec:
    """Create SQLSpec instance with adapter."""
    dbs = SQLSpec()
    dbs.add_config(config)
    return dbs

async def test_basic_select(sqlspec: SQLSpec, config: YourAdapterConfig) -> None:
    """Test basic SELECT query."""
    async with sqlspec.provide_session(config) as session:
        result = await session.execute("SELECT 1 AS value")
        row = result.one()
        assert row["value"] == 1

async def test_parameter_binding(sqlspec: SQLSpec, config: YourAdapterConfig) -> None:
    """Test parameter binding with adapter's style."""
    async with sqlspec.provide_session(config) as session:
        # Test with adapter's native parameter style
        result = await session.execute(
            "SELECT $1 AS value",  # or ?, :1, etc.
            [42]
        )
        assert result.one()["value"] == 42

async def test_transaction_rollback(sqlspec: SQLSpec, config: YourAdapterConfig) -> None:
    """Test transaction rollback."""
    async with sqlspec.provide_session(config) as session:
        await session.execute("CREATE TABLE test (id INT)")

        async with session.begin():
            await session.execute("INSERT INTO test VALUES (1)")
            raise Exception("Rollback test")

        # Verify rollback happened
        result = await session.execute("SELECT COUNT(*) FROM test")
        assert result.scalar() == 0

Test Coverage Requirements:

  1. Basic CRUD operations (SELECT, INSERT, UPDATE, DELETE)

  2. Parameter binding (all supported styles)

  3. Transaction management (commit, rollback, nested)

  4. Exception handling (constraint violations, connection errors)

  5. Type conversion (date, datetime, JSON, binary, etc.)

  6. Special features (COPY, bulk operations, etc.)

Step 5: Export Public API

Define the public interface in __init__.py:

# sqlspec/adapters/your_adapter/__init__.py

"""Your Database adapter for SQLSpec."""

from sqlspec.adapters.your_adapter.config import YourAdapterConfig
from sqlspec.adapters.your_adapter.driver import (
    YourAdapterDriver,
    YourAdapterExceptionHandler,
    your_adapter_statement_config,
)

__all__ = (
    "YourAdapterConfig",
    "YourAdapterDriver",
    "YourAdapterExceptionHandler",
    "your_adapter_statement_config",
)

Best Practices

Performance Optimization

Connection Pooling

Implement connection pooling for production use:

@dataclass
class YourAdapterPoolConfig(SQLConfig):
    """Configuration with connection pooling."""

    pool_size: int = 10
    max_overflow: int = 5
    pool_timeout: int = 30

    async def create_pool(self) -> Any:
        """Create connection pool."""
        return await your_database.create_pool(
            host=self.host,
            database=self.database,
            min_size=self.pool_size,
            max_size=self.pool_size + self.max_overflow,
            timeout=self.pool_timeout,
        )

Statement Caching

Enable statement caching in your StatementConfig:

StatementConfig(
    enable_caching=True,  # Cache parsed SQL statements
    # ...
)

Prepared Statements

Use prepared statements when supported:

async def _execute_statement(self, cursor: Any, statement: "SQL") -> "ExecutionResult":
    """Execute with prepared statement support."""
    if statement.should_prepare:
        prepared = await cursor.prepare(statement.sql)
        result = await prepared.fetch(*statement.parameters)
    else:
        result = await cursor.execute(statement.sql, statement.parameters)
    # ...

Type Conversion

Implement database-specific type converters:

# sqlspec/adapters/your_adapter/type_converter.py

from datetime import datetime
from typing import Any

def convert_timestamp(value: Any) -> datetime:
    """Convert database timestamp to Python datetime."""
    if isinstance(value, str):
        return datetime.fromisoformat(value)
    return value

def convert_json(value: Any) -> dict | list:
    """Convert database JSON to Python dict/list."""
    import json
    if isinstance(value, str):
        return json.loads(value)
    return value

Then register in ParameterStyleConfig:

ParameterStyleConfig(
    type_coercion_map={
        datetime: convert_timestamp,
        dict: convert_json,
    },
    # ...
)

Security Considerations

SQL Injection Prevention

Always use parameterized queries:

# GOOD: Parameterized
await session.execute("SELECT * FROM users WHERE id = $1", [user_id])

# BAD: String interpolation
await session.execute(f"SELECT * FROM users WHERE id = {user_id}")

Validator Configuration

Enable SQLSpec’s built-in validators:

StatementConfig(
    enable_validation=True,  # Security and performance validators
    # ...
)

Exception Sanitization

Avoid leaking sensitive information in error messages:

def _map_exception(self, exc: Exception) -> SQLSpecError:
    """Map exception without exposing internals."""
    # Don't include full SQL or connection strings in messages
    msg = "Database operation failed"
    return SQLSpecError(msg) from exc

Logging and Observability

Add comprehensive logging:

from sqlspec.utils.logging import get_logger

logger = get_logger("adapters.your_adapter")

async def _execute_statement(self, cursor: Any, statement: "SQL") -> "ExecutionResult":
    """Execute statement with logging."""
    logger.debug("Executing statement", extra={
        "sql": statement.sql,
        "param_count": len(statement.parameters),
    })

    try:
        result = await cursor.execute(statement.sql, statement.parameters)
        logger.debug("Statement executed successfully", extra={
            "rowcount": cursor.rowcount,
        })
        return result
    except Exception as e:
        logger.error("Statement execution failed", extra={
            "error": str(e),
        }, exc_info=True)
        raise

Common Pitfalls

Pitfall 1: Skipping Parameter Extraction

# WRONG: Directly modifying SQL without extracting parameter info
normalized_sql = sql.replace("$1", ":param1")

# RIGHT: Use ParameterValidator first
param_info = validator.extract_parameter_info(sql, parameters)
normalized_sql = converter.convert_sql(sql, ParameterStyle.NUMERIC, ParameterStyle.NAMED_COLON)

Pitfall 2: Incorrect Parameter Ordering

# WRONG: Assuming parameters stay in original order after AST transformation
final_params = original_params

# RIGHT: Track parameter mapping through transformations
final_params = param_processor.map_parameters_to_final_positions(
    original_params, param_info, transformed_sql
)

Pitfall 3: Missing Exception Mapping

# WRONG: Let database exceptions bubble up
result = await cursor.execute(sql, params)

# RIGHT: Use exception handler context manager
async with self.handle_database_exceptions():
    result = await cursor.execute(sql, params)

Pitfall 4: Not Implementing Data Dictionary

The data dictionary enables introspection features:

@property
def data_dictionary(self) -> "AsyncDataDictionaryBase":
    """Required for metadata queries."""
    return YourAdapterDataDictionary(self._connection)

Reference Implementations

Study these existing adapters as examples:

Simple Adapters:

  • sqlspec/adapters/sqlite/: Minimal async adapter with qmark parameters

  • sqlspec/adapters/duckdb/: Embedded database with rich type system

Complex Adapters:

  • sqlspec/adapters/asyncpg/: Full-featured with connection pooling, COPY support, numeric parameters

  • sqlspec/adapters/psycopg/: Both sync and async, multiple parameter styles

  • sqlspec/adapters/oracledb/: Named parameters, complex type conversion

Special Features:

  • sqlspec/adapters/bigquery/: Cloud database with authentication

  • sqlspec/adapters/adbc/: ADBC protocol, Arrow-native results

Summary Checklist

Before submitting your adapter, verify:



Configuration class inherits from SQLConfig



Driver class inherits from AsyncDriverAdapterBase or SyncDriverAdapterBase



StatementConfig correctly defines parameter styles



MANDATORY parameter parsing flow is implemented



Exception handler maps database errors to SQLSpec exceptions



All abstract methods are implemented (begin, commit, rollback, etc.)



Data dictionary is implemented for metadata queries



Comprehensive tests cover CRUD, transactions, parameters, and types



Public API is exported in __init__.py



Documentation includes usage examples



Code follows SQLSpec standards (see CLAUDE.md)



Type hints are stringified for non-builtin types



No from __future__ import annotations



Logging uses sqlspec.utils.logging

Further Resources