atomdb.sql module

Copyright (c) 2018-2022, CodeLV.

Distributed under the terms of the MIT License.

The full license is in the file LICENSE.txt, distributed with this software.

Created on Aug 2, 2018

class atomdb.sql.ConnectionProxy[source]

Bases: atom.atom.Atom

An wapper for a connection to be used with async with syntax that does nothing but passes the existing connection when entered.

connection

A member class which supports value initialization.

A plain Value provides support for default values and factories, but does not perform any type checking or validation. It serves as a useful base class for scalar members and can be used for cases where type checking is not needed (like private attributes).

class atomdb.sql.RelatedInstance(*args, **kwargs)[source]

Bases: atom.instance.ForwardInstance

args
kwargs
optional
resolve
class atomdb.sql.RelatedTyped(*args, **kwargs)[source]

Bases: atom.typed.ForwardTyped

args
kwargs
optional
resolve
class atomdb.sql.Relation(item: Callable[[], Type[atomdb.base.Model]], default=None)[source]

Bases: atom.containerlist.ContainerList

A member which serves as a fk relation backref

resolve() → Type[atomdb.base.Model][source]
property to: Type[atomdb.base.Model]
class atomdb.sql.SQLBinding[source]

Bases: atom.atom.Atom

compiler(statement, parameters, **kwargs)[source]
connect(**kwargs)[source]
contextual_connect(**kwargs)[source]
create(entity, **kwargs)[source]
property dialect

Get the dialect of the database.

drop(entity, **kwargs)[source]
property engine
execute(object_, *multiparams, **params)[source]
execution_options(**kw)[source]
manager

Model Manager

property name
queue

The queue

schema_for_object(obj)[source]
async wait()[source]
class atomdb.sql.SQLMeta(name, bases, dct)[source]

Bases: atomdb.base.ModelMeta

Both the pk and _id are aliases to the primary key

class atomdb.sql.SQLModel[source]

Bases: atomdb.base.Model

A model that can be saved and restored to and from a database supported by sqlalchemy.

async delete(connection=None)[source]

Alias to delete this object in the database

async load(connection=None, reload: bool = False, fields: Optional[Sequence[str]] = None)[source]

Alias to load this object from the database

connection: Connection

The connection instance to use in a transaction

reload: Bool

If True force reloading the state even if the state has already been loaded.

fields: Sequence[str]

Optional list of field names to load. Use this to refresh specific fields from the database.

objects: ClassVar[ModelManager] = <atomdb.sql.SQLTableProxy object>

Use SQL object manager

async classmethod restore(state: Dict[str, Any], force: Optional[bool] = None, prefetched: Optional[Dict[Any, Dict[str, Any]]] = None, **kwargs: Any) → atomdb.sql.T[source]

Restore an object from the database using the primary key. Save a ref in the table’s object cache. If force is True, update the cache if it exists.

state: Mapping[str, Any]

A mapping of field name to value. May contain result of a join (eg state of multiple models prefexed with the table name).

force: Optional[bool]

Whether to force calling restorestate. This is used to to avoid restoring cached objects.

prefetched: Optional[dict]

A mapping of prefetched related values. If present the objects primary key is looked up and added to the state.

model: SQLModel

The restored or cached model.

async save(force_insert: bool = False, force_update: bool = False, update_fields: Optional[Sequence[str]] = None, connection=None)[source]

Alias to save this object to the database

force_insert: Bool

Ensure that save performs an insert

force_update: Bool

Ensure that save performs an update

update_fields: Iterable[str]

If given, only update the given fields

connection: Connection

The connection instance to use in a transaction

result: Value

Update or save result

serializer: ClassVar[ModelSerializer] = <atomdb.sql.SQLModelSerializer object>

Use SQL serializer

class atomdb.sql.SQLModelManager[source]

Bases: atomdb.base.ModelManager

Manages models via aiopg, aiomysql, or similar libraries supporting SQLAlchemy tables. It stores a table for each class and when accessed on a Model subclass it returns a table proxy binding.

cache

Cache results.

conventions

Constraint naming convenctions

create_table_and_restore_fn(cls: Type[atomdb.sql.SQLModel]) → sqlalchemy.sql.schema.Table[source]

Generate the sqlalchemy table and optimized restore function. This is done here to make sure that foreign and forwarded members are now resolved.

create_tables() → Dict[Type[atomdb.sql.SQLModel], sqlalchemy.sql.schema.Table][source]

Create sqlalchemy tables for all registered SQLModels

database

Used to access the database

metadata

Metadata

proxies

Table proxy cache

class atomdb.sql.SQLModelSerializer[source]

Bases: atomdb.base.ModelSerializer

Uses sqlalchemy to lookup the model.

flatten_object(obj: atomdb.base.Model, scope: Dict[int, Any]) → Any[source]

Serialize a model for entering into the database

obj: Model

The object to unflatten

scope: Dict

The scope of references available for circular lookups

result: Object

The flattened object

async get_object_state(obj, state, scope)[source]

