asyncmy#

Async MySQL adapter using asyncmy, an asyncio-native MySQL driver written in Cython. Provides asynchronous execution, connection pooling, and PyMySQL-compatible wire protocol handling.

Configuration#

class sqlspec.adapters.asyncmy.AsyncmyConfig[source]#

Bases: AsyncDatabaseConfig[Connection, AsyncmyPool, AsyncmyDriver]

Configuration for Asyncmy database connections.

driver_type#: alias of AsyncmyDriver

__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)[source]#

Initialize Asyncmy configuration.

Parameters:

connection_config¶ -- Connection and pool configuration parameters
connection_instance¶ -- Existing pool instance to use
migration_config¶ -- Migration configuration
statement_config¶ -- Statement configuration override
driver_features¶ -- Driver feature configuration (TypedDict or dict)
bind_key¶ -- Optional unique identifier for this configuration
extension_config¶ -- Extension-specific configuration
observability_config¶ -- Adapter-level observability overrides for lifecycle hooks and observers
**kwargs¶ -- Additional keyword arguments

async create_connection()[source]#

Open a standalone connection owned by the caller.

The connection carries the same connection settings and creation hook the pool applies, consumes no pool slot, and must be closed by the caller.

Return type:: Connection
Returns:: An Asyncmy connection instance.

async provide_pool(*args, **kwargs)[source]#

Provide async pool instance.

Return type:: Pool
Returns:: The async connection pool.

get_signature_namespace()[source]#

Get the signature namespace for Asyncmy types.

Return type:: dict[str, typing.Any]
Returns:: Dictionary mapping type names to types.

get_event_runtime_hints()[source]#

Return queue polling defaults for Asyncmy adapters.

Return type:: EventRuntimeHints

Connection Parameters#

class sqlspec.adapters.asyncmy.AsyncmyConnectionParams[source]#

Bases: TypedDict

Asyncmy connection parameters.

class sqlspec.adapters.asyncmy.AsyncmySSLParams[source]#

Bases: TypedDict

Asyncmy TLS parameters.

Pool Parameters#

class sqlspec.adapters.asyncmy.AsyncmyPoolParams[source]#

Bases: AsyncmyConnectionParams

Asyncmy pool parameters.

Driver Features#

class sqlspec.adapters.asyncmy.AsyncmyDriverFeatures[source]#

Bases: TypedDict

Asyncmy driver feature flags.

MySQL/MariaDB handle JSON natively, but custom serializers can be provided for specialized use cases.

enable_local_infile_bulk_load: Use native LOCAL INFILE for eligible Arrow rows.: Defaults to the connection's local_infile or allow_local_infile opt-in. Set False to force executemany on an opted-in connection.
json_serializer: Custom JSON serializer function.: Defaults to sqlspec.utils.serializers.to_json. Use for performance (orjson) or custom encoding.
json_deserializer: Custom JSON deserializer function.: Defaults to sqlspec.utils.serializers.from_json. Use for performance (orjson) or custom decoding.
on_connection_create: Async callback executed when a connection is acquired from pool.: Receives the raw asyncmy connection for low-level driver configuration. Called exactly once per physical connection using WeakSet tracking.
enable_events: Enable database event channel support.: Defaults to True when extension_config["events"] is configured. Provides pub/sub capabilities via table-backed queue (MySQL/MariaDB have no native pub/sub). Requires extension_config["events"] for migration setup.
events_backend: Event channel backend selection.: Only option: "poll_queue" (durable table-backed queue with lease-based retries and acknowledgements). MySQL/MariaDB do not have native pub/sub, so poll_queue is the only backend. Defaults to "poll_queue".

Driver#

class sqlspec.adapters.asyncmy.AsyncmyDriver[source]#

Bases: AsyncDriverAdapterBase

MySQL/MariaDB database driver using AsyncMy client library.

Implements asynchronous database operations for MySQL and MariaDB servers with support for parameter style conversion, type coercion, error handling, and transaction management.

