Service

infra.adapters.service

Summary

Generic CRUD service adaptor over a Mongo repository.

---

Bridge pydantic models and mongo-ops documents, exposing create, read, update, patch, delete, and search operations for every domain that needs them.

Classes

CrudsServiceAdaptor

CrudsServiceAdaptor(*, repo=None)

Bases: Generic[Model, Schema]

Adapt a generic Mongo repository to a model/schema CRUD service.

Attributes:

Name	Type	Description
repo	BaseRepository	Repository backing all CRUD operations.
model	Type[Model]	Pydantic model type handled by this service.
schema	Type[Schema]	Document schema type persisted by the repository.

Initialize the adaptor with a backing repository.

Parameters:

Name	Type	Description	Default
repo	BaseRepository	Repository used for all operations.	None

Functions

create async

create(data: Model | dict) -> Schema

Create a document from a model or raw mapping.

Parameters:

Name	Type	Description	Default
data	Model | dict	Model instance or field mapping to persist.	required

Returns:

Name	Type	Description
Schema	Schema	Persisted document.

delete async

delete(id: PyObjectId) -> bool

Delete a document by primary key.

Parameters:

Name	Type	Description	Default
id	PyObjectId	Document identifier.	required

Returns:

Name	Type	Description
bool	bool	True when a document was deleted.

model_to_dict staticmethod

model_to_dict(model_obj: Model) -> dict

Serialize a model into a sparse dictionary.

Parameters:

Name	Type	Description	Default
model_obj	Model	Model instance to serialize.	required

Returns:

Name	Type	Description
dict	dict	Fields that are set and not None.

patch async

patch(id: PyObjectId, data: dict) -> Schema | None

Partially update a document by primary key.

Parameters:

Name	Type	Description	Default
id	PyObjectId	Document identifier.	required
data	dict	Fields to patch.	required

Returns:

Type	Description
Schema | None	Schema | None: Updated document, or None when absent.

read async

read(id: PyObjectId) -> Schema | None

Read a document by primary key.

Parameters:

Name	Type	Description	Default
id	PyObjectId	Document identifier.	required

Returns:

Type	Description
Schema | None	Schema | None: Matching document, or None when absent.

search async

search(
    filter: dict | None = None, *args, **kwargs
) -> list[Schema]

List documents matching an optional filter.

Parameters:

Name	Type	Description	Default
filter	dict | None	Query filter passed to the repository.	None

Returns:

Type	Description
list[Schema]	list[Schema]: Matching documents.

Notes

Extra positional and keyword arguments are forwarded to the repository.

update async

update(id: PyObjectId, data: Model | dict) -> Schema | None

Replace a document by primary key.

Parameters:

Name	Type	Description	Default
id	PyObjectId	Document identifier.	required
data	Model | dict	Replacement model or field mapping.	required

Returns:

Type	Description
Schema | None	Schema | None: Updated document, or None when absent.