PyMySQL

Pure-Python MySQL driver for sync usage.

Configuration

class sqlspec.adapters.pymysql.PyMysqlConfig

Bases: SyncDatabaseConfig[Connection, PyMysqlConnectionPool, PyMysqlDriver]

Configuration for PyMySQL synchronous connections.

driver_type

alias of PyMysqlDriver

__init__(*, connection_config=None, connection_instance=None, migration_config=None, statement_config=None, driver_features=None, bind_key=None, extension_config=None, observability_config=None, **kwargs)

create_connection()

Open a standalone connection owned by the caller.

The connection carries the same parameters and creation hook the pool applies, but it is not the pool's thread-local connection, so closing it leaves the pool usable.

Returns:

A newly opened connection.

Return type:

Connection

get_signature_namespace()

Get the signature namespace for this database configuration.

Returns a dictionary of type names to objects (classes, functions, or other callables) that should be registered with Litestar's signature namespace to prevent serialization attempts on database-specific structures.

Return type:

dict[str, typing.Any]

Returns:

Dictionary mapping type names to objects.

get_event_runtime_hints()

Return default event runtime hints for this configuration.

Return type:

EventRuntimeHints

Cloud SQL Connector

PyMySQL configs can use the in-process Google Cloud SQL Python Connector by installing the cloud-sql extra and enabling the connector in driver_features:

from sqlspec.adapters.pymysql import PyMysqlConfig

config = PyMysqlConfig(
    connection_config={
        "user": "app-user",
        "password": "secret",
        "database": "app",
    },
    driver_features={
        "enable_cloud_sql": True,
        "cloud_sql_instance": "project:region:instance",
        "cloud_sql_ip_type": "PRIVATE",
    },
)

When enable_cloud_sql is true, cloud_sql_instance is required and must use project:region:instance format. Host, port, socket, and direct auth connection values are passed through the connector rather than opened directly by PyMySQL.

Driver

class sqlspec.adapters.pymysql.PyMysqlDriver

Bases: SyncDriverAdapterBase

MySQL/MariaDB database driver using PyMySQL.

__init__(connection, statement_config=None, driver_features=None)

Initialize driver adapter with connection and configuration.

Parameters:

• connection (Connection) -- Database connection instance

• statement_config (StatementConfig | None) -- Statement configuration for the driver

• driver_features (dict[str, typing.Any] | None) -- Driver-specific features like extensions, secrets, and connection callbacks

• observability -- Optional runtime handling lifecycle hooks, observers, and spans

dispatch_execute(cursor, statement)

Execute a single SQL statement.

Must be implemented by each driver for database-specific execution logic.

Parameters:

• cursor (Any) -- Database cursor/connection object

• statement (SQL) -- SQL statement object with all necessary data and configuration

Return type:

ExecutionResult

Returns:

ExecutionResult with execution data

dispatch_execute_many(cursor, statement)

Execute SQL with multiple parameter sets (executemany).

Must be implemented by each driver for database-specific executemany logic.

Parameters:

• cursor (Any) -- Database cursor/connection object

• statement (SQL) -- SQL statement object with all necessary data and configuration

Return type:

ExecutionResult

Returns:

ExecutionResult with execution data for the many operation

dispatch_execute_script(cursor, statement)

Execute a SQL script containing multiple statements.

Default implementation splits the script and executes statements individually. Drivers can override for database-specific script execution methods.

Parameters:

• cursor (Any) -- Database cursor/connection object

• statement (SQL) -- SQL statement object with all necessary data and configuration

Return type:

ExecutionResult

Returns:

ExecutionResult with script execution data including statement counts

begin()

Begin a database transaction on the current connection.

Return type:

None

commit()

Commit the current transaction on the current connection.

Return type:

None

rollback()

Rollback the current transaction on the current connection.

Return type:

None

with_cursor(connection)

Create and return a context manager for cursor acquisition and cleanup.

