Classes

Databases

class quazy.DBFactory(connection_pool: DBPoolLike, translator: type[Translator], named_factory: type[DBRowFactoryLike], dict_factory: type[DBRowFactoryLike], class_factory: type[DBRowFactoryLike])

Basic database factory class

Use this class to arrange connection to a database, parse table models and run queries.

Note

It supports a Postgres database only at the moment

Example

db = DBFactory.postgres(conninfo=”postgresql://quazy:quazy@localhost/quazy”) db._debug_mode = True db.bind_module() db.clear() db.create() …

property translator

Translator contains all database specific primitives to construct SQL code

static postgres(conninfo: str, **kwargs) → DBFactory | None

Proxy constructor Postgres specific connection

Parameters:

• conninfo – connection string to connect to a database

• **kwargs – all keywords arguments passed to psycopg constructor

Returns:

DBFactory instance or None

Example:

db = DBFactory.postgres("postgresql://quazy:quazy@localhost/quazy")

static postgres_pool(conninfo: str, **kwargs) → DBFactory | None

Proxy constructor Postgres specific pool

Parameters:

• conninfo – connection string to connect to a database

• **kwargs – all keywords arguments passed to psycopg constructor, except debug_mode

Returns:

DBFactory instance or None

Example:

db = DBFactory.postgres_pool(conninfo="postgresql://quazy:quazy@localhost/quazy")

static sqlite(conn_uri: str, **kwargs) → DBFactory | None

Proxy constructor SQLite specific connection

Parameters:

• conn_uri – connection string to connect to a database

• kwargs – all keywords arguments passed to sqlite3 constructor, except debug_mode

Example:

db = DBFactory.sqlite("file:test.db?mode=rwc")

close() → Awaitable[None] | None

Close the database connection and release all resources

bind(cls: type[DBTable], schema: str = 'public')

Bind a specific table to the factory instance

Parameters:

• cls – Table to bind

• schema – schema name

bind_module(name: str = None, schema: str = 'public')

Bind a specific module by name to the database factory

Parameters:

• name – module name to bind. If not specified, bind current module

• schema – schema name to use

unbind(schema: str = 'public')

Unbind all tables

Parameters:

schema – schema name to unbind. “Public” by default. If “None”, unbind all tables from all schemas.

query(table_class: type[DBTableT] | None = None, name: str | None = None) → DBQuery[DBTableT]

Create DBQuery instance

Create a DBQuery instance bound to a specified DBTable or to the whole schema

Parameters:

• table_class – DBTable class to use or None

• name – name of the query for subquery request

Returns:

DBQuery instance

Example:

q = db.query(Street).select('name')
q = db.query().select(name=lambda s: s.street.name)

get(table_class: type[DBTableT], pk: Any = None, **fields) → Awaitable[DBTableT] | DBTableT

Request one row from a database table

Hint

This method is not intended to be used directly. It is shorter to call the get method from the DBTable instance.

Parameters:

• table_class – DBTable class to use

• pk – primary key value to filter row (optional)

• **fields – field values to filter row if no pk is specified (optional)

Returns:

DBTable instance

connection() → AsyncGenerator[DBConnectionLike] | Generator[DBConnectionLike]

Context manager for connection

clear(schema: str = None) → Awaitable[None] | None

Drop all known tables in a database

Parameters:

schema – schema name to use

Warning

It works without attention. Please double-check before calling.

all_tables(schema: str = None, for_stub: bool = False) → list[type[DBTable]]

Get all known tables in the database

Parameters:

• schema – schema name to use

• for_stub – whether to return tables only for stub purposes (uses internally)

Returns:

List of table classes

create(schema: str = None) → Awaitable[None] | None

Create all added tables in the database

insert(item: DBTableT, _curr: DBCursorLike = None) → Awaitable[DBTableT] | DBTableT

Insert item into the database

Parameters:

item – instance of DBTable to insert

Returns:

updated item, just for chain calls

Note

Inserted item could contain subtables of subclasses and many-to-many sets. In this case they will be bulk inserted.

update(item: DBTableT, _curr: DBCursorLike = None) → Awaitable[DBTableT] | DBTableT

Update item changes to a database

Parameters:

item – instance of DBTable to insert

Returns:

updated item, just for chain calls

Note

If an item instance has subtables of subclasses, they will be bulk updated (delete and insert).

select(query: DBQuery | str, as_dict: bool = False) → AsyncGenerator[Iterator[DBTable | SimpleNamespace | dict[str, Any]]] | Generator[Iterator[DBTable | SimpleNamespace | dict[str, Any]]]

Select items from the database

It performs a prepared query to a database and yields results. Is not intended for direct calls. Use preferably select method of DBQuery instance. Result type depends on query type: if a query is based on a specific DBTable, the result is an instance of DBTable if a query is based on a whole schema, the result is SimpleNamespace

Parameters:

• query – instance of DBQuery or string

• as_dict – results yield as dict instead of instance of DBTable or SimpleNamespace or named tuple

Warning

This method is intended only for data reading. To make modification request, use execute() instead.

Yields:

instance of DBTable or SimpleNamespace or dict

execute(query: str, as_dict: bool = False) → AsyncGenerator[Iterator[DBTable | SimpleNamespace | dict[str, Any]]] | Generator[Iterator[DBTable | SimpleNamespace | dict[str, Any]]]

Execute text query directly into database.

Similar to select() method, but allow changes.

Parameters:

• query – string

• as_dict – results yield as dict instead named tuple

update_many(query: DBQuery[DBTableT], **values) → Awaitable[None] | None

Update items in the database by a query

Parameters:

• query – instance of DBQuery bound to a table

• **values – dict of values to update

describe(query: DBQuery[DBTableT] | str) → list[DBField]

Describe query result fields information

It performs a prepared query to a database requesting zero rows just to collect information about fields.

Parameters:

query – instance of DBQuery or string

Returns:

list of DBField instances

save(_item: DBTableT, _lookup_field: str | None = None, _curr: DBCursorLike = None, **kwargs) → Awaitable[DBTableT] | DBTableT

Save item to the database

It checks whether an item needs to be inserted or updated. Specify lookup_field to avoid searching by a primary key.

Parameters:

• _item – instance of DBTable

• _lookup_field – field name or None

• kwargs – additional values to update item fields before saving it to the database

Returns:

updated instance, just for chain calls

delete(table: type[DBTable] = None, *, item: DBTableT = None, id: Any = None, items: Iterator[DBTableT] = None, query: DBQuery[DBTableT] = None, filter: Callable[[DBTableT], DBSQL] = None, _curr: DBCursorLike = None) → Awaitable[None] | None

Delete an item or items from the database

It has many possible ways to delete expected items:

• by item

• by table and id (for a primary key)

• by item list

• by query based on DBTable

• by table and lamba filter

Parameters:

• table – DBTable class

• item – instance of DBTable

• id – instance primary key id

• items – iterable of DBTable instances

• query – instance of DBQuery based on DBTable

• filter – callable lamba added to a query

Raises:

QuazyWrongOperation – wrong arguments usage

Fields

class quazy.DBField(column: str = '', pk: bool = False, cid: bool = False, body: bool = False, property: bool = False, indexed: bool = False, unique: bool = False, default: Union[Any, Callable[[DBTable], Any]] = <class 'quazy.db_field.Unassigned'>, default_sql: str = None, reverse_name: str = None, ux: Optional[UX] = None)

Table field description class

property type

field type class

name: str = ''

field name in Python

column: str = ''

field/column name in database

pk: bool = False

is it a primary key?

cid: bool = False

is it storage of table name for derived tables ?

ref: bool = False

is it a foreign key (reference)?

body: bool = False

is it a body field for properties?

property: bool = False

is it a properties field?

required: bool = True

is a field not null ?

indexed: bool = False

is it indexed for fast search ?

unique: bool = False

is it unique ?

default

alias of Unassigned

default_sql: str = None

default value at SQL level

reverse_name: str = None

reverse name for reference fields

ux: UX | None = None

UX/UI specific attributes

class quazy.UX(name: str = '', type: type = None, title: str = '', width: int = None, choices: Mapping[str, Any]=None, blank: bool = False, readonly: bool = False, multiline: bool = False, hidden: bool = False, sortable: bool = True, resizable: bool = True, meta: dict[str, Any]=<factory>)

Base class for visual representation of field, UI specific

This class contains the most common visual representation properties. It isn’t used in Quazy directly, but it helps to integrate with any GUI framework.

name: str = ''

type: type = None

title: str = ''

user level title of a field

width: int = None

integer size in GUI specific units (usually, letters amount)

choices: Mapping[str, Any] = None

select value by user level title from dropdown list

blank: bool = False

allow field unfilled

readonly: bool = False

disable modifications of a field

multiline: bool = False

enable multiline editor for a text field

hidden: bool = False

hide field from UI

sortable: bool = True

allows sorting by field values in tables

resizable: bool = True

allows resizing column of field in tables

meta: dict[str, Any]

additional data for UI specific

property field: DBField

reference to original field

Table class

class quazy.DBTable(**initial)

Table model constructor

All class attributes are considered as database table fields. Types could be set with annotations or/and directly as DBField instance. There are several special class attributes used to set DBTable details and behaviour.

Note

attribute named pk is reserved as property for direct primary key access

_table_: ClassVar[str]

database table internal name

_title_: ClassVar[str]

user-friendly table name

_schema_: ClassVar[str]

explicit schema name

_just_for_typing_: ClassVar[bool]

internal flag used for migrations

_extendable_: ClassVar[bool]

set extendable flag for a table

_discriminator_: ClassVar[Any]

SQL safe CID value to specify table in extended mode

_meta_: ClassVar[bool]

table marked as meta table

_lookup_field_: ClassVar[str]

specify field name for text search. For integrations only.

_use_slots_: ClassVar[bool]

use slots for database fields (enabled by default)

_validate_: ClassVar[bool]

True if pydantic is installed)

