bovid.aggregates.custom_field.repo module

class bovid.aggregates.custom_field.repo.CustomFieldRepoProtocol

Bases: RepoProtocol[CustomFieldAgg], Protocol

class bovid.aggregates.custom_field.repo.CustomFieldRepoFake

Bases: CustomFieldRepoProtocol, RepoFake[CustomFieldAgg]

__init__(storage)
Parameters:

storage (dict[int, CustomFieldAgg])

exists_by_name(field_names)
Parameters:

field_names (Sequence[str])

Return type:

bool

retrieve_by_name(field_names)
Parameters:

field_names (Sequence[str])

Return type:

Sequence[tuple[int, CustomFieldAgg]]

unregister(field_id)
Parameters:

field_id (int)

Return type:

bool

class bovid.aggregates.custom_field.repo.CustomFieldRepoSQL

Bases: CustomFieldRepoProtocol

__init__(conn)
Parameters:

conn (Connection)

Return type:

None

conn: Connection
commit()
Return type:

None

rollback()
Return type:

None

close()
Return type:

None

create(custom_field)

Create a new entity.

If an entity exists that matches this entity no new entity is created and the existing ID is returned. There is a return value that reports if this was the case.

Parameters:

  • agg – The domain model object to add to the repository.

  • custom_field (CustomFieldAgg)

Returns:

The ID of model in storage (can be new or old ID) already_exists: True if entity already existed.

Return type:

id

See also

If you want to in-place replace an existing entity but keep the same ID use the replace method if available.

If you want to update some fields of an entity see the update method.

list()

List all aggregates.

Returns:

list of tuples of (aggregate_id, aggregate)

Return type:

aggregates

retrieve(field_id)

Get an entity by reference.

Parameters:

  • id – The ID of model to retrieve.

  • field_id (int)

Returns:

The ID of model retrieved. agg: The domain model object to add to the repository.

Return type:

id

exists_by_name(field_names)
Parameters:

field_names (Sequence[str])

Return type:

bool

retrieve_by_name(field_names)
Parameters:

field_names (Sequence[str])

Return type:

Sequence[tuple[int, CustomFieldAgg]]

update(field_id, patch)

Partial update of an aggregate.

Parameters:

  • agg_id – Reference of Aggregate you want to update.

  • patch (Mapping[Hashable, Any]) – An unstructured and partial version of the Aggregate. These values will replace the current ones in the stored Aggregate.

  • field_id (int)

Returns:

If Aggregate exists the patched version, otherwise None. success: Whether the actual write operation of the patched Aggregate succeeded.

Return type:

maybe_model

See also

See the replace method if you want to replace the entire model and not just a diff.

replace(field_id, field)

Get an entity by reference.

Parameters:

  • id – The ID of model to retrieve.

  • agg – The domain model object to add to the repository.

  • field_id (int)

  • field (CustomFieldAgg)

Returns:

Reports if replacement was successful.

Return type:

success

See also

See the update method if just want to update some specific parts of the model with a patch.

unregister(field_id)
Parameters:

field_id (int)

Return type:

bool

delete(field_ids)

Delete aggregates.

Parameters:

  • agg_ids – The IDs of aggregates to delete

  • field_ids (Sequence[int])

Returns:

success

Return type:

bool

class bovid.aggregates.custom_field.repo.CustomFieldUnitOfWork

Bases: UnitOfWorkProtocol[BovidServerState, CustomFieldAgg]

repo: CustomFieldRepoFake | CustomFieldRepoSQL
__init__(repo)

Method generated by attrs for class CustomFieldUnitOfWork.

Parameters:

repo (CustomFieldRepoFake | CustomFieldRepoSQL)

Return type:

None

commit()
Return type:

None

rollback()
Return type:

None