API Reference

Complete API reference for the aiosql integration.

SQLFileLoader

class sqlspec.loader.SQLFileLoader[source]

Bases: object

Loads and parses SQL files with aiosql-style named queries.

Loads SQL files containing named queries (using – name: syntax) and retrieves them by name.

__init__(*, encoding='utf-8', storage_registry=None, runtime=None)[source]

Initialize the SQL file loader.

Parameters:

  • encoding (str) – Text encoding for reading SQL files.

  • storage_registry (StorageRegistry | None) – Storage registry for handling file URIs.

  • runtime (ObservabilityRuntime | None) – Observability runtime for instrumentation.

encoding
storage_registry
set_observability_runtime(runtime)[source]

Attach an observability runtime used for instrumentation.

Return type:

None

load_sql(*paths)[source]

Load SQL files and parse named queries.

Parameters:

*paths (str | Path) – One or more file paths or directory paths to load.

Return type:

None

add_named_sql(name, sql, dialect=None)[source]

Add a named SQL query directly without loading from a file.

Parameters:

  • name (str) – Name for the SQL query.

  • sql (str) – Raw SQL content.

  • dialect (str | None) – Optional dialect for the SQL statement.

Raises:

ValueError – If query name already exists.

Return type:

None

get_file(path)[source]

Get a loaded SQLFile object by path.

Parameters:

path (str | Path) – Path of the file.

Return type:

SQLFile | None

Returns:

SQLFile object if loaded, None otherwise.

get_file_for_query(name)[source]

Get the SQLFile object containing a query.

Parameters:

name (str) – Query name (hyphens are converted to underscores).

Return type:

SQLFile | None

Returns:

SQLFile object if query exists, None otherwise.

list_queries()[source]

List all available query names.

Return type:

list[str]

Returns:

Sorted list of query names.

list_files()[source]

List all loaded file paths.

Return type:

list[str]

Returns:

Sorted list of file paths.

has_query(name)[source]

Check if a query exists.

Parameters:

name (str) – Query name to check.

Return type:

bool

Returns:

True if query exists.

clear_cache()[source]

Clear all cached files and queries.

Return type:

None

clear_file_cache()[source]

Clear the file cache only, keeping loaded queries.

Return type:

None

get_query_text(name)[source]

Get raw SQL text for a query.

Parameters:

name (str) – Query name.

Return type:

str

Returns:

Raw SQL text.

Raises:

SQLFileNotFoundError – If query not found.

get_sql(name)[source]

Get a SQL object by statement name.

Parameters:

name (str) – Name of the statement (from – name: in SQL file). Hyphens in names are converted to underscores.

Return type:

SQL

Returns:

SQL object ready for execution.

Raises:

SQLFileNotFoundError – If statement name not found.

For complete SQLFileLoader documentation, see Base.

aiosql Adapters

AiosqlAsyncAdapter

class sqlspec.extensions.aiosql.AiosqlAsyncAdapter[source]

Bases: _AiosqlAdapterBase[AsyncDriverAdapterBase], AsyncDriverAdapterProtocol

Asynchronous adapter that implements aiosql protocol using SQLSpec drivers.

This adapter bridges aiosql’s async driver protocol with SQLSpec’s async drivers, enabling queries loaded by aiosql to be executed with SQLSpec async drivers.

is_aio_driver: ClassVar[bool] = True
__init__(driver)[source]

Initialize the async adapter.

Parameters:

driver (AsyncDriverAdapterBase) – SQLSpec async driver to use for execution

async select(conn, query_name, sql, parameters, record_class=None)[source]

Execute a SELECT query and return results as list.

Parameters:

  • conn (Any) – Database connection

  • query_name (str) – Name of the query

  • sql (str) – SQL string

  • parameters (Union[Dict[str, Any], List[Any], None]) – Query parameters

  • record_class (Any | None) – Deprecated - use schema_type in driver.execute instead

Return type:

list[Any]

Returns:

List of query result rows

Note

The record_class parameter is ignored for compatibility. Use schema_type in driver.execute or _sqlspec_schema_type in parameters for type mapping.

async select_one(conn, query_name, sql, parameters, record_class=None)[source]

Execute a SELECT query and return first result.

Parameters:

  • conn (Any) – Database connection

  • query_name (str) – Name of the query

  • sql (str) – SQL string

  • parameters (Union[Dict[str, Any], List[Any], None]) – Query parameters

  • record_class (Any | None) – Deprecated - use schema_type in driver.execute instead

Return type:

Any | None

Returns:

First result row or None

Note

The record_class parameter is ignored for compatibility. Use schema_type in driver.execute or _sqlspec_schema_type in parameters for type mapping.

async select_value(conn, query_name, sql, parameters)[source]

Execute a SELECT query and return first value of first row.

Parameters:
Return type:

Any | None

Returns:

First value of first row or None

async select_cursor(conn, query_name, sql, parameters)[source]

Execute a SELECT query and return cursor context manager.

Parameters:
Yields:

Cursor-like object with results

Return type:

_AsyncCursorContextManager[typing.Any]

async insert_update_delete(conn, query_name, sql, parameters)[source]

Execute INSERT/UPDATE/DELETE.

Parameters:
Return type:

None

Note

Returns None per aiosql async protocol

async insert_update_delete_many(conn, query_name, sql, parameters)[source]

Execute INSERT/UPDATE/DELETE with many parameter sets.