Type:

validate fields types using pydantic (default

_metadata_: ClassVar[dict[str, Any]]

any custom metadata

class DB

DBTable meta-subclass with internal information

It has only class-based attributes, intended for read-only. Instances aren’t supported.

db: ClassVar[DBFactory] = None

DBFactory linked to a table, if already specified

table: ClassVar[str | None] = None

database table name

source_table: ClassVar[str | None] = None

user defined database table name

title: ClassVar[str | None] = None

user-friendly table name

schema: ClassVar[str | None] = None

database schema name

source_schema: ClassVar[str | None] = None

user-defined schema name

just_for_typing: ClassVar[bool] = False

internal flag used for migrations

snake_name: ClassVar[str]

internal flag used for migrations

extendable: ClassVar[bool] = False

support for extendable classes

cid: ClassVar[DBField | None] = None

CID field reference (if declared)

is_root: ClassVar[bool] = False

is table a root of extendable tables chain

discriminator: ClassVar[Any | None] = None

derived table inner code

owner: ClassVar[Union[str, type[DBTable]] | None] = None

table owner of subtable

subtables: ClassVar[dict[str, type[DBTable]] | None] = None

subtables list

meta: ClassVar[bool] = False

table marked as meta table

pk: ClassVar[DBField | None] = None

reference to primary field DBField

body: ClassVar[DBField | None] = None

reference to body field or None

many_fields: ClassVar[dict[str, DBManyField] | None] = None

dict of field sets, when this table is referred from another table

many_to_many_fields: ClassVar[dict[str, DBManyToManyField] | None] = None

dict of field sets, when two tables referred to each other

fields: ClassVar[dict[str, DBField]]

all fields dict

self_fields: ClassVar[list[str] | None]

field names, declared in this class only

defaults: ClassVar[dict[str, Any]]

default values for fields

lookup_field: ClassVar[str | None] = None

field name for text search for integrations

use_slots: ClassVar[bool] = False

use slots for database fields

validate: ClassVar[bool] = False

validate fields types using pydantic

validators: ClassVar[dict[str, TypeAdapter[Any]]]

dict of validators for fields

metadata: ClassVar[dict[str, Any]]

any custom metadata

class ItemGetter(db: DBFactory | DBFactoryAsync, table: type[DBTableT], field_name: str, pk_value: Any, view: str = None)

class ItemGetterAsync(db: DBFactory | DBFactoryAsync, table: type[DBTableT], field_name: str, pk_value: Any, view: str = None)

class ListGetter(db: DBFactory, table: type[DBTable], field_name: str, pk_value: Any)

class ListGetterAsync(db: DBFactory, table: type[DBTable], field_name: str, pk_value: Any)

classmethod check_db()

Check whether DBTable is assigned to DBFactory

Raises:

QuazyWrongOperation – table is not assigned

classmethod get(pk: Any = None, **fields) → Awaitable[Self] | Self

Get DBTable instance by primary key value

Parameters:

• pk – primary key value to get an item (optional)

• **fields – fields values to find item if no pk is specified (optional)

save(**kwargs) → Awaitable[Self] | Self

Save DBTable instance changes to a database

Parameters:

kwargs – additional values to update item fields before saving it to the database

load(selected_field_name: str | None = None) → Awaitable[None] | None

Load related items from foreign tables

Parameters:

selected_field_name – any related field name to load, if not specified, all related fields will be loaded.

delete() → Awaitable[None] | None

Delete DBTable instance from a database

classmethod query(name: str | None = None) → DBQuery[Self]

Create a DBQuery instance for queries associated with this table

Parameters:

name – name of the query for subquery request

Hint

Use identical method select() for your preference.

classmethod select(*field_names: str, **fields: FDBSQL) → DBQuery[Self]

Create a DBQuery instance and specify selected fields

Read DBQuery.select() for details.

property pk

get a primary key value

inspect() → str

Inspect table in simple text format

key: value (type)

classmethod _view_(item: DBQueryField[Self])

virtual method to override DBTable item presentation

Originally, each table item is requester as a primary key value (integer number for ex.). It is more convenient to see user-friendly presentation, like name, caption or several fields combined.

Example:

class User(DBTable):
    name: str

    @classmethod
    def _view_(cls, item: DBQueryField):
        return item.name

classmethod get_lookup_field(item: DBQueryField) → DBSQL | None

return lookup field

_before_update(db: DBFactory)

abstract event before update to the database

_after_update(db: DBFactory)

abstract event after update to the database

_before_insert(db: DBFactory)

abstract event before insert to the database

_after_insert(db: DBFactory)

abstract event after insert to the database

_before_delete(db: DBFactory)

abstract event before delete from the database

_after_delete(db: DBFactory)

abstract event after delete from the database

quazy.dbtable(cls=None, /, *, table: str = None, title: str = None, schema: str = None, extendable: bool = None, discriminator: Any = None, meta: bool = False, lookup_field: str = None, use_slots: bool = False, validate: bool = VALIDATE_DATA, metadata: dict[str, Any] = None)

quazy.dbtable(cls=None, **kwargs)

dataclass()-styled decorator for DBTable subclasses

Example:

@dbtable
class User:
    name: str

Queries

class quazy.DBQuery(db: DBFactory, table_class: type[DBTableT] | None = None, name: str = '')

Query base class

Create it with DBFactory.query() or DBTable.query().

copy()

Make a copy of a query

reuse()

Put context generated query into the hash

get_scheme() → Generator[SimpleNamespace | DBQueryField[DBTableT]]

Scheme object for query context

Scheme contains

• snake names of all tables assigned to a database, if a query is not bound to a table

• all fields otherwise

Each attribute of a scheme works as an expression generator.

var(key: str, value: Any | None = None) → DBSQL

Define variable to pass to query.

Put variable to a query to avoid big query reconstruction.

Parameters:

• key – variable name

• value – variable value

Example:

q = db.query(Figures).select("name")
q.filter(lambda x: x.angles == q.var('angles'))
for angle in range(3, 7):
    q['angle'] = angle
    print(q.fetch_one())

sql(sql_text: str, *args: Any) → DBSQL

Add raw SQL to query.

Parameters:

• sql_text – SQL text to add

• args – arguments to substitute in SQL text, place mark is {}

with_query(subquery: DBQuery, not_materialized: bool = False) → DBSubqueryField

Use another query result field for this query.

Example:

q = db.query(Sales).select("date", "sum")
q2 = db.query()
sub = q2.with_query(q)
q2.select(total_sum=q2.sum(sub.sum))

Parameters:

• subquery – subquery to use

• not_materialized – ask the database engine to not request a whole query result set

Returns:

DBSubqueryField with result field names directly accessible for expressions

select(*field_names: str, **fields: FDBSQL) → DBQuery[DBTableT]

Specify a list of selected fields

Don’t call this method if you want to fetch a list of DBTable instances (with all fields). Otherwise, include “pk” in field_names or you will get a list of named tuples.

Parameters:

• *field_names – names of fields to select

• **fields – fields to select, where values can be lambdas

Returns:

DBQuery for chain calls

select_all() → DBQuery[DBTableT]

Select all possible fields for this query.

This is similar to a SELECT * FROM … query.

Note

This method prevents fetching DBTable instances to avoid collision with specific fields. Use select_objects instead.

Returns:

DBQuery for chain calls

select_objects() → DBQuery[DBTableT]

Select all fields specified for this query.

Returns:

DBQuery for chain calls

distinct() → DBQuery[DBTableT]

Select only different rows for this query.

Add a DISTINCT clause to a SELECT … statement.

sort_by(*fields: FDBSQL, desc: bool = False) → DBQuery[DBTableT]

Add sorting to a query

Parameters:

• *fields – fields to sort, can be field name, field number or lambda expression

• desc – sort ascending if False

Returns:

DBQuery for chain calls

filter(_expression: FDBSQL | DBTable = None, **kwargs) → DBQuery[DBTableT]

Add filter to a query

Filter can be applied by a common lambda expression or by specific field/value pairs.

Hint

Use identical method name where for your preference.

Example:

just_teens = Kids.select().filter(age=18)
older_then = Kids.select().filter(lambda x: x.age > 18)

Parameters:

• _expression – lambda expression to filter

• **kwargs – field/value pairs to filter

Returns:

DBQuery for chain calls

where(_expression: FDBSQL | DBTable = None, **kwargs) → DBQuery[DBTableT]

Add filter to a query

Filter can be applied by a common lambda expression or by specific field/value pairs.

Hint

Use identical method name where for your preference.

Example:

just_teens = Kids.select().filter(age=18)
older_then = Kids.select().filter(lambda x: x.age > 18)

Parameters:

• _expression – lambda expression to filter

• **kwargs – field/value pairs to filter

Returns:

DBQuery for chain calls

exclude(_expression: FDBSQL = None, **kwargs) → DBQuery[DBTableT]

Filter elements to exclude from a query

Works like a negative filter (excluding elements from a selection)

Example:

no_teens = Kids.select().exclude(age=18)
youngsters = Kids.select().exclude(lambda x: x.age > 18)

Parameters:

• _expression – lambda expression to filter

• **kwargs – field/value pairs to filter

Returns:

DBQuery for chain calls

group_filter(expression: FDBSQL) → DBQuery[DBTableT]

Filter applied to group fields. See below

Hint

This method is not necessary to call, because expression resolver detects aggregated functions calls automatically.

Parameters:

expression – lambda expression to filter

Returns:

DBQuery for chain calls

group_by(*fields: FDBSQL) → DBQuery[DBTableT]

Specify group fields for aggregated results

This is a query analogue to GROUP BY … statement.

Parameters:

*fields – list of field names or expressions to group

Returns:

DBQuery for chain calls

set_window(offset: int | None = None, limit: int | None = None) → DBQuery[DBTableT]

Set the query result window using SQL offset/limit features

This is analogue to SELECT a, b, c FROM table OFFSET … LIMIT … statement.

sum(expr: FDBSQL) → DBSQL

Use aggregated function sum as a part of the expression

Example:

q = db.query(Posts)
q = q.group_by("topic").select("topic", total_views=q.sum("views_counter"))

count(expr: FDBSQL = None, distinct: bool = False) → DBSQL

Use aggregated function count as a part of the expression

If no argument is specified, count all result rows.

Parameters:

• expr – expression as a path to the table field

• distinct – count distinct rows

Example:

q = db.query(Posts)
q = q.group_by("topic").select("topic", total_views=q.count)

avg(expr: FDBSQL) → DBSQL

Use aggregated function avg (average) as a part of the expression

min(expr: DBSQL) → DBSQL

Use aggregated function min as a part of the expression

max(expr: DBSQL) → DBSQL

Use aggregated function max as a part of the expression

case() → DBConditionField

Make DBConditionField object for conditional values

This is an analogue to SQL CASE … statement.

Example:

q = db.query(User)
c = q.case().
    condition("baby", lambda x: x.age <= 1).
    condition("toddler", lambda x: x.age <= 3).
    condition("kid", lambda x: x.age < 18).
    default("adult")
q.select("name", age_category=c)

execute(as_dict: bool = False) → AsyncGenerator[DBTableT] | Generator[DBTableT]

Execute query and yields database cursor to fetch one or more result rows.

Parameters:

as_dict – whether to return dict instead of DBTable/SimpleNamespace

Yields:

database cursor

describe() → Awaitable[list[DBField]] | list[DBField]

Request all result fields information.

See DBFactory.describe()

fetch_one(as_dict: bool = False) → Awaitable[DBTableT | Any] | DBTableT | Any

Execute query and fetch first result row

get(pk_id: Any) → DBTableT | None

Request and get one row by the primary key identifier

classmethod any(expr_list: Iterator[DBSQL]) → DBSQL

Produce expression with several alternatives.

This is analogue to ex1 OR ex2 OR ex3 …

Parameters:

expr_list – list, tuple or other iterator of expressions

Example:

colors = ('green', 'yellow', 'red')
q = db.query(Apple)
q.filter(q.any(lambda x: x.color == color for color in colors))

fetch_all(as_dict: bool = False) → Awaitable[list[DBTableT | Any]] | list[DBTableT | Any]

Execute a query and fetch all result rows as a list

fetch_value() → Awaitable[Any] | Any

Execute a query and fetch first column value of first result row

fetch_list(index: int | str = 0) → Awaitable[list[DBTableT | Any]] | list[DBTableT | Any]

Execute a query and fetch the first column of all result rows as a list of values

Parameters:

index – column index or name

exists() → bool

Execute a query and check whether the first result row exists

fetch_aggregate(function: str, expr: FDBSQL = None) → Awaitable[Any] | Any

Execute subquery to fetch aggregate function result value

This group of functions is intended to estimate query metrics and numbers before real execution.

Example:

q = db.query(Posts).filter(lambda x: x.created_at >= datetime.now() - timedelta(days=10))
print(q.fetch_count())

Parameters:

• function – SQL-friendly aggregate function name

• expr – any expression, like lambdas or DBSQL

Returns:

integer or float requested value

fetch_count(expr: FDBSQL = None) → Awaitable[int | None] | int | None

Execute subquery to fetch aggregate function count result value

fetch_sum(expr: FDBSQL) → Awaitable[Any] | Any

Execute subquery to fetch aggregate function sum result value

fetch_max(expr: FDBSQL) → Awaitable[Any] | Any

Execute subquery to fetch aggregate function max result value

fetch_min(expr: FDBSQL) → Awaitable[Any] | Any

Execute subquery to fetch aggregate function min result value

fetch_avg(expr: FDBSQL) → Awaitable[Any] | Any

Execute subquery to fetch aggregate function avg result value

update(**values) → Awaitable[DBQuery[DBTableT]] | DBQuery[DBTableT]

Updates the current query object with the specified values

chained(id_name: str, next_name: str, start_value: Any) → DBQuery[DBTableT]

Select chained rows via recursive request

Parameters:

• id_name – name of the field with an original identifier

• next_name – name of the field with an identifier of the next row

• start_value – starting identifier value for the first row in the chain

Example:

class Chained(DBTable):
    index: int
    next: int
    name: str

q = Chained.chained("index", "next", 1)
print(q.fetch_all())

freeze() → DBQuery[DBTableT]

Build SQL and freeze the query object to prevent further changes

property is_frozen: bool

Whether the query object is frozen