Utilities

class keg_elements.db.utils.CollectionUpdater(entity, attr_name, data)

Update a collection attribute of a model object.

Takes a model object, the attribute name of the collection to be updated, and the data to update the collection with. If a record in the given data matches a record in the collection, the record is edited. Otherwise, the given record is added to the collection.

All records in the collection that are not included in the given data are removed from the collection.

find_child(data)

Find the child record associated with the related object

update()

Update the objects associated with the entity

Objects are matched by keys found in supplied data, if any. If no keys are supplied, the existing objects in the collection are removed, and new records are constructed from the data.

The way SQLAlchemy handles collection updates presents a limitation for collections having unique constraints. If we drop/recreate the record with the same unique value, we will get a unique constraint exception. To work around this, we cache info from matched objects in state, then remove/flush objects first. This clears the table for the new records from adds/edits.

keg_elements.db.utils.has_column(orm_cls_or_table, column_name)

Searches an entity for a column having the given key.

While a column’s default key is the entity attribute name, a column may be given an explicit key instead. This method uses sqlalchemy-utils to find the column.

keg_elements.db.utils.raises_check_exc(contraint_name)

pytest helper and context manager to ensure an exception result is a check constraint error.

keg_elements.db.utils.raises_unique_exc(constraint_name)

pytest helper and context manager to ensure an exception result is a duplicate key error.

keg_elements.db.utils.randemail(length, randomizer=<function randchars>)

Generate a random email address at the given length. :param length: must be at least 7 or the function will throw a ValueError. :param randomizer: is a function for generating random characters. It must have an identical interface to randchars. The default function is randchars.

keg_elements.db.utils.random_int(column, default_range)

Find a random number that satisfies the given column’s data size and meta info.

Parameters:

• column – SA column to find a range by type/info and generate a random number.

• default_range – Tuple. Fallback range if type/info does not provide a range.

Returns:

Random integer.

keg_elements.db.utils.random_numeric(column)

Find a random number that satisfies the given column’s precision and scale.

keg_elements.db.utils.session_commit()

Commit the db session, and roll back if there is a failure.

Raises the exception that caused the rollback.

keg_elements.db.utils.session_flush()

Flush the db session, and roll back if there is a failure.

Raises the exception that caused the rollback.

class keg_elements.db.utils.utcnow(*clauses: Any)

keg_elements.db.utils.validate_unique_exc(exc, constraint_name=None)

Check the given exception to see if it indicates a duplicate key error.

Supports SQLite, PostgreSQL, and MSSQL dialects.

Optionally provide a constraint_name kwarg for a stricter test.