Load the object state if needed. Since the __model__ is not saved to the db tables with SQL we know that if it’s “probably” there because a query was used.

registry

Store all registered models

class atomdb.sql.SQLQuerySet[source]

Bases: atom.atom.Atom, Generic[atomdb.sql.T]

aggregate(*args, func=None)[source]
async all(*args, **kwargs) → Sequence[atomdb.sql.T][source]

Get the all results matching the query. This will force restore the items even if it was in the cache.

results: list[Model]

The models entry matching the query

clone(**kwargs) → atomdb.sql.SQLQuerySet[atomdb.sql.T][source]
connection

A member class which supports value initialization.

A plain Value provides support for default values and factories, but does not perform any type checking or validation. It serves as a useful base class for scalar members and can be used for cases where type checking is not needed (like private attributes).

async count(*args, **kwargs) → int[source]
async delete(*args, **kwargs)[source]
distinct(*args)[source]

Apply distinct on the given column.

args: List[str or column]

Fields that must be distinct.

query: SQLQuerySet

A clone of this queryset with the distinct terms added.

distinct_clauses

A member which allows list values.

Assigning to a list creates a copy. The orginal list will remain unmodified. This is similar to the semantics of the assignment operator on the C++ STL container classes.

async exists(*args, **kwargs) → bool[source]
filter(*args, **kwargs: Dict[str, Any])[source]

Filter the query by the given parameters. This accepts sqlalchemy filters by arguments and django-style parameters as kwargs.

args: List

List of sqlalchemy filters

kwargs: Dict[str, object]

Django style filters to use

query: SQLQuerySet

A clone of this queryset with the filter terms added.

filter_clauses

A member which allows list values.

Assigning to a list creates a copy. The orginal list will remain unmodified. This is similar to the semantics of the assignment operator on the C++ STL container classes.

force_restore

A value of type bool.

async get(*args, **kwargs) → Optional[atomdb.sql.T][source]

Get the first result matching the query. Unlike django this will NOT raise an error if multiple objects would be returned or an entry does not exist. This will force restore the item even if it was in the cache.

model: Optional[Model]

The first entry matching the query

limit(limit: int)[source]
limit_count

A value of type int.

By default, ints are strictly typed. Pass strict=False to the constructor to enable int casting for longs and floats.

max(*columns)[source]
min(*columns)[source]
mode(*columns)[source]
offset(offset: int)[source]
order_by(*args)[source]

Order the query by the given fields.

args: List[str or column]

Fields to order by. A “-” prefix denotes decending.

query: SQLQuerySet

A clone of this queryset with the ordering terms added.

order_clauses

A member which allows list values.

Assigning to a list creates a copy. The orginal list will remain unmodified. This is similar to the semantics of the assignment operator on the C++ STL container classes.

outer_join

A value of type bool.

async prefetch() → Optional[Dict[Any, Dict[str, Any]]][source]

Perform a prefetch lookup and populate the cache.

prefetch_clauses

A member which allows list values.

Assigning to a list creates a copy. The orginal list will remain unmodified. This is similar to the semantics of the assignment operator on the C++ STL container classes.

Define related fields to prefetch in a separate query.

args: List[str]

List of related fields fetchs

query: SQLQuerySet

A clone of this queryset with the prefetch fields added.

proxy

Proxy

query(query_type: str = 'select', *columns, **kwargs)[source]
query_offset

A value of type int.

By default, ints are strictly typed. Pass strict=False to the constructor to enable int casting for longs and floats.

related_clauses

A member which allows list values.

Assigning to a list creates a copy. The orginal list will remain unmodified. This is similar to the semantics of the assignment operator on the C++ STL container classes.

Define related fields to join in the query.

args: List[str]

List of related fields to join.

outer_join: Bool

If given set whether or not a left outer join is used.

query: SQLQuerySet

A clone of this queryset with the related field terms added.

sum(*columns)[source]
async update(**values)[source]

Perform an update of the given values.

async values(*args, distinct: bool = False, flat: bool = False, group_by: Optional[Sequence[Union[str, sqlalchemy.sql.schema.Column]]] = None) → Sequence[Any][source]

Returns the results as a list of dict instead of models.

args: List[str or column]

List of columns to select

distinct: Bool

Return only distinct rows

flat: Bool

Requires exactly one arg and will flatten the result into a single list of values.

group_by: List[str or column]

Optional Columns to group by

results: List

List of results depending on the parameters described above

class atomdb.sql.SQLTableProxy[source]

Bases: atom.atom.Atom, Generic[atomdb.sql.T]

cache

Cache of pk: obj using weakrefs

connection(connection=None)[source]

Create a new connection or the return given connection as an async contextual object.

connection: Database connection or None

The connection to return

connection: Database connection

The database connection or one that may be used with async with

connection_kwarg

Key used to pull the connection out of filter kwargs

async create(**state) → atomdb.sql.T[source]

Create a and save model with the given state.

The connection parameter is popped from this state.

state: Dict

The state to use to initialize the object.

result: Tuple[Model, Bool]

