add DBOperator

Author: giladfuchs

fastapi-postgres/app/service/auth.py

@@ -6,6 +6,7 @@
 
 from common.config import conf
 from common.db_model.models import TeacherModel
+from common.enums import DBOperator
 from common.serializers import DBQuery, FilterQuery, Token
 
 oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")
@@ -23,7 +24,7 @@ def _inject_user_filter(
             user_auth: TeacherModel.table = Depends(Auth.jwt_required),
         ) -> FilterQuery:
             filter_query.query.append(
-                DBQuery(key=field_name, opt="eq", value=int(user_auth.id))
+                DBQuery(key=field_name, opt=DBOperator.eq, value=int(user_auth.id))
             )
             return filter_query
 
@@ -41,8 +42,8 @@ def error_authenticate(cls) -> HTTPException:
     async def authenticate_user(cls, _id: int, phone: int) -> Token:
         filter_query = FilterQuery(
             query=[
-                DBQuery(key="id", opt="eq", value=_id),
-                DBQuery(key="phone", opt="eq", value=phone),
+                DBQuery(key="id", opt=DBOperator.eq, value=_id),
+                DBQuery(key="phone", opt=DBOperator.eq, value=phone),
             ]
         )
 

fastapi-postgres/app/service/create_data.py

@@ -16,7 +16,7 @@ class FactoryModel:
     def teacher(cls) -> dict:
         return {
             "id": randint(10, 900),
-            "phone": float(randint(10_000_000, 9_999_999_999))
+            "phone": float(randint(10_000_000, 9_999_999_999)),
         }
 
     @classmethod

fastapi-postgres/common/config.py

@@ -1,4 +1,5 @@
 import os
+
 from pydantic_settings import BaseSettings
 
 # load .env locally if present (won't affect server)

fastapi-postgres/common/db_model/__init__.py

@@ -8,24 +8,24 @@
 from sqlmodel.ext.asyncio.session import AsyncSession
 
 from common.config import conf
-from common.enums import ModelType
+from common.enums import DBOperator, ModelType
 from common.serializers import BaseTable, DBQuery, FilterQuery, RowLike
 from common.utils.errors import ErrorService
 from common.utils.log import logger
 from common.utils.parse_obj import set_elements_by_dict
 
 opts = {
-    "eq": "__eq__",
-    "ne": "__ne__",
-    "lt": "__lt__",
-    "le": "__le__",
-    "gt": "__gt__",
-    "ge": "__ge__",
-    "like": "like",
-    "ilike": "ilike",  # PostgreSQL only
-    "in": "in_",
-    "not_in": "notin_",
-    "is_null": "is_",
+    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_",
 }
 
 
@@ -195,7 +195,7 @@ async def delete_rows(
     @classmethod
     async def get_by_id(cls, _id: str | int, **kwargs):
         filter_query = FilterQuery(
-            query=[DBQuery(key=cls.table.id.key, opt="eq", value=int(_id))],
+            query=[DBQuery(key=cls.table.id.key, opt=DBOperator.eq, value=int(_id))],
             relation_model=kwargs.get("relation_model", False),
         )
         return await cls.fetch_rows(filter_query=filter_query, limit=1)

fastapi-postgres/common/enums/__init__.py

@@ -17,6 +17,20 @@ class ModelType(EnumBase):
     assignment = "assignment"
 
 
+class DBOperator(EnumBase):
+    eq = "eq"
+    ne = "ne"
+    lt = "lt"
+    le = "le"
+    gt = "gt"
+    ge = "ge"
+    like = "like"
+    ilike = "ilike"  # PostgreSQL only
+    in_ = "in_"
+    not_in = "not_in"
+    is_null = "is_null"
+
+
 class Grade(EnumBase):
     A = "A"
     B = "B"

fastapi-postgres/common/serializers/__init__.py

@@ -1,13 +1,13 @@
-from __future__ import annotations
-
-from typing import Any,  List, Optional, Set, TypeAlias
+from typing import Any, List, Optional, Set, TypeAlias
 
 from pydantic import BaseModel
 from pydantic import Field as PydanticField
 from sqlalchemy.inspection import inspect as sa_inspect
 from sqlalchemy.orm.attributes import NO_VALUE
 from sqlmodel import Field, SQLModel
 
+from common.enums import DBOperator
+
 
 class BaseTable(SQLModel):
     def model_dump(
@@ -87,7 +87,7 @@ def __eq__(self, other):
 
 
 class DBQuery(BaseModel):
-    opt: str
+    opt: DBOperator
     key: str
     value: Any
 

fastapi-postgres/tests/test_all.py

@@ -4,7 +4,7 @@
 from httpx import AsyncClient
 
 from app.service.create_data import FactoryModel
-from common.enums import ModelType
+from common.enums import DBOperator, ModelType
 
 
 class Api:
@@ -47,7 +47,7 @@ async def fetch_rows(self, *, model: ModelType, body: dict | None = None) -> lis
         return data
 
     async def delete_row(self, *, model: ModelType, _id: int | str) -> None:
-        body = {"query": [{"key": "id", "value": _id, "opt": "eq"}]}
+        body = {"query": [{"key": "id", "value": _id, "opt": DBOperator.eq}]}
         r = await self.client.request(
             "DELETE", f"/{model}", content=json.dumps(body), headers=self.headers
         )