make for query columns

Author: giladfuchs

fastapi-postgres/app/service/auth.py

@@ -32,7 +32,7 @@ def _inject_user_filter(
         return Depends(_inject_user_filter)
 
     @classmethod
-    async def authenticate_user(cls, _id: int, phone: int) -> Token:
+    async def authenticate_user(cls, _id: int, phone: str) -> Token:
         filter_query = FilterQuery(
             query=[
                 DBQuery(key="id", opt=DBOperator.eq, value=_id),

fastapi-postgres/app/service/create_data.py

@@ -16,15 +16,15 @@ class FactoryModel:
     def teacher(cls) -> dict:
         return {
             "id": randint(10, 900),
-            "phone": float(randint(10_000_000, 9_999_999_999)),
+            "phone": str(randint(10_000_000, 9_999_999_999)),
         }
 
     @classmethod
     def student(cls, teacher_id: int | None = None) -> dict:
         data = {
             "name": fake.name(),
             "grade": choice(Grade.values()),
-            "phone": float(randint(10**9, 10**10)),
+            "phone": str(randint(10**9, 10**10)),
         }
         if teacher_id is not None:
             data["teacher_id"] = teacher_id

fastapi-postgres/common/db_model/__init__.py

@@ -1,35 +1,20 @@
-import asyncio
 from contextlib import asynccontextmanager
 from typing import Any, Type
 
 from pydantic import BaseModel
-from sqlalchemy.exc import IntegrityError, OperationalError
+from sqlmodel import and_, update
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
-from sqlalchemy.orm import selectinload
-from sqlmodel import select
 from sqlmodel.ext.asyncio.session import AsyncSession
 
 from common.config import conf
+from common.db_model.query_db import QueryDB
 from common.enums import DBOperator, ModelType
 from common.serializers import BaseTable, DBQuery, FilterQuery, RowLike
 from common.utils import BaseUtils
 
-opts = {
-    DBOperator.eq: "__eq__",
-    DBOperator.ne: "__ne__",
-    DBOperator.lt: "__lt__",
-    DBOperator.le: "__le__",
-    DBOperator.gt: "__gt__",
-    DBOperator.ge: "__ge__",
-    DBOperator.like: "like",
-    DBOperator.ilike: "ilike",  # PostgreSQL only
-    DBOperator.in_: "in_",
-    DBOperator.not_in: "notin_",
-    DBOperator.is_null: "is_",
-}
-
-
-class DBModel(BaseUtils):
+
+class DBModel(BaseUtils, QueryDB):
     table: Type[BaseTable] | None = None
     payload: Type[BaseModel] | None = None
     model_type: ModelType | None = None
@@ -66,6 +51,29 @@ async def get_session(cls):
         finally:
             await session.close()
 
+    @classmethod
+    def _raise_db_error(cls, ex, row=None):
+        dump = (
+            row
+            if isinstance(row, dict)
+            else (row.model_dump() if hasattr(row, "model_dump") else None)
+        )
+        if dump:
+            cls.logger.error(f"{dump}: {ex}")
+        else:
+            cls.logger.error(str(ex))
+        if isinstance(ex, IntegrityError):
+            orig = str(getattr(ex, "orig", ""))
+            if "duplicate key value" in orig:
+                field = "unique"
+                if "Key (" in orig:
+                    try:
+                        field = orig.split("Key (")[1].split(")=")[0]
+                    except Exception:
+                        pass
+                cls.error_400(details=f"unique:{field}")
+        cls.error_400(details=ex)
+
     @classmethod
     async def add_update(cls, row: BaseTable | list[BaseTable]):
         async with cls.get_session() as session:
@@ -85,100 +93,78 @@ async def add_update(cls, row: BaseTable | list[BaseTable]):
                     await session.refresh(row)
                     return row
             except Exception as ex:
-                dump = "" if isinstance(row, list) else row.model_dump()
-                cls.logger.error(f"{dump}: {ex}")
-                if isinstance(ex, IntegrityError):
-                    orig = str(getattr(ex, "orig", ""))
-                    if "duplicate key value" in orig:
-                        field = "unique"
-                        if "Key (" in orig:
-                            try:
-                                field = orig.split("Key (")[1].split(")=")[0]
-                            except Exception:
-                                pass
-                        cls.error_400(details=f"unique:{field}")
-                cls.error_400(details=ex)
+                cls._raise_db_error(ex, row)
+
+    @classmethod
+    async def update_exist_bulk(cls, filter_query: FilterQuery, values: dict) -> int:
+        statement = update(cls.table).values(**values)
+
+        conditions = cls.generate_where_query(filter_query.query)
+
+        if conditions:
+            statement = statement.where(and_(*conditions))
+
+        try:
+            async with cls.get_session() as session:
+                results = await session.exec(statement)
+                return results.rowcount
+        except Exception as ex:
+            cls._raise_db_error(ex, values)
 
     @classmethod
     async def add_or_find_update_single(
-            cls,
-            add_or_id: str | int,
-            body: BaseModel | dict[str, Any],
-            **kwargs: Any,
+        cls,
+        add_or_id: str | int,
+        body: BaseModel,
+        **kwargs: Any,
     ) -> BaseTable:
         user_auth = kwargs.get("user_auth")
 
         if add_or_id != "add":
-            db_obj = await cls.get_by_id(_id=add_or_id, **kwargs)
+            _id = int(add_or_id)
+            query = [DBQuery(key=cls.table.id.key, opt=DBOperator.eq, value=_id)]
+
+            if user_auth:
+                query.append(
+                    DBQuery(
+                        key=conf.AUTH_PARENT_FIELD,
+                        opt=DBOperator.eq,
+                        value=user_auth.id,
+                    )
+                )
+
+            body_dict = body.model_dump()
+            values = {
+                key: value for key, value in body_dict.items() if key not in {"id"}
+            }
 
-            if db_obj and (
-                    not user_auth or user_auth.id == getattr(db_obj, conf.AUTH_PARENT_FIELD)
-            ):
-                cls.set_elements_by_dict(db_obj, body, exclude_items=["id"])
-            else:
+            updated = await cls.update_exist_bulk(
+                filter_query=FilterQuery(query=query),
+                values=values,
+            )
+
+            if not updated:
                 cls.error_400(details="not found")
 
-        else:
-            db_obj = cls.table()
-            cls.set_elements_by_dict(db_obj, body)
+            body_dict["id"] = _id
+            new_obj = cls.table(**body_dict)
 
+        else:
+            db_obj = cls.table(**body.model_dump())
             if user_auth and hasattr(db_obj, conf.AUTH_PARENT_FIELD):
                 setattr(db_obj, conf.AUTH_PARENT_FIELD, user_auth.id)
 
-        new_obj = await cls.add_update(row=db_obj)
-        return new_obj
-
-    # Generate SQLAlchemy filter conditions with AND logic from a list of DBQuery objects
-    @classmethod
-    def generate_where_query(cls, query: list[DBQuery]):
-        # Build a tuple of SQLAlchemy expressions by applying the operator (e.g., ==, >=) on each field
-        ans = tuple(
-            getattr(getattr(cls.table, q.key), opts.get(q.opt, q.opt))(q.value)
-            for q in query
-            if hasattr(cls.table, q.key)
-            and hasattr(getattr(cls.table, q.key), opts.get(q.opt, q.opt))
-        )
-        return ans
-
-    @classmethod
-    def build_query(cls, filter_query: FilterQuery, offset: int = 0, limit: int = 1000):
-        statement = select(cls.table)
-
-        if filter_query.query:
-            statement = statement.where(*cls.generate_where_query(filter_query.query))
-
-        if filter_query.relation_model:
-            statement = statement.options(selectinload("*"))
+            new_obj = await cls.add_update(row=db_obj)
 
-        if filter_query.sort:
-            try:
-                field, direction = filter_query.sort.split(":")
-                col = getattr(cls.table, field, None)
-                if col:
-                    direction = direction.lower()
-                    statement = statement.order_by(
-                        col.asc() if direction == "asc" else col.desc()
-                    )
-            except Exception:
-                raise ValueError(
-                    f"Invalid sort format: '{filter_query.sort}'. Expected 'field:asc' or 'field:desc'"
-                )
-
-        if offset:
-            statement = statement.offset(offset)
-
-        if limit:
-            statement = statement.limit(limit)
-
-        return statement
+        return new_obj
 
     @classmethod
     async def fetch_rows(
-            cls,
-            filter_query: FilterQuery = FilterQuery(),
-            offset: int = 0,
-            limit: int = 1000,
-            as_dict: bool = False,
+        cls,
+        filter_query: FilterQuery = FilterQuery(),
+        offset: int = 0,
+        limit: int = 1000,
+        as_dict: bool = False,
     ) -> RowLike | list[RowLike]:
         statement = cls.build_query(filter_query, offset, limit)
 
@@ -204,19 +190,22 @@ async def fetch_rows(
 
     @classmethod
     async def delete_rows(
-            cls, filter_query: FilterQuery = FilterQuery(), offset: int = 0
+        cls, filter_query: FilterQuery = FilterQuery(), offset: int = 0
     ):
         try:
             async with cls.get_session() as session:
+                filter_query.columns = [cls.table.id]
                 select_stmt = cls.build_query(filter_query, offset, limit=0)
                 results = (await session.exec(select_stmt)).all()
                 if not results:
                     cls.error_400(details="not delete nothing")
+
                 for obj in results:
                     await session.delete(obj)
+                return results
 
         except Exception as ex:
-            cls.error_400(details=ex)
+            cls._raise_db_error(ex)
 
     @classmethod
     async def get_by_id(cls, _id: str | int, **kwargs):