chapps.dbmodels module

CHAPPS data schemata

Database schema expressed as SQLAlchemy ORM Models

Basic data models correspond to Pydantic models for API, but more models are required for the database side:

  • There is a subclass of the metaclass, DeclarativeMeta, which provides extra logic to the database models to automate common queries and tasks.

  • There are table definitions for the join tables.

  • There is a special class for generalizing the logic of joining the tables.

chapps.dbmodels.convention = {'ck': 'ck_%(table_name)s_%(constraint_name)s', 'fk': 'fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s', 'ix': 'ix_%(column_0_label)s', 'pk': 'pk_%(table_name)s', 'uq': 'uq_%(table_name)s_%(column_0_name)s'}

SQLA’s recommended naming convention for constraints

class chapps.dbmodels.DB_Customizations[source]

Bases: sqlalchemy.orm.decl_api.DeclarativeMeta

Custom ORM metaclass

In order to make common tasks more concise, and to conceal the need for complex data massaging in order to produce appropriate data structures for the different kinds of queries, a fair amount of custom code is introduced into the ORM models via this metaclass. These routines allow calls to the ORM class itself, to get it to return statement objects (objects defined in sqlalchemy.sql.expression).

select_by_id(id)[source]

SELECT (load) a single object by ID

Parameters

id (int) –

windowed_list_by_ids(*, ids=[], subquery=None, skip=0, limit=1000, q='%')[source]

SELECT a window of objects

Parameters

  • ids (Optional[List[int]]) – a list of IDs

  • subquery (Optional[Subquery]) – a sub-SELECT which yields IDs

  • skip (Optional[int]) – start the window after skip entries

  • limit (Optional[int]) – include up to limit entries in the result

  • q (Optional[str]) – a substring to match against the name field

Raises

ValueError – unless one of ids or subquery is provided

Returns

a Select implementing the window

Return type

sqlalchemy.sql.expression.Select

If both of subquery and ids are not empty, subquery prevails and ids is ignored.

select_by_ids(ids, assoc=None)[source]

Return a select statement for a list of objects, optionally with eager-loaded associations

Parameters
select_names_by_id(ids)[source]

Return a Select for the names corresponding to the provided IDs

Parameters

ids (List[int]) –

select_by_pattern(q)[source]

Return a Select for all records with names which include q as a substring

Parameters

q (str) –

select_by_name(q)[source]

Return a Select for the record whose name exactly matches q

Parameters

q (str) –

windowed_list(q='%', skip=0, limit=1000)[source]

Return a Select for a window of select_by_pattern()

Parameters
remove_by_id(ids)[source]

Return a Delete for the listed IDs (ids may be a scalar ID also)

Parameters

ids (Union[int, List[int]]) –

update_by_id(item)[source]

Return an Update statement for the specified item

Parameters

  • cls (type) – an ORM model class (subclass of DB_Base)

  • item (cls) – an instance of an ORM model of the type stored in cls

Returns

an Update statement object representing the new values of item

Return type

sqlalchemy.sql.expression.Update

chapps.dbmodels.DB_Base

DB_Base serves as the base of all ORM models

This class itself contains literally no code apart from documentation. All of the magic provided for the ORM layer is implemented in the metaclass, DB_Customizations. To be clear, all of the methods defined in the metaclass become available as class methods on the classes derived from DB_Base, because it is of type DB_Customizations.

chapps.dbmodels.quota_user = Table('quota_user', MetaData(), Column('user_id', Integer(), ForeignKey('users.id'), table=<quota_user>, primary_key=True, nullable=False), Column('quota_id', Integer(), ForeignKey('quotas.id'), table=<quota_user>, nullable=False), schema=None)

the quota_user join table

chapps.dbmodels.domain_user = Table('domain_user', MetaData(), Column('user_id', Integer(), ForeignKey('users.id'), table=<domain_user>, primary_key=True, nullable=False), Column('domain_id', Integer(), ForeignKey('domains.id'), table=<domain_user>, primary_key=True, nullable=False), schema=None)

the domain_user join table

chapps.dbmodels.email_user = Table('email_user', MetaData(), Column('user_id', Integer(), ForeignKey('users.id'), table=<email_user>, primary_key=True, nullable=False), Column('email_id', Integer(), ForeignKey('emails.id'), table=<email_user>, primary_key=True, nullable=False), schema=None)

the email_user join table

class chapps.dbmodels.Quota[source]

Bases: sqlalchemy.orm.decl_api.Base

ORM Model for Quota definitions

Each record/instance has these columns/attributes:

