aiosql package

Subpackages

aiosql.adapters package

Submodules

aiosql.aiosql module

aiosql.aiosql.from_path(sql_path: str | ~pathlib.Path, driver_adapter: str | ~typing.Callable[[...], ~aiosql.types.SyncDriverAdapterProtocol | ~aiosql.types.AsyncDriverAdapterProtocol], record_classes: ~typing.Dict | None = None, kwargs_only: bool = True, attribute: str | None = '__', args: ~typing.List[~typing.Any] = [], kwargs: ~typing.Dict[str, ~typing.Any] = {}, loader_cls: ~typing.Type[~aiosql.query_loader.QueryLoader] = <class 'aiosql.query_loader.QueryLoader'>, queries_cls: ~typing.Type[~aiosql.queries.Queries] = <class 'aiosql.queries.Queries'>, ext: ~typing.Tuple[str] = ('.sql',), encoding=None)

Load queries from a .sql file, or directory of .sql files.

Parameters:

sql_path - Path to a .sql file or directory containing .sql files.
driver_adapter - Either a string to designate one of the aiosql built-in database driver adapters. One of many available for SQLite, Postgres and MySQL. If you have defined your own adapter class, you may pass its constructor.
record_classes - (optional) DEPRECATED Mapping of strings used in “record_class”
kwargs_only - (optional) Whether to only use named parameters on query execution, default is True.
attribute - (optional) . attribute access substitution, defaults to "__"", None disables the feature.
args - (optional) adapter creation args (list), forwarded to cursor creation by default.
kwargs - (optional) adapter creation args (dict), forwarded to cursor creation by default. declarations to the python classes which aiosql should use when marshaling SQL results.
loader_cls - (optional) Custom constructor for QueryLoader extensions.
queries_cls - (optional) Custom constructor for Queries extensions.
ext - (optional) allowed file extensions for query files, default is (“.sql”,).
encoding - (optional) encoding for reading files.

Returns: Queries

Usage:

queries = aiosql.from_path("./sql", "psycopg2")
queries = aiosql.from_path("./sql", MyDBAdapter)

aiosql.aiosql.from_str(sql: str, driver_adapter: str | ~typing.Callable[[...], ~aiosql.types.SyncDriverAdapterProtocol | ~aiosql.types.AsyncDriverAdapterProtocol], record_classes: ~typing.Dict | None = None, kwargs_only: bool = True, attribute: str | None = '__', args: ~typing.List[~typing.Any] = [], kwargs: ~typing.Dict[str, ~typing.Any] = {}, loader_cls: ~typing.Type[~aiosql.query_loader.QueryLoader] = <class 'aiosql.query_loader.QueryLoader'>, queries_cls: ~typing.Type[~aiosql.queries.Queries] = <class 'aiosql.queries.Queries'>)

Load queries from a SQL string.

Parameters:

sql - A string containing SQL statements and aiosql name.
driver_adapter - Either a string to designate one of the aiosql built-in database driver adapters. One of many available for SQLite, Postgres and MySQL. If you have defined your own adapter class, you can pass it’s constructor.
kwargs_only - (optional) whether to only use named parameters on query execution, default is True.
attribute - (optional) . attribute access substitution, defaults to "__", None disables the feature.
args - (optional) adapter creation args (list), forwarded to cursor creation by default.
kwargs - (optional) adapter creation args (dict), forwarded to cursor creation by default.
record_classes - (optional) DEPRECATED Mapping of strings used in “record_class” declarations to the python classes which aiosql should use when marshaling SQL results.
loader_cls - (optional) Custom constructor for QueryLoader extensions.
queries_cls - (optional) Custom constructor for Queries extensions.

Returns: Queries

Usage:

Loading queries from a SQL string.

import sqlite3
import aiosql

sql_text = """
-- name: get-all-greetings
-- Get all the greetings in the database
select * from greetings;

-- name: get-user-by-username^
-- Get all the users from the database,
-- and return it as a dict
select * from users where username = :username;
"""