Returns a context manager that yields a cursor for database operations. Concrete implementations handle database-specific cursor creation and cleanup.

Return type:

PyMysqlCursor

dispatch_select_stream(statement, chunk_size)

Return a native PyMySQL row stream backed by an unbuffered SSCursor.

Return type:

Optional[SyncRowStream[dict[str, typing.Any]]]

handle_database_exceptions()

Handle database-specific exceptions and wrap them appropriately.

Return type:

PyMysqlExceptionHandler

Returns:

Exception handler with deferred exception pattern for mypyc compatibility. The handler stores mapped exceptions in pending_exception rather than raising from __exit__ to avoid ABI boundary violations.

select_to_storage(statement, destination, /, *parameters, statement_config=None, partitioner=None, format_hint=None, telemetry=None, **kwargs)

Stream a SELECT statement directly into storage.

Parameters:

• statement (SQL | str) -- SQL statement to execute.

• destination (str | Path) -- Storage destination path.

• parameters (Any) -- Query parameters.

• statement_config (StatementConfig | None) -- Optional statement configuration.

• partitioner (dict[str, object] | None) -- Optional partitioner configuration.

• format_hint (Optional[Literal['jsonl', 'json', 'parquet', 'arrow-ipc', 'csv']]) -- Optional format hint for storage.

• telemetry (StorageTelemetry | None) -- Optional telemetry dict to merge.

Return type:

StorageBridgeJob

Returns:

StorageBridgeJob with execution telemetry.

load_from_arrow(table, source, *, partitioner=None, overwrite=False, telemetry=None)

Load Arrow data into the target table.

Parameters:

• table (str) -- Target table name.

• source (Union[ArrowResult, typing.Any]) -- Arrow data source.

• partitioner (dict[str, object] | None) -- Optional partitioner configuration.

• overwrite (bool) -- Whether to overwrite existing data.

Return type:

StorageBridgeJob

Returns:

StorageBridgeJob with execution telemetry.

load_from_storage(table, source, *, file_format, partitioner=None, overwrite=False)

Load artifacts from storage into the target table.

Parameters:

• table (str) -- Target table name.

• source (str | Path) -- Storage source path.

• file_format (Literal['jsonl', 'json', 'parquet', 'arrow-ipc', 'csv']) -- File format of source.

• partitioner (dict[str, object] | None) -- Optional partitioner configuration.

• overwrite (bool) -- Whether to overwrite existing data.

Return type:

StorageBridgeJob

Returns:

StorageBridgeJob with execution telemetry.

property data_dictionary: PyMysqlDataDictionary

Get the data dictionary for this driver.

Returns:

Data dictionary instance for metadata queries

collect_rows(cursor, fetched)

Collect PyMySQL rows for the direct execution path.

Return type:

tuple[list[typing.Any], list[str], int]

resolve_rowcount(cursor)

Resolve rowcount from PyMySQL cursor for the direct execution path.

Return type:

int

Connection Parameters

class sqlspec.adapters.pymysql.PyMysqlConnectionParams

Bases: TypedDict

PyMySQL connection parameters.

class sqlspec.adapters.pymysql.config.PyMysqlSslParams

Bases: TypedDict

Mapping-style PyMySQL SSL parameters.

Passing an ssl mapping is deprecated by PyMySQL but remains supported for compatibility with existing SQLSpec configs.

Pool Parameters

class sqlspec.adapters.pymysql.PyMysqlPoolParams

Bases: PyMysqlConnectionParams

PyMySQL pool parameters.

Driver Features

class sqlspec.adapters.pymysql.PyMysqlDriverFeatures

Bases: TypedDict

PyMySQL driver feature flags.

json_serializer: Custom JSON serializer function.

Defaults to sqlspec.utils.serializers.to_json.

json_deserializer: Custom JSON deserializer function.

Defaults to sqlspec.utils.serializers.from_json.

on_connection_create: Callback executed when a connection is created.

