Skip to content

Base DB CRUDs

BaseCRUD

Base class that implements CRUD for the database using TortoiseORM.

Attributes:

Name Type Description
model Model

Tortoise Model on which CRUD operations are applied.
schema Schema

PydanticModel generated from the Tortoise Model.
Source code in ms_core/bases/base_crud.py 
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
class BaseCRUD[Model: TortoiseModel, Schema: PydanticModel]:
    """Base class that implements CRUD for the database using TortoiseORM.

    Attributes:
        model: Tortoise Model on which CRUD operations are applied.
        schema: PydanticModel generated from the Tortoise Model.
    """

    model: Model
    schema: Schema

    @classmethod
    async def create(cls, payload: PydanticModel, **kwargs) -> Schema:
        """Implements the create operation.

        Args:
            payload: The data to create the instance.
            **kwargs: Additional arguments for the creation process.

        Returns:
            The created instance as a PydanticModel.
        """
        instance = await cls.model.create(
            **payload.model_dump(exclude_none=True), **kwargs
        )

        return await cls.schema.from_tortoise_orm(instance)

    @classmethod
    async def get_or_create(cls, **kwargs) -> tuple[Schema, bool]:
        """Implements the get or create operation.

        If the item is not found, it will try to create it from the provided kwargs.

        Args:
            **kwargs: Arguments to search for and create the instance if not found.

        Returns:
            The found or created instance and a boolean indicating whether it was created.

        Raises:
            IntegrityError: If creation fails.
        """
        instance, status = await cls.model.get_or_create(**kwargs)

        return await cls.schema.from_tortoise_orm(instance), status

    @classmethod
    @multimethod
    async def get_by_id(cls, id_: int) -> Schema | None:
        """Fetches an item by its ID.

        Args:
            id_: The ID of the item.

        Returns:
            The corresponding PydanticModel or None if not found.
        """
        return await cls.get_by(id=id_)

    @classmethod
    async def get_by(cls, **kwargs) -> Schema | None:
        """Fetches an item based on the provided kwargs.

        Args:
            **kwargs: Criteria to search for the item.

        Returns:
            The corresponding PydanticModel or None if not found.
        """
        if not (instance := await cls.model.get_or_none(**kwargs)):
            return None

        return await cls.schema.from_tortoise_orm(instance)

    @classmethod
    @multimethod
    async def get_all(
            cls,
            prefetch: bool = False,
            limit: int = 50,
            offset: int = 0,
            **extra_filters
    ) -> list[Schema]:
        """Bulk fetches items from the database.

        Args:
            prefetch: Whether to prefetch related data. Defaults to False.
            limit: The number of items to fetch. Defaults to 50.
            offset: The offset from which to start fetching. Defaults to 0.
            **extra_filters: Additional filters to apply.

        Returns:
            A list of fetched PydanticModels. Empty list if no items are found.
        """
        query = cls.model.all().filter(**extra_filters).offset(offset).limit(limit)
        result = []

        for item in await query:
            if prefetch:
                result.append(await cls.schema.from_tortoise_orm(item))
            else:
                result.append(cls.schema.model_construct(**item.__dict__))

        return result

    @classmethod
    async def filter_by(cls, **kwargs) -> list[Schema] | None:
        """Exposes the filter method of Tortoise Model.

        Args:
            **kwargs: Criteria to filter the items.

        Returns:
            A list of matching items or None if not found.
        """
        try:
            return [
                await cls.schema.from_tortoise_orm(item)
                for item in await cls.model.filter(**kwargs)
            ]

        except DoesNotExist as _:
            return None

    @classmethod
    async def update_by(cls, payload: Schema | dict, **kwargs) -> Schema | None:
        """Implements the update operation.

        Args:
            payload: The data to update the instance.
            **kwargs: Criteria to find the item to be updated.

        Returns:
            The updated instance or None if not found.
        """
        if not (instance := await cls.model.get_or_none(**kwargs)):
            return None

        as_dict = (
            payload.items()
            if isinstance(payload, dict)
            else payload.model_dump().items()
        )

        await instance.update_from_dict(
            {key: value for key, value in as_dict if value is not None}
        ).save()
        return await cls.schema.from_tortoise_orm(instance)

    @classmethod
    async def delete_by(cls, **kwargs) -> bool:
        """Implements the delete operation.

        Deletes an item based on the provided kwargs.

        Args:
            **kwargs: Search parameters to find the item to delete.

        Returns:
            True if the item was successfully deleted, False otherwise.
        """
        if not (instance := await cls.model.get_or_none(**kwargs)):
            return False

        await instance.delete()
        return True

