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, *, 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',))

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” 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”,)

Returns: Queries

Usage:

`python >>> 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, *, 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. * 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.

```python 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)

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. One of “sqlite3”, “psycopg2”, “aiosqlite”, or “asyncpg”. If you have defined your own adapter class, you can pass it’s constructor.

add_child_queries(child_name: str, child_queries: Queries)

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])

Add query methods to Queries instance.

add_query(query_name: str, fn: Callable)

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',)) → Dict[str, QueryDatum | Dict]
load_query_data_from_file(path: Path, ns_parts: List[str] = []) → List[QueryDatum]
load_query_data_from_sql(sql: str, ns_parts: List[str] = [], fname: Path | None = None) → 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, int] | None

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, *, 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 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, *, 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',))

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” 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”,)

Returns: Queries

Usage:

`python >>> 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, *, 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. * 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.

```python 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.