GPQBot

1
from typing import Any, Dict, TypeVar
2

3
from sqlalchemy import delete, insert, select, update
4
from sqlalchemy.inspection import inspect
5

6
from app.db.sqlalchemy import AsyncSession
7

8
T = TypeVar("T")  # noqa: WPS111
9

10

11
class CRUD:
12
    """CRUD operations for models."""
13

14
    def __init__(self, session: AsyncSession, cls_model: Any):
15
        self._session = session
16
        self._cls_model = cls_model
17

18
    async def create(self, *, model_data: Dict[str, Any]) -> Any:
19
        """Create object."""
20
        query = insert(self._cls_model).values(**model_data)
21

22
        res = await self._session.execute(query)  # type: ignore
23
        return res.inserted_primary_key  # type: ignore
24

25
    async def update(
26
        self,
27
        *,
28
        pkey_val: Any,
29
        model_data: Dict[str, Any],
30
    ) -> None:
31
        """Update object by primary key."""
32
        primary_key = inspect(self._cls_model).primary_key[0]
33
        query = (
34
            update(self._cls_model)  # type: ignore
35
            .where(primary_key == pkey_val)
36
            .values(**model_data)
37
            .execution_options(synchronize_session="fetch")
38
        )
39

40
        await self._session.execute(query)
41

42
    async def delete(self, *, pkey_val: Any) -> None:
43
        """Delete object by primary key value."""
44
        primary_key = inspect(self._cls_model).primary_key[0].name
45
        query = (
46
            delete(self._cls_model)  # type: ignore
47
            .where(getattr(self._cls_model, primary_key) == pkey_val)
48
            .execution_options(synchronize_session="fetch")
49
        )
50

51
        await self._session.execute(query)
52

53
    async def get(self, *, pkey_val: Any) -> Any:
54
        """Get object by primary key."""
55
        primary_key = inspect(self._cls_model).primary_key[0]
56
        query = select(self._cls_model).where(primary_key == pkey_val)
57

58
        rows = await self._session.execute(query)  # type: ignore
59
        return rows.scalars().one()
60

61
    async def get_or_none(self, *, pkey_val: Any) -> Any:
62
        """Get object by primary key or none."""
63
        primary_key = inspect(self._cls_model).primary_key[0]
64
        query = select(self._cls_model).where(primary_key == pkey_val)
65

66
        rows = await self._session.execute(query)  # type: ignore
67
        return rows.scalar()
68

69
    async def all(
70
        self,
71
    ) -> Any:
72
        """Get all objects by db model."""
73
        query = select(self._cls_model)
74

75
        rows = await self._session.execute(query)
76
        return rows.scalars().all()
77

78
    async def get_by_field(self, *, field: str, field_value: Any) -> Any:
79
        """Return objects from db with condition field=val."""
80
        query = select(self._cls_model).where(
81
            getattr(self._cls_model, field) == field_value
82
        )
83

84
        rows = await self._session.execute(query)  # type: ignore
85
        return rows.scalars().all()