create(payload, **kwargs) async classmethod

Implements the create operation.

Parameters:

Name Type Description Default
payload PydanticModel

The data to create the instance.

 required
**kwargs

Additional arguments for the creation process.

 {}

Returns:

Type Description
Schema

The created instance as a PydanticModel.
Source code in ms_core/bases/base_crud.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
@classmethod
async def create(cls, payload: PydanticModel, **kwargs) -> Schema:
    """Implements the create operation.

    Args:
        payload: The data to create the instance.
        **kwargs: Additional arguments for the creation process.

    Returns:
        The created instance as a PydanticModel.
    """
    instance = await cls.model.create(
        **payload.model_dump(exclude_none=True), **kwargs
    )

    return await cls.schema.from_tortoise_orm(instance)

delete_by(**kwargs) async classmethod

Implements the delete operation.

Deletes an item based on the provided kwargs.

Parameters:

Name Type Description Default
**kwargs

Search parameters to find the item to delete.

 {}

Returns:

Type Description
bool

True if the item was successfully deleted, False otherwise.
Source code in ms_core/bases/base_crud.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
@classmethod
async def delete_by(cls, **kwargs) -> bool:
    """Implements the delete operation.

    Deletes an item based on the provided kwargs.

    Args:
        **kwargs: Search parameters to find the item to delete.

    Returns:
        True if the item was successfully deleted, False otherwise.
    """
    if not (instance := await cls.model.get_or_none(**kwargs)):
        return False

    await instance.delete()
    return True

filter_by(**kwargs) async classmethod

Exposes the filter method of Tortoise Model.

Parameters:

Name Type Description Default
**kwargs

Criteria to filter the items.

 {}

Returns:

Type Description
list[Schema] | None

A list of matching items or None if not found.
Source code in ms_core/bases/base_crud.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
@classmethod
async def filter_by(cls, **kwargs) -> list[Schema] | None:
    """Exposes the filter method of Tortoise Model.

    Args:
        **kwargs: Criteria to filter the items.

    Returns:
        A list of matching items or None if not found.
    """
    try:
        return [
            await cls.schema.from_tortoise_orm(item)
            for item in await cls.model.filter(**kwargs)
        ]

    except DoesNotExist as _:
        return None

get_all(prefetch=False, limit=50, offset=0, **extra_filters) async classmethod

Bulk fetches items from the database.

Parameters:

Name Type Description Default
prefetch bool

Whether to prefetch related data. Defaults to False.

 False
limit int

The number of items to fetch. Defaults to 50.

 50
offset int

The offset from which to start fetching. Defaults to 0.

 0
**extra_filters

Additional filters to apply.

 {}

Returns:

Type Description
list[Schema]

A list of fetched PydanticModels. Empty list if no items are found.
Source code in ms_core/bases/base_crud.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
@classmethod
@multimethod
async def get_all(
        cls,
        prefetch: bool = False,
        limit: int = 50,
        offset: int = 0,
        **extra_filters
) -> list[Schema]:
    """Bulk fetches items from the database.

    Args:
        prefetch: Whether to prefetch related data. Defaults to False.
        limit: The number of items to fetch. Defaults to 50.
        offset: The offset from which to start fetching. Defaults to 0.
        **extra_filters: Additional filters to apply.

    Returns:
        A list of fetched PydanticModels. Empty list if no items are found.
    """
    query = cls.model.all().filter(**extra_filters).offset(offset).limit(limit)
    result = []

    for item in await query:
        if prefetch:
            result.append(await cls.schema.from_tortoise_orm(item))
        else:
            result.append(cls.schema.model_construct(**item.__dict__))

    return result

