aiosql package

Subpackages

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 = False, *, 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.

  • kwargs_only - Whether to only use named parameters on query execution.

  • 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.

  • 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 = False, *, 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 - Whether to only use named parameters on query execution.

  • 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 = False)

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.

Parameters:

  • 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.

  • kwargs_only: whether to reject positional parameters.

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 QuaryDatum

load_from_tree(query_data_tree: Dict[str, QueryDatum | Dict])

Load Queries from a QuaryDataTree

aiosql.query_loader module

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

Bases: object

load_query_data_from_dir_path(dir_path, ext=('.sql',), encoding=None) Dict[str, QueryDatum | Dict]
load_query_data_from_file(path: Path, ns_parts: List[str] = [], encoding=None) List[QueryDatum]
load_query_data_from_sql(sql: str, ns_parts: List[str], fname: Path | str = '<unknown>') List[QueryDatum]

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: List | Dict) Any | None
async insert_update_delete(conn: Any, query_name: str, sql: str, parameters: List | Dict) None
async insert_update_delete_many(conn: Any, query_name: str, sql: str, parameters: List | Dict) None
process_sql(query_name: str, op_type: SQLOperationType, sql: str) str
async select(conn: Any, query_name: str, sql: str, parameters: List | Dict, record_class: Callable | None) List
async select_cursor(conn: Any, query_name: str, sql: str, parameters: List | Dict) AsyncContextManager[Any]
async select_one(conn: Any, query_name: str, sql: str, parameters: List | Dict, record_class: Callable | None) Any | None
async select_value(conn: Any, query_name: str, sql: str, parameters: List | Dict) Any | None
class aiosql.types.QueryDatum(query_name, doc_comments, operation_type, sql, record_class, signature, floc)

Bases: NamedTuple

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

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

operation: SQLOperationType
sql: str
class aiosql.types.SQLOperationType(value, names=None, *values, module=None, qualname=None, type=None, start=1, boundary=None)

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: List | Dict) Any | None
insert_update_delete(conn: Any, query_name: str, sql: str, parameters: List | Dict) int
insert_update_delete_many(conn: Any, query_name: str, sql: str, parameters: List | Dict) int
process_sql(query_name: str, op_type: SQLOperationType, sql: str) str
select(conn: Any, query_name: str, sql: str, parameters: List | Dict, record_class: Callable | None) List
select_cursor(conn: Any, query_name: str, sql: str, parameters: List | Dict) ContextManager[Any]
select_one(conn: Any, query_name: str, sql: str, parameters: List | Dict, record_class: Callable | None) Any | None
select_value(conn: Any, query_name: str, sql: str, parameters: List | Dict) 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 identifies colon-variables (aka _named_ style) in SQL code

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 = False, *, 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.

  • kwargs_only - Whether to only use named parameters on query execution.

  • 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.

  • 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 = False, *, 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 - Whether to only use named parameters on query execution.

  • 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.