A tuple of the object and a bool indicating if it was just created

create_table()[source]

A wrapper for create which catches the create queries then executes them

drop_table()[source]
property engine

Retrieve the database engine.

async execute(*args, **kwargs)[source]
async fetchall(query: Union[str, sqlalchemy.sql.base.Executable], connection=None)[source]

Fetch all results for the query.

query: String or Query

The query to execute

connection: Database connection

The connection to use or a new one will be created

rows; List

List of rows returned, NOT objects

async fetchmany(query, size=None, connection=None)[source]

Fetch size results for the query.

query: String or Query

The query to execute

size: Int or None

The number of results to fetch

connection: Database connection

The connection to use or a new one will be created

rows: List

List of rows returned, NOT objects

async fetchone(query: Union[str, sqlalchemy.sql.base.Executable], connection=None)[source]

Fetch a single result for the query.

query: String or Query

The query to execute

connection: Database connection

The connection to use or a new one will be created

rows: Object or None

The row returned or None

async get_or_create(**filters) → Tuple[atomdb.sql.T, bool][source]

Get or create a model matching the given criteria

filters: Dict

The filters to use to retrieve the object

result: Tuple[Model, Bool]

A tuple of the object and a bool indicating if it was just created

model

Model which owns the table

restore_kwarg

Key used to pass the force restore option

async scalar(query: Union[str, sqlalchemy.sql.base.Executable], connection=None)[source]

Fetch the scalar result for the query.

query: String or Query

The query to execute

connection: Database connection

The connection to use or a new one will be created

result: Object or None

The the first column of the first row or None

table

Table this is a proxy to

atomdb.sql.atom_member_to_sql_column(model: Type[atomdb.sql.SQLModel], member: atom.catom.Member, **kwargs) → sqlalchemy.sql.type_api.TypeEngine[source]

Convert the atom member type to an sqlalchemy table column type See https://docs.sqlalchemy.org/en/latest/core/type_basics.html

atomdb.sql.create_table(model: Type[atomdb.sql.SQLModel], metadata: sqlalchemy.sql.schema.MetaData) → sqlalchemy.sql.schema.Table[source]

Create an sqlalchemy table by inspecting the Model and generating a column for each member.

model: SQLModel

The atom model

  1. https://docs.sqlalchemy.org/en/latest/core/metadata.html

atomdb.sql.create_table_column(model: Type[atomdb.sql.SQLModel], member: atom.catom.Member) → sqlalchemy.sql.schema.Column[source]

Converts an Atom member into a sqlalchemy data type.

model: Model

The model which owns this member

member: Member

The atom member

column: Column

An sqlalchemy column

  1. https://docs.sqlalchemy.org/en/latest/core/types.html

atomdb.sql.find_sql_models() → Iterator[Type[atomdb.sql.SQLModel]][source]

Finds all non-abstract imported SQLModels by looking up subclasses of the SQLModel.

cls: SQLModel

atomdb.sql.generate_sql_restorestate(cls: Type[atomdb.sql.SQLModel]) → Callable[[atomdb.base.M, Dict[str, Any], Optional[Dict[int, Any]]], None][source]

Generate an optimized restore function for the SQL model. The generated function creates “inline” dict key lookups for the table columns that may have been joined or renamed. This avoids having to do this at runtime.

async atomdb.sql.get_cached_model(cls: Type[atomdb.sql.T], pk: Any, state: Dict[str, Any]) → Optional[atomdb.sql.T][source]

Retrieve a model from the cache using the given pk. If the cached object does not exist attempt to restore it from the state otherwise create a model that has not been loaded and only contains the id.

cls: Type[SQLModel]

The class to lookup.

pk: Any

The primary key to look for.

state: StateType

The state from a join query.

obj: Optional[SQLModel]

If the pk is not None an instance of cls.

atomdb.sql.py_type_to_sql_column(model: Type[atomdb.base.Model], member: atom.catom.Member, types: Union[Type, Tuple[Type, ...]], **kwargs) → sqlalchemy.sql.type_api.TypeEngine[source]

Convert the python type to an alchemy table column type

atomdb.sql.resolve_member_column(model: Type[atomdb.sql.SQLModel], field: str, related_clauses: Optional[List[str]] = None) → sqlalchemy.sql.schema.Column[source]

Get the sqlalchemy column for the given model and field.

model: atomdb.sql.Model

The model to lookup

field: String

The field name

result: sa.Column

A tuple containing the through table (or None) and the sqlalchemy column.

atomdb.sql.resolve_relation(model: Type[atomdb.sql.SQLModel], field: str) → Tuple[atom.catom.Member, Type[atomdb.base.Model], atom.catom.Member, sqlalchemy.sql.schema.Column][source]

Lookup a Relation.

model: SQLModel

The model to lookup.

field: str

Path to a Relation, Typed, or Instance marked with store=False

result: tuple[Member, SQLModel, Member, sa.Column]

A tuple of the related field on the given model, the other model it points to, and the field on that model that points back to this model, and the Column.