Parameters:

  • conn (Any) – Database connection

  • query_name (str) – Name of the query

  • sql (str) – SQL string

  • parameters (Union[Dict[str, Any], List[Any], None]) – Sequence of parameter sets

Return type:

None

Note

Returns None per aiosql async protocol

async insert_returning(conn, query_name, sql, parameters)[source]

Execute INSERT with RETURNING and return result.

Parameters:
Return type:

Any | None

Returns:

Returned value or None

AiosqlSyncAdapter

class sqlspec.extensions.aiosql.AiosqlSyncAdapter[source]

Bases: _AiosqlAdapterBase[SyncDriverAdapterBase], SyncDriverAdapterProtocol

Synchronous adapter that implements aiosql protocol using SQLSpec drivers.

This adapter bridges aiosql’s synchronous driver protocol with SQLSpec’s sync drivers, enabling queries loaded by aiosql to be executed with SQLSpec drivers.

is_aio_driver: ClassVar[bool] = False
__init__(driver)[source]

Initialize the sync adapter.

Parameters:

driver (SyncDriverAdapterBase) – SQLSpec sync driver to use for execution

select(conn, query_name, sql, parameters, record_class=None)[source]

Execute a SELECT query and return results as generator.

Parameters:

  • conn (Any) – Database connection (passed through to SQLSpec driver)

  • query_name (str) – Name of the query

  • sql (str) – SQL string

  • parameters (Union[Dict[str, Any], List[Any], None]) – Query parameters

  • record_class (Any | None) – Deprecated - use schema_type in driver.execute instead

Yields:

Query result rows

Return type:

Generator[Any, None, None]

Note

The record_class parameter is ignored for compatibility. Use schema_type in driver.execute or _sqlspec_schema_type in parameters for type mapping.

select_one(conn, query_name, sql, parameters, record_class=None)[source]

Execute a SELECT query and return first result.

Parameters:

  • conn (Any) – Database connection

  • query_name (str) – Name of the query

  • sql (str) – SQL string

  • parameters (Union[Dict[str, Any], List[Any], None]) – Query parameters

  • record_class (Any | None) – Deprecated - use schema_type in driver.execute instead

Return type:

tuple[Any, ...] | None

Returns:

First result row or None

Note

The record_class parameter is ignored for compatibility. Use schema_type in driver.execute or _sqlspec_schema_type in parameters for type mapping.

select_value(conn, query_name, sql, parameters)[source]

Execute a SELECT query and return first value of first row.

Parameters:
Return type:

Any | None

Returns:

First value of first row or None

select_cursor(conn, query_name, sql, parameters)[source]

Execute a SELECT query and return cursor context manager.

Parameters:
Yields:

Cursor-like object with results

Return type:

Generator[Any, None, None]

insert_update_delete(conn, query_name, sql, parameters)[source]

Execute INSERT/UPDATE/DELETE and return affected rows.

Parameters:
Return type:

int

Returns:

Number of affected rows

insert_update_delete_many(conn, query_name, sql, parameters)[source]

Execute INSERT/UPDATE/DELETE with many parameter sets.

Parameters:

  • conn (Any) – Database connection

  • query_name (str) – Name of the query

  • sql (str) – SQL string

  • parameters (Union[Dict[str, Any], List[Any], None]) – Sequence of parameter sets

Return type:

int

Returns:

Number of affected rows

insert_returning(conn, query_name, sql, parameters)[source]

Execute INSERT with RETURNING and return result.

Parameters:
Return type:

Any | None

Returns:

Returned value or None

Query Operators

The aiosql adapter supports all aiosql query operators:

Operator

Meaning

Returns

(none)

Select many

List of rows

^

Select one

Single row or None

$

Select value

Single value or None

!

Insert/Update/Delete

Rows affected (sync) / None (async)

*!

Insert/Update/Delete many

Rows affected (sync) / None (async)

#

Script

None

Usage Examples

SQLFileLoader Example

Direct usage of SQLFileLoader (for advanced use cases):

from sqlspec.loader import SQLFileLoader

# Create and load
loader = SQLFileLoader()
loader.load_sql("queries/")

# Get query
query = loader.get_sql("get_user")

# Execute with parameters
result = await session.execute(query, user_id=1)
user = result.one()

Recommended usage via SQLSpec:

from sqlspec import SQLSpec

spec = SQLSpec()
spec.load_sql_files("queries/")

# Get query
query = spec.get_sql("get_user")

# Execute with parameters
async with spec.provide_session(config) as session:
    result = await session.execute(query, user_id=1)
    user = result.one()

aiosql Adapter Example (Async)

import aiosql
from sqlspec.extensions.aiosql import AiosqlAsyncAdapter

# Create adapter
adapter = AiosqlAsyncAdapter(driver)

# Load queries
queries = aiosql.from_path("queries.sql", adapter)

# Execute
user = await queries.get_user(conn, user_id=1)

aiosql Adapter Example (Sync)

import aiosql
from sqlspec.extensions.aiosql import AiosqlSyncAdapter

# Create adapter
adapter = AiosqlSyncAdapter(driver)

# Load queries
queries = aiosql.from_path("queries.sql", adapter)

# Execute
user = queries.get_user(conn, user_id=1)

Type Aliases

Common imports:

# SQLFileLoader

# aiosql adapters

# For type hints

See Also