Receives the raw pymysql connection for low-level driver configuration. Runs after connection creation.

enable_events: Enable database event channel support. events_backend: Event channel backend selection. enable_local_infile_bulk_load: Route load_from_arrow through LOAD DATA LOCAL INFILE.

Defaults to the connection's local_infile or allow_local_infile opt-in. Set False to force executemany on an opted-in connection.

enable_cloud_sql: Enable Google Cloud SQL connector integration.

Requires cloud-sql-python-connector package. Defaults to False (explicit opt-in required).

cloud_sql_instance: Cloud SQL instance connection name.

Format: "project:region:instance" Required when enable_cloud_sql is True.

cloud_sql_enable_iam_auth: Enable IAM database authentication.

Defaults to False for passwordless authentication.

cloud_sql_ip_type: IP address type for connection.

Options: "PUBLIC", "PRIVATE", "PSC" Defaults to "PRIVATE".

Connection Pool

class sqlspec.adapters.pymysql.pool.PyMysqlConnectionPool

Bases: object

Thread-local connection manager for PyMySQL.

__init__(connection_parameters, recycle_seconds=86400, health_check_interval=30.0, on_connection_create=None, connection_factory=None)

Initialize the thread-local connection manager.

Parameters:

• connection_parameters (dict[str, typing.Any]) -- PyMySQL connection parameters

• recycle_seconds (int) -- Connection recycle time in seconds (default 24h)

• health_check_interval (float) -- Seconds of idle time before running health check

• on_connection_create (typing.Callable[[pymysql.connections.Connection], None] | None) -- Callback executed when connection is created

• connection_factory (typing.Callable[[], Connection] | None) -- Optional factory for custom connection creation

new_connection()

Open a standalone connection configured like a pooled one.

The result is owned by the caller: it is not thread-local and is not tracked for pool shutdown.

Returns:

A newly opened, fully configured connection.

Return type:

Connection

get_connection()

Get a thread-local connection.

Yields:

A thread-local database connection.

close()

Close every connection this pool opened, on any thread.

Return type:

None

release(connection)

Release connection back to the pool, sanitizing transactions.

Return type:

None

size()

Report total active connections managed by this pool.

Return type:

int

Data Dictionary

class sqlspec.adapters.pymysql.data_dictionary.PyMysqlDataDictionary

Bases: MySQLSyncDataDictionary

MySQL-specific sync data dictionary for pymysql.

dialect: ClassVar[str] = 'mysql'

Dialect identifier. Must be defined by subclasses as a class attribute.

Extension Settings

Use the configuration types below in their corresponding extension_config namespace: "litestar", "events", or "adk" as supported by this adapter.

class sqlspec.adapters.pymysql.litestar.PyMysqlLitestarConfig

Bases: LitestarConfig

PyMysql-specific Litestar settings.

Use inside extension_config["litestar"] with this adapter's session store.

table_options: NotRequired[str]

Table DDL options.

index_options: NotRequired[str]

Index DDL options.

class sqlspec.adapters.pymysql.adk.PyMysqlADKConfig

Bases: ADKConfig

PyMySQL-specific ADK extension settings.

Use these keys inside extension_config["adk"] with the PyMySQL ADK store.

enable_event_generated_columns: NotRequired[bool]

Create MySQL generated columns and indexes for common ADK event JSON paths.

enable_covering_indexes: NotRequired[bool]

Add hot-path payload columns to MySQL ADK event replay indexes.

session_table_options: NotRequired[str]

Raw MySQL table options appended to the ADK session table.

events_table_options: NotRequired[str]

Raw MySQL table options appended to the ADK events table.

app_state_table_options: NotRequired[str]

Raw MySQL table options appended to the ADK app state table.

user_state_table_options: NotRequired[str]

Raw MySQL table options appended to the ADK user state table.

memory_table_options: NotRequired[str]

Raw MySQL table options appended to the ADK memory table.