queries = aiosql.from_str(sql_text, "sqlite3")
queries.get_all_greetings(conn)
queries.get_user_by_username(conn, username="willvaughn")

aiosql.aiosql.register_adapter(name: str, adapter: Callable[[...], SyncDriverAdapterProtocol | AsyncDriverAdapterProtocol]): Register or override an adapter.

aiosql.queries module

class aiosql.queries.Queries(driver_adapter: SyncDriverAdapterProtocol | AsyncDriverAdapterProtocol, kwargs_only: bool = True)

Bases: object

Container object with dynamic methods built from SQL queries.

The -- name: definition comments in the content of the SQL determine what the dynamic methods of this class will be named.

Much of the needed pre-processing is performed in QueryLoader.

Parameters:

param driver_adapter:

Either a string to designate one of the aiosql built-in database driver

adapters (e.g. “sqlite3”, “psycopg”). If you have defined your own adapter class, you can pass its constructor.
param kwargs_only:

whether to reject positional parameters, defaults to true.

add_child_queries(child_name: str, child_queries: Queries) → None

Adds a Queries object as a property.

Parameters:

child_name - The property name to group the child queries under.
child_queries - Queries instance to add as sub-queries.

add_queries(queries: List[QueryFn]) → None: Add query methods to Queries instance.

add_query(query_name: str, fn: Callable) → None

Adds a new dynamic method to this class.

Parameters:

query_name - The method name as found in the SQL content.
fn - The loaded query function.

property available_queries: List[str]

Returns listing of all the available query methods loaded in this class.

Returns: list[str] List of dot-separated method accessor names.

load_from_list(query_data: List[QueryDatum]): Load Queries from a list of QueryDatum

load_from_tree(query_data_tree: Dict[str, QueryDatum | Dict]): Load Queries from a QueryDataTree

aiosql.query_loader module

class aiosql.query_loader.QueryLoader(driver_adapter: SyncDriverAdapterProtocol | AsyncDriverAdapterProtocol, record_classes: Dict[str, Any] | None, attribute: str | None = None)

Bases: object

Load Queries.

This class holds the various utilities to read SQL files and build QueryDatum, which will be transformed as functions in Queries.

param driver_adapter:

driver name or class.
param record_classes:

nothing of dict.
param attribute:

string to insert in place of ..

load_query_data_from_dir_path(dir_path, ext=('.sql',), encoding=None) → Dict[str, QueryDatum | Dict]: Load queries from a directory.

load_query_data_from_file(path: Path, ns_parts: List[str] = [], encoding=None) → List[QueryDatum]: Load queries from a file.

load_query_data_from_sql(sql: str, ns_parts: List[str], fname: Path | str = '<unknown>') → List[QueryDatum]: Load queries from a string.

aiosql.types module

class aiosql.types.AsyncDriverAdapterProtocol(*args, **kwargs)

Bases: Protocol

async execute_script(conn: Any, sql: str) → str

async insert_returning(conn: Any, query_name: str, sql: str, parameters: Dict[str, Any] | List[Any] | None) → Any | None

async insert_update_delete(conn: Any, query_name: str, sql: str, parameters: Dict[str, Any] | List[Any] | None) → None

async insert_update_delete_many(conn: Any, query_name: str, sql: str, parameters: Dict[str, Any] | List[Any] | None) → None

process_sql(query_name: str, op_type: SQLOperationType, sql: str) → str

async select(conn: Any, query_name: str, sql: str, parameters: Dict[str, Any] | List[Any] | None, record_class: Callable | None) → List

async select_cursor(conn: Any, query_name: str, sql: str, parameters: Dict[str, Any] | List[Any] | None) → AsyncContextManager[Any]

async select_one(conn: Any, query_name: str, sql: str, parameters: Dict[str, Any] | List[Any] | None, record_class: Callable | None) → Any | None

async select_value(conn: Any, query_name: str, sql: str, parameters: Dict[str, Any] | List[Any] | None) → Any | None