get_by(**kwargs) async classmethod

Fetches an item based on the provided kwargs.

Parameters:

Name Type Description Default
**kwargs

Criteria to search for the item.

 {}

Returns:

Type Description
Schema | None

The corresponding PydanticModel or None if not found.
Source code in ms_core/bases/base_crud.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
@classmethod
async def get_by(cls, **kwargs) -> Schema | None:
    """Fetches an item based on the provided kwargs.

    Args:
        **kwargs: Criteria to search for the item.

    Returns:
        The corresponding PydanticModel or None if not found.
    """
    if not (instance := await cls.model.get_or_none(**kwargs)):
        return None

    return await cls.schema.from_tortoise_orm(instance)

get_by_id(id_) async classmethod

Fetches an item by its ID.

Parameters:

Name Type Description Default
id_ int

The ID of the item.

 required

Returns:

Type Description
Schema | None

The corresponding PydanticModel or None if not found.
Source code in ms_core/bases/base_crud.py 
54
55
56
57
58
59
60
61
62
63
64
65
@classmethod
@multimethod
async def get_by_id(cls, id_: int) -> Schema | None:
    """Fetches an item by its ID.

    Args:
        id_: The ID of the item.

    Returns:
        The corresponding PydanticModel or None if not found.
    """
    return await cls.get_by(id=id_)

get_or_create(**kwargs) async classmethod

Implements the get or create operation.

If the item is not found, it will try to create it from the provided kwargs.

Parameters:

Name Type Description Default
**kwargs

Arguments to search for and create the instance if not found.

 {}

Returns:

Type Description
tuple[Schema, bool]

The found or created instance and a boolean indicating whether it was created.

Raises:

Type Description
IntegrityError

If creation fails.
Source code in ms_core/bases/base_crud.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
@classmethod
async def get_or_create(cls, **kwargs) -> tuple[Schema, bool]:
    """Implements the get or create operation.

    If the item is not found, it will try to create it from the provided kwargs.

    Args:
        **kwargs: Arguments to search for and create the instance if not found.

    Returns:
        The found or created instance and a boolean indicating whether it was created.

    Raises:
        IntegrityError: If creation fails.
    """
    instance, status = await cls.model.get_or_create(**kwargs)

    return await cls.schema.from_tortoise_orm(instance), status

update_by(payload, **kwargs) async classmethod

Implements the update operation.

Parameters:

Name Type Description Default
payload Schema | dict

The data to update the instance.

 required
**kwargs

Criteria to find the item to be updated.

 {}

Returns:

Type Description
Schema | None

The updated instance or None if not found.
Source code in ms_core/bases/base_crud.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
@classmethod
async def update_by(cls, payload: Schema | dict, **kwargs) -> Schema | None:
    """Implements the update operation.

    Args:
        payload: The data to update the instance.
        **kwargs: Criteria to find the item to be updated.

    Returns:
        The updated instance or None if not found.
    """
    if not (instance := await cls.model.get_or_none(**kwargs)):
        return None

    as_dict = (
        payload.items()
        if isinstance(payload, dict)
        else payload.model_dump().items()
    )

    await instance.update_from_dict(
        {key: value for key, value in as_dict if value is not None}
    ).save()
    return await cls.schema.from_tortoise_orm(instance)

I18nCRUD

Bases: BaseCRUD[Model, Schema]

Alternated CRUD for i18n models

Source code in ms_core/bases/i18n_crud.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
class I18nCRUD[Model: TortoiseModel, Schema: PydanticModel](BaseCRUD[Model, Schema]):
    """Alternated CRUD for i18n models"""

    @classmethod
    @BaseCRUD.get_by_id.register
    async def get_by_id(cls, id_: int, lang: str) -> Schema | None:
        return await cls.get_by(id=id_, tuple_lang=lang)

    @classmethod
    @BaseCRUD.get_all.register
    async def get_all(cls, lang: str = None, *args, **kwargs) -> list[Model]:
        return await super().get_all(*args, **kwargs, tuple_lang=lang)