kerno.repository.sqlalchemy module

A base class for SQLAlchemy-based repositories.

class kerno.repository.sqlalchemy.BaseSQLAlchemyRepository(kerno: kerno.kerno.Kerno, session_factory: Optional[Any] = None)

Bases: object

Base class for a SQLAlchemy-based repository.

SAS = 'session factory'

add(entity: kerno.typing.Entity) → kerno.typing.Entity

Add an object to the SQLAlchemy session, then return it.

add_all(entities: Sequence[kerno.typing.Entity]) → None

Add model instances to the SQLAlchemy session.

create_or_update(cls: type, values: Dict[str, Any] = {}, **filters) → Tuple[Any, bool]

Load and update entity if it exists, else create one.

First obtain either an existing object or a new one, based on filters. Then apply values and return a tuple (object, is_new).

delete(entity: kerno.typing.Entity) → None

Delete an entity from the database.

flush() → None

Obtain IDs on new objects and update state on the database.

Without committing the transaction.

get_or_create(cls: type, **filters) → Tuple[Any, bool]

Retrieve or add object; return a tuple (object, is_new).

is_new is False if the object already existed in the database.

new_sas(session_factory)

Obtain a new SQLAlchemy session instance.

update_association(cls: type, field: str, ids: Sequence[int], filters: Dict[str, Any], synchronize_session=None) → List[kerno.typing.Entity]

Update a many-to-many relationship. Return only NEW associations.

class kerno.repository.sqlalchemy.Query

Bases: Iterable, Generic[kerno.typing.Entity]

Typing stub for a returned SQLAlchemy query.

This is purposely very incomplete. It is intended to be used as return value in repository methods, such that user code can use, but not change, the returned query, which is what we like to do in this architecture.

If you want a more complete implementation, try https://github.com/dropbox/sqlalchemy-stubs

Their query stub is at https://github.com/dropbox/sqlalchemy-stubs/blob/master/sqlalchemy-stubs/orm/query.pyi

all() → List[kerno.typing.Entity]

count() → int

delete() → None

first() → Optional[kerno.typing.Entity]

get(ident) → Optional[kerno.typing.Entity]

one() → kerno.typing.Entity

yield_per(count: int) → List[kerno.typing.Entity]

class kerno.repository.sqlalchemy.SpyRepo(**kw)

Bases: object

Nice test double, can be inspected at the end of a test.

add(entity: kerno.typing.Entity) → kerno.typing.Entity

add_all(entities: Sequence[kerno.typing.Entity]) → None

delete(entity) → None

flush() → None

kerno.repository.sqlalchemy.update_association(cls: type, field: str, ids: Sequence[int], filters: Dict[str, Any], sas, synchronize_session=None) → List[kerno.typing.Entity]

Update a many-to-many relationship. Return only NEW associations.

When you have a many-to-many relationship, there is an association table between 2 main tables. The problem of setting the data in this case is a recurring one and it is solved here. Existing associations might be deleted and some might be created.

Example usage:

user = session.query(User).get(1)
# Suppose there's a many-to-many relationship to Address,
# through an entity in the middle named UserAddress which contains
# just the columns user_id and address_id.
new_associations = update_association(
    cls=UserAddress,      # the association class
    field='address_id'     # name of the remote foreign key
    ids=[5, 42, 89],        # the IDs of the user's addresses
    filters={"user": user},  # to load existing associations
    sas=my_sqlalchemy_session,
)
for item in new_associations:
    print(item)

This method returns a list of any new association instances because you might want to finish the job by doing something more with them (e. g. setting other attributes).

A new query is needed to retrieve the totality of the associations.