class aiosql.types.QueryDatum(query_name, doc_comments, operation_type, sql, record_class, signature, floc, attributes, parameters)

Bases: NamedTuple

attributes: Dict[str, Dict[str, str]] | None: Alias for field number 7

doc_comments: str: Alias for field number 1

floc: Tuple[Path | str, int]: Alias for field number 6

operation_type: SQLOperationType: Alias for field number 2

parameters: List[str] | None: Alias for field number 8

query_name: str: Alias for field number 0

record_class: Any: Alias for field number 4

signature: Signature | None: Alias for field number 5

sql: str: Alias for field number 3

class aiosql.types.QueryFn(*args, **kwargs)

Bases: Protocol

attributes: Dict[str, Dict[str, str]] | None

operation: SQLOperationType

parameters: List[str] | None

sql: str

class aiosql.types.SQLOperationType(*values)

Bases: Enum

Enumeration of aiosql operation types.

INSERT_RETURNING = 0

INSERT_UPDATE_DELETE = 1

INSERT_UPDATE_DELETE_MANY = 2

SCRIPT = 3

SELECT = 4

SELECT_ONE = 5

SELECT_VALUE = 6

class aiosql.types.SyncDriverAdapterProtocol(*args, **kwargs)

Bases: Protocol

execute_script(conn: Any, sql: str) → str

insert_returning(conn: Any, query_name: str, sql: str, parameters: Dict[str, Any] | List[Any] | None) → Any | None

insert_update_delete(conn: Any, query_name: str, sql: str, parameters: Dict[str, Any] | List[Any] | None) → int

insert_update_delete_many(conn: Any, query_name: str, sql: str, parameters: Dict[str, Any] | List[Any] | None) → int

process_sql(query_name: str, op_type: SQLOperationType, sql: str) → str

select(conn: Any, query_name: str, sql: str, parameters: Dict[str, Any] | List[Any] | None, record_class: Callable | None) → Generator[Any, None, None]

select_cursor(conn: Any, query_name: str, sql: str, parameters: Dict[str, Any] | List[Any] | None) → ContextManager[Any]

select_one(conn: Any, query_name: str, sql: str, parameters: Dict[str, Any] | List[Any] | None, record_class: Callable | None) → Tuple[Any, ...] | None

select_value(conn: Any, query_name: str, sql: str, parameters: Dict[str, Any] | List[Any] | None) → Any | None

aiosql.utils module

exception aiosql.utils.SQLLoadException

Bases: Exception

Raised when there is a problem loading SQL content from a file or directory

exception aiosql.utils.SQLParseException

Bases: Exception

Raised when there was a problem parsing the aiosql comment annotations in SQL

aiosql.utils.VAR_REF = re.compile('(?P<dquote>"(""|[^"])+")|(?P<squote>\\\'(\\\'\\\'|[^\\\'])*\\\')|(?P<lead>[^:]):(?P<var_name>\\w+)(?=[^:]?)'): Pattern to identify colon-variables (aka _named_ style) in SQL code

aiosql.utils.VAR_REF_DOT = re.compile('(?P<dquote>"(""|[^"])+")|(?P<squote>\\\'(\\\'\\\'|[^\\\'])*\\\')|(?P<lead>[^:]):(?P<var_name>\\w+\\.\\w+)(?=[^:]?)'): Pattern to identify colon-variables with a simple attribute in SQL code.

aiosql.utils.log = <Logger aiosql (WARNING)>: Shared package logging.

Module contents

exception aiosql.SQLLoadException

Bases: Exception

Raised when there is a problem loading SQL content from a file or directory

exception aiosql.SQLParseException

Bases: Exception

Raised when there was a problem parsing the aiosql comment annotations in SQL