__init__(connection, statement_config=None, driver_features=None)[source]#

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

async dispatch_execute(cursor, statement)[source]#

Execute single SQL statement.

Handles parameter processing, result fetching, and data transformation for MySQL/MariaDB operations.

Parameters:

cursor¶ (Any) -- AsyncMy cursor object
statement¶ (SQL) -- SQL statement to execute

Returns:

Statement execution results with data or row counts

Return type:

ExecutionResult

async dispatch_execute_many(cursor, statement)[source]#

Execute SQL statement with multiple parameter sets.

Uses AsyncMy's executemany for batch operations with MySQL type conversion and parameter processing.

Parameters:

cursor¶ (Any) -- AsyncMy cursor object
statement¶ (SQL) -- SQL statement with multiple parameter sets

Returns:

Batch execution results

Return type:

ExecutionResult

async dispatch_execute_script(cursor, statement)[source]#

Execute SQL script with statement splitting and parameter handling.

Splits multi-statement scripts and executes each statement sequentially. Parameters are embedded as static values for script execution compatibility.

Parameters:

cursor¶ (Any) -- AsyncMy cursor object
statement¶ (SQL) -- SQL script to execute

Returns:

Script execution results with statement count

Return type:

ExecutionResult

async begin()[source]#

Begin a database transaction.

Explicitly starts a MySQL transaction to ensure proper transaction boundaries.

Raises:: SQLSpecError -- If transaction initialization fails
Return type:: None

async commit()[source]#

Commit the current transaction.

Raises:: SQLSpecError -- If transaction commit fails
Return type:: None

async rollback()[source]#

Rollback the current transaction.

Raises:: SQLSpecError -- If transaction rollback fails
Return type:: None

with_cursor(connection)[source]#

Create cursor context manager for the connection.

Parameters:: connection¶ (Connection) -- AsyncMy database connection
Returns:: Context manager for cursor operations
Return type:: AsyncmyCursor

dispatch_select_stream(statement, chunk_size)[source]#

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

Return type:: Optional[AsyncRowStream[dict[str, typing.Any]]]

handle_database_exceptions()[source]#

Provide exception handling context manager.

Returns:: Context manager for AsyncMy exception handling
Return type:: AsyncmyExceptionHandler

async select_to_storage(statement, destination, /, *parameters, statement_config=None, partitioner=None, format_hint=None, telemetry=None, **kwargs)[source]#

Execute a query and stream Arrow-formatted results into storage.

Return type:: StorageBridgeJob

async load_from_arrow(table, source, *, partitioner=None, overwrite=False, telemetry=None)[source]#

Load Arrow data using batched inserts or opt-in native LOCAL INFILE.

Return type:: StorageBridgeJob

async load_from_storage(table, source, *, file_format, partitioner=None, overwrite=False)[source]#

Load staged artifacts from storage into MySQL.

Return type:: StorageBridgeJob

property data_dictionary: AsyncmyDataDictionary#

Get the data dictionary for this driver.

Returns:: Data dictionary instance for metadata queries

collect_rows(cursor, fetched)[source]#

Collect asyncmy rows for the direct execution path.

Return type:: tuple[list[typing.Any], list[str], int]

resolve_rowcount(cursor)[source]#

Resolve rowcount from asyncmy cursor for the direct execution path.

Return type:: int

Data Dictionary#

class sqlspec.adapters.asyncmy.data_dictionary.AsyncmyDataDictionary[source]#

Bases: MySQLAsyncDataDictionary

MySQL-specific async data dictionary for asyncmy.

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.asyncmy.litestar.AsyncmyLitestarConfig[source]#

Bases: LitestarConfig

Asyncmy-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.asyncmy.adk.AsyncmyADKConfig[source]#

Bases: ADKConfig

asyncmy-specific ADK extension settings.

Use these keys inside extension_config["adk"] with the asyncmy 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.