crud_base updated to allow querying only necessary columns, ORM ditched for sqlalchemy core for many of the crud functions

Author: igorbenav

src/app/api/v1/posts.py

@@ -32,6 +32,7 @@ async def write_post(
 
     post_internal_dict = post.model_dump()
     post_internal_dict["created_by_user_id"] = db_user.id
+
     post_internal = PostCreateInternal(**post_internal_dict)
     return await crud_posts.create(db=db, object=post_internal)
 
@@ -70,7 +71,7 @@ async def read_post(
     return db_post
 
 
-@router.patch("/{username}/post/{id}", response_model=PostRead)
+@router.patch("/{username}/post/{id}")
 @cache(
     "{username}_post_cache", 
     resource_id_name="id", 
@@ -95,7 +96,8 @@ async def patch_post(
     if db_post is None:
         raise HTTPException(status_code=404, detail="Post not found")
 
-    return await crud_posts.update(db=db, object=values, db_object=db_post, id=id)
+    await crud_posts.update(db=db, object=values, id=id)
+    return {"message": "Post updated"}
 
 
 @router.delete("/{username}/post/{id}")
@@ -122,13 +124,9 @@ async def erase_post(
     if db_post is None:
         raise HTTPException(status_code=404, detail="Post not found")
 
-    deleted_post = await crud_posts.delete(db=db, db_object=db_post, id=id)
-    if deleted_post.is_deleted == True:
-        message = {"message": "Post deleted"}
-    else:
-        message = {"message": "Something went wrong"}
-
-    return message
+    await crud_posts.delete(db=db, db_row=db_post, id=id)
+    
+    return {"message": "Post deleted"}
 
 
 @router.delete("/{username}/db_post/{id}", dependencies=[Depends(get_current_superuser)])
@@ -152,11 +150,4 @@ async def erase_db_post(
         raise HTTPException(status_code=404, detail="Post not found")
 
     await crud_posts.db_delete(db=db, db_object=db_post, id=id)
-    deleted_post = await crud_posts.get(db=db, id=id)
-    
-    if deleted_post is None:
-        message = {"message": "Post deleted"}
-    else:
-        message = {"message": "Something went wrong"}
-
-    return message
+    return {"message": "Post deleted from the database"}

src/app/api/v1/users.py

@@ -15,13 +15,17 @@
 router = fastapi.APIRouter(tags=["users"])
 
 @router.post("/user", response_model=UserRead, status_code=201)
-async def write_user(request: Request, user: UserCreate, db: Annotated[AsyncSession, Depends(async_get_db)]):
-    db_user = await crud_users.get(db=db, email=user.email)
-    if db_user:
+async def write_user(
+    request: Request, 
+    user: UserCreate, 
+    db: Annotated[AsyncSession, Depends(async_get_db)]
+):
+    email_row = await crud_users.exists(db=db, email=user.email)
+    if email_row:
         raise HTTPException(status_code=400, detail="Email is already registered")
 
-    db_user = await crud_users.get(db=db, username=user.username)
-    if db_user:
+    username_row = await crud_users.exists(db=db, username=user.username)
+    if username_row:
         raise HTTPException(status_code=400, detail="Username not available")
 
     user_internal_dict = user.model_dump()
@@ -56,7 +60,7 @@ async def read_user(request: Request, username: str, db: Annotated[AsyncSession,
     return db_user
 
 
-@router.patch("/user/{username}", response_model=UserRead)
+@router.patch("/user/{username}")
 async def patch_user(
     request: Request, 
     values: UserUpdate,
@@ -72,17 +76,17 @@ async def patch_user(
         raise privileges_exception
 
     if values.username != db_user.username:
-        existing_username = await crud_users.get(db=db, username=values.username)
-        if existing_username is not None:
+        existing_username = await crud_users.exists(db=db, username=values.username)
+        if existing_username:
             raise HTTPException(status_code=400, detail="Username not available")
 
     if values.email != db_user.email:
-        existing_email = await crud_users.get(db=db, email=values.email)
+        existing_email = await crud_users.exists(db=db, email=values.email)
         if existing_email:
             raise HTTPException(status_code=400, detail="Email is already registered")
 
-    db_user = await crud_users.update(db=db, object=values, db_object=db_user)
-    return db_user
+    await crud_users.update(db=db, object=values, username=username)
+    return {"message": "User updated"}
 
 
 @router.delete("/user/{username}")
@@ -99,8 +103,8 @@ async def erase_user(
     if db_user.username != current_user.username:
         raise privileges_exception
 
-    db_user = await crud_users.delete(db=db, db_object=db_user, username=username)
-    return db_user
+    await crud_users.delete(db=db, db_row=db_user, username=username)
+    return {"message": "User deleted"}
 
 
 @router.delete("/db_user/{username}", dependencies=[Depends(get_current_superuser)])
@@ -109,9 +113,9 @@ async def erase_db_user(
     username: str,
     db: Annotated[AsyncSession, Depends(async_get_db)]
 ):
-    db_user = await crud_users.get(db=db, username=username)
-    if db_user is None:
+    db_user = await crud_users.exists(db=db, username=username)
+    if not db_user:
         raise HTTPException(status_code=404, detail="User not found")
 
-    db_user = await crud_users.db_delete(db=db, db_object=db_user, username=username)
-    return db_user
+    db_user = await crud_users.db_delete(db=db, username=username)
+    return {"message": "User deleted from the database"}

src/app/crud/crud_base.py

@@ -2,8 +2,11 @@
 from datetime import datetime
 
 from pydantic import BaseModel
-from sqlalchemy import select, update
+from sqlalchemy import select, update, delete
 from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.engine.row import Row
+
+from .helper import _extract_matching_columns_from_schema, _extract_matching_columns_from_kwargs
 
 ModelType = TypeVar("ModelType")
 CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
@@ -24,78 +27,68 @@ async def create(
         await db.commit()
         return db_object
 
-    async def get(self, db: AsyncSession, **kwargs) -> ModelType | None:
-        query = select(self._model).filter_by(**kwargs)
-        result = await db.execute(query)
-        return result.scalar_one_or_none()
+    async def get(self, db: AsyncSession, schema_to_select: Type[BaseModel] | None = None, **kwargs) -> ModelType | None:
+        to_select = _extract_matching_columns_from_schema(model=self._model, schema=schema_to_select)
+        stmt = select(*to_select) \
+            .filter_by(**kwargs)
+        
+        result = await db.execute(stmt)
+        return result.first()
 
-    async def get_multi(
-            self, db: AsyncSession, offset: int = 0, limit: int = 100, **kwargs
-    ) -> List[ModelType]:
-        query = select(self._model) \
+    async def get_multi(self, db: AsyncSession, offset: int = 0, limit: int = 100, schema_to_select: Type[BaseModel] | None = None, **kwargs) -> List[ModelType]:
+        to_select = _extract_matching_columns_from_schema(model=self._model, schema=schema_to_select)
+        stmt = select(*to_select) \
             .filter_by(**kwargs) \
             .offset(offset) \
             .limit(limit)
 
-        result = await db.execute(query)
-        return result.scalars().all()
-
-    async def update(
-            self,
-            db: AsyncSession,
-            object: Union[UpdateSchemaType, Dict[str, Any]],
-            db_object: ModelType | None = None,
-            **kwargs
-    ) -> ModelType | None:
-        db_object = db_object or await self.get(db=db, **kwargs)
-        if db_object:
-            if isinstance(object, dict):
-                update_data = object
-            else:
-                update_data = object.model_dump(exclude_unset=True)
-            
-            update_data.update({"updated_at": datetime.utcnow()})
-            for field in object.__dict__:
-                if field in update_data:
-                    setattr(db_object, field, update_data[field])
-            db.add(db_object)
-            await db.commit()
-
-        return db_object
+        result = await db.execute(stmt)
+        return result.all()
+    
+    async def exists(self, db: AsyncSession, **kwargs) -> bool:
+        to_select = _extract_matching_columns_from_kwargs(model=self._model, kwargs=kwargs)
+        stmt = select(*to_select) \
+            .filter_by(**kwargs) \
+            .limit(1)
+        result = await db.execute(stmt)
+        
+        return result.first() is not None
+        
+    async def update(self, db: AsyncSession, object: Union[UpdateSchemaType, Dict[str, Any]], **kwargs) -> ModelType | None:
+        if isinstance(object, dict):
+            update_data = object
+        else:
+            update_data = object.model_dump(exclude_unset=True)
+        update_data["updated_at"] = datetime.utcnow()
 
-    async def db_delete(
-            self,
-            db: AsyncSession,
-            db_object: ModelType | None = None,
-            **kwargs
-    ):
-        db_object = db_object or await self.get(db=db, **kwargs)
-        await db.delete(db_object)
+        stmt = update(self._model) \
+            .filter_by(**kwargs) \
+            .values(update_data)
+        
+        await db.execute(stmt)
         await db.commit()
 
-        return db_object
+    async def db_delete(self, db: AsyncSession, **kwargs):
+        stmt = delete(self._model).filter_by(**kwargs)
+        await db.execute(stmt)
+        await db.commit()
 
-    async def delete(
-            self,
-            db: AsyncSession,
-            db_object: ModelType | None = None,
-            **kwargs
-    ) -> ModelType | None:
-        db_object = db_object or await self.get(db=db, **kwargs)
-        if db_object:
-            if "is_deleted" in db_object.__dict__.keys():
+    async def delete(self, db: AsyncSession, db_row: Row | None = None, **kwargs) -> ModelType | None:
+        db_row = db_row or await self.get(db=db, **kwargs)
+        if db_row:
+            if "is_deleted" in db_row:
                 object_dict = {
                     "is_deleted": True,
                     "deleted_at": datetime.utcnow()
                 }
-                query = update(self._model) \
+                stmt = update(self._model) \
                     .filter_by(**kwargs) \
                     .values(object_dict)
 
-                await db.execute(query)
+                await db.execute(stmt)
                 await db.commit()
-                await db.refresh(db_object)
-            else:
-                db_object = await self.db_delete(db=db, db_object=db_object, **kwargs)
 
-        return db_object
+            else:
+                stmt = delete(self._model).filter_by(**kwargs)
+                await db.execute(stmt)
+                await db.commit()

src/app/crud/helper.py

@@ -0,0 +1,42 @@
+from typing import Any, List, Type
+
+from pydantic import BaseModel
+
+from app.core.database import Base
+
+def _extract_matching_columns_from_schema(model: Type[Base], schema: Type[BaseModel] | None) -> List[Any]:
+    """
+    Retrieves a list of ORM column objects from a SQLAlchemy model that match the field names in a given Pydantic schema.
+
+    Parameters
+    ----------
+    model: Type[Base]
+        The SQLAlchemy ORM model containing columns to be matched with the schema fields.
+    schema: Type[BaseModel]
+        The Pydantic schema containing field names to be matched with the model's columns.
+
+    Returns
+    -------
+    List[Any]
+        A list of ORM column objects from the model that correspond to the field names defined in the schema.
+    """
+    column_list = list(model.__table__.columns)
+    if schema is not None:
+        schema_fields = schema.__fields__.keys()
+        column_list = []
+        for column_name in schema_fields:
+            if hasattr(model, column_name):
+                column_list.append(getattr(model, column_name))
+    
+    return column_list
+
+
+def _extract_matching_columns_from_kwargs(model: Type[Base], kwargs: dict) -> List[Any]:
+    if kwargs is not None:
+        kwargs_fields = kwargs.keys()
+        column_list = []
+        for column_name in kwargs_fields:
+            if hasattr(model, column_name):
+                column_list.append(getattr(model, column_name))
+    
+    return column_list