aiosql.from_path(sql_path: str | ~pathlib.Path, driver_adapter: str | ~typing.Callable[[...], ~aiosql.types.SyncDriverAdapterProtocol | ~aiosql.types.AsyncDriverAdapterProtocol], record_classes: ~typing.Dict | None = None, kwargs_only: bool = True, attribute: str | None = '__', args: ~typing.List[~typing.Any] = [], kwargs: ~typing.Dict[str, ~typing.Any] = {}, loader_cls: ~typing.Type[~aiosql.query_loader.QueryLoader] = <class 'aiosql.query_loader.QueryLoader'>, queries_cls: ~typing.Type[~aiosql.queries.Queries] = <class 'aiosql.queries.Queries'>, ext: ~typing.Tuple[str] = ('.sql',), encoding=None)

Load queries from a .sql file, or directory of .sql files.

Parameters:

sql_path - Path to a .sql file or directory containing .sql files.
driver_adapter - Either a string to designate one of the aiosql built-in database driver adapters. One of many available for SQLite, Postgres and MySQL. If you have defined your own adapter class, you may pass its constructor.
record_classes - (optional) DEPRECATED Mapping of strings used in “record_class”
kwargs_only - (optional) Whether to only use named parameters on query execution, default is True.
attribute - (optional) . attribute access substitution, defaults to "__"", None disables the feature.
args - (optional) adapter creation args (list), forwarded to cursor creation by default.
kwargs - (optional) adapter creation args (dict), forwarded to cursor creation by default. declarations to the python classes which aiosql should use when marshaling SQL results.
loader_cls - (optional) Custom constructor for QueryLoader extensions.
queries_cls - (optional) Custom constructor for Queries extensions.
ext - (optional) allowed file extensions for query files, default is (“.sql”,).
encoding - (optional) encoding for reading files.

Returns: Queries

Usage:

queries = aiosql.from_path("./sql", "psycopg2")
queries = aiosql.from_path("./sql", MyDBAdapter)

aiosql.from_str(sql: str, driver_adapter: str | ~typing.Callable[[...], ~aiosql.types.SyncDriverAdapterProtocol | ~aiosql.types.AsyncDriverAdapterProtocol], record_classes: ~typing.Dict | None = None, kwargs_only: bool = True, attribute: str | None = '__', args: ~typing.List[~typing.Any] = [], kwargs: ~typing.Dict[str, ~typing.Any] = {}, loader_cls: ~typing.Type[~aiosql.query_loader.QueryLoader] = <class 'aiosql.query_loader.QueryLoader'>, queries_cls: ~typing.Type[~aiosql.queries.Queries] = <class 'aiosql.queries.Queries'>)

Load queries from a SQL string.

Parameters:

sql - A string containing SQL statements and aiosql name.
driver_adapter - Either a string to designate one of the aiosql built-in database driver adapters. One of many available for SQLite, Postgres and MySQL. If you have defined your own adapter class, you can pass it’s constructor.
kwargs_only - (optional) whether to only use named parameters on query execution, default is True.
attribute - (optional) . attribute access substitution, defaults to "__", None disables the feature.
args - (optional) adapter creation args (list), forwarded to cursor creation by default.
kwargs - (optional) adapter creation args (dict), forwarded to cursor creation by default.
record_classes - (optional) DEPRECATED Mapping of strings used in “record_class” declarations to the python classes which aiosql should use when marshaling SQL results.
loader_cls - (optional) Custom constructor for QueryLoader extensions.
queries_cls - (optional) Custom constructor for Queries extensions.

Returns: Queries

Usage:

Loading queries from a SQL string.

import sqlite3
import aiosql

sql_text = """
-- name: get-all-greetings
-- Get all the greetings in the database
select * from greetings;

-- name: get-user-by-username^
-- Get all the users from the database,
-- and return it as a dict
select * from users where username = :username;
"""

queries = aiosql.from_str(sql_text, "sqlite3")
queries.get_all_greetings(conn)
queries.get_user_by_username(conn, username="willvaughn")

aiosql.register_adapter(name: str, adapter: Callable[[...], SyncDriverAdapterProtocol | AsyncDriverAdapterProtocol]): Register or override an adapter.