id

integer primary key

name

unique string of 32 chars or less

quota

unique integer transmission attempt limit

__init__(**kwargs)

A simple constructor that allows initialization from kwargs.

Sets attributes on the constructed instance using the names and values in kwargs.

Only keys that are present as attributes of the instance’s class are allowed. These could be, for example, any mapped columns or relationships.

class chapps.dbmodels.Domain[source]

Bases: sqlalchemy.orm.decl_api.Base

ORM Model for Domain definitions

Each record/instance has these columns/attributes:

id

integer primary key

name

unique string of 64 chars or less

greylist

if True perform greylisting

check_spf

if True perform SPF enforcement

__init__(**kwargs)

A simple constructor that allows initialization from kwargs.

Sets attributes on the constructed instance using the names and values in kwargs.

Only keys that are present as attributes of the instance’s class are allowed. These could be, for example, any mapped columns or relationships.

class chapps.dbmodels.Email[source]

Bases: sqlalchemy.orm.decl_api.Base

ORM Model for Email definitions

Each record/instance has these columns/attributes:

id

integer primary key

name

unique string of 128 chars or less

__init__(**kwargs)

A simple constructor that allows initialization from kwargs.

Sets attributes on the constructed instance using the names and values in kwargs.

Only keys that are present as attributes of the instance’s class are allowed. These could be, for example, any mapped columns or relationships.

class chapps.dbmodels.User[source]

Bases: sqlalchemy.orm.decl_api.Base

ORM model for User objects

Each record/instance has these columns/attributes:

id

integer auto-incremented primary key

name

unique string of up to 128 chars

quota

associated Quota object; there can be only one Quota associated to a User

domains

list of associated Domain objects

a User may be associated to more than one Domain, and a Domain may be associated to more than one User

emails

list of associated Email objects

a User may be associated to more than one Email, and an Email may be associated to more than one User

__init__(**kwargs)

A simple constructor that allows initialization from kwargs.

Sets attributes on the constructed instance using the names and values in kwargs.

Only keys that are present as attributes of the instance’s class are allowed. These could be, for example, any mapped columns or relationships.

class chapps.dbmodels.JoinAssoc[source]

Bases: object

Represent joining two data tables via a two-column join table

There is probably some way to use SQLA for this but I am using it wrongly, to avoid loading associations just in order to link them to other objects via a join table

__init__(source_model, source_id, *, assoc_name, assoc_type, assoc_model, assoc_id, table)[source]

Define a join association

Parameters

  • source_model (DB_Base) – a reference to the dbmodel class of the source object

  • source_id (str) – label of the source object’s ID column in the join table

  • assoc_name (str) – attribute name of the association

  • assoc_type (type) – usually int or List[int]; should be the type for the API to expect when setting up the route metadata

  • assoc_model (DB_Base) – a reference to the dbmodel class of the associated object

  • assoc_id (str) – label of associated object’s ID column in the join table

  • table (sqlalchemy.schema.Table) – a reference to the join table schema; it will be a constant in this module, generally

select()[source]

Convenience method

Returns

a SELECT on the join table

Return type

sqlalchemy.sql.expression.Select

insert()[source]

Convenience method

Returns

an INSERT on the join table

Return type

sqlalchemy.sql.expression.Insert

delete()[source]

Convenience method

Returns

a DELETE FROM on the join table

Return type

sqlalchemy.sql.expression.Delete

update()[source]

Convenience method

Returns

an UPDATE on the join table

Return type

sqlalchemy.sql.expression.Update

values(item, assoc)[source]

Get value-insertion or update mapping

Parameters

  • item (Union[DB_Base, int]) – the source item or its ID

  • assoc (Union[int,List[int]]) – the association or associations to specify

Return type

List[Dict[str, int]]

Using the item passed in, which may be an ORM object descended from DB_Base or an int, construct a list of dictionaries mapping the join table’s ID column names correctly onto the source item ID and the associated item ID(s).

Return type

List[Dict[str, int]]

Parameters

assoc (Union[int, List[int]]) –

where_tuples(item_id, assoc)[source]

Get tuples suitable for use in an SQLAlchemy WHERE clause

Parameters
property source_col
property assoc_col
insert_assoc(item_id, vals)[source]
Parameters

item_id (int) –

delete_assoc(item_id, vals)[source]
Parameters

item_id (int) –

update_assoc(item_id, assoc_id)[source]
Parameters

  • item_id (int) –

  • assoc_id (int) –

select_ids_by_source_id(item_id)[source]
Parameters

item_id (int) –