fix: update Pydantic schemas for improved validation

Author: Pythonian

api/v1/routes/blog.py

@@ -1,15 +1,16 @@
 import logging
 
 from fastapi import APIRouter, Depends, HTTPException, Query, status
+from fastapi.exceptions import RequestValidationError
 from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.orm import Session
 
 from api.db.database import get_db
 from api.v1.schemas.blog import (
-    BlogCreateSchema,
-    BlogListResponseSchema,
-    BlogResponseSchema,
-    BlogUpdateSchema,
+    BlogCreate,
+    BlogListResponse,
+    BlogResponse,
+    BlogUpdate,
 )
 from api.v1.services.blog import BlogService
 
@@ -20,13 +21,13 @@
 
 @blog.post(
     "",
-    response_model=BlogResponseSchema,
+    response_model=BlogResponse,
     status_code=status.HTTP_201_CREATED,
 )
 async def create_blog(
-    blog: BlogCreateSchema,
+    blog: BlogCreate,
     db: Session = Depends(get_db),
-) -> BlogResponseSchema:
+) -> BlogResponse:
     try:
         return BlogService.create_blog(db, blog)
     except ValueError as e:
@@ -51,14 +52,14 @@ async def create_blog(
 
 @blog.get(
     "",
-    response_model=BlogListResponseSchema,
+    response_model=BlogListResponse,
     status_code=status.HTTP_200_OK,
 )
 async def list_blog(
     page: int = Query(1, ge=1),
     page_size: int = Query(10, ge=1, le=100),
     db: Session = Depends(get_db),
-) -> BlogListResponseSchema:
+) -> BlogListResponse:
     try:
         return BlogService.list_blog(db, page, page_size)
     except SQLAlchemyError as e:
@@ -77,13 +78,13 @@ async def list_blog(
 
 @blog.get(
     "/{id}",
-    response_model=BlogResponseSchema,
+    response_model=BlogResponse,
     status_code=status.HTTP_200_OK,
 )
 async def read_blog(
     id: int,
     db: Session = Depends(get_db),
-) -> BlogResponseSchema:
+) -> BlogResponse:
     try:
         return BlogService.read_blog(db, id)
     except ValueError as e:
@@ -108,16 +109,22 @@ async def read_blog(
 
 @blog.patch(
     "/{id}",
-    response_model=BlogResponseSchema,
+    response_model=BlogResponse,
     status_code=status.HTTP_200_OK,
 )
 async def update_blog(
     id: int,
-    blog_update: BlogUpdateSchema,
+    blog_update: BlogUpdate,
     db: Session = Depends(get_db),
-) -> BlogResponseSchema:
+) -> BlogResponse:
     try:
         return BlogService.update_blog(db, id, blog_update)
+    except RequestValidationError as e:
+        logger.warning(f"Validation error: {e}")
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail=e.errors(),
+        )
     except ValueError as e:
         logger.warning(str(e))
         raise HTTPException(

api/v1/schemas/blog.py

@@ -6,35 +6,18 @@
 from pydantic import BaseModel, Field
 
 
-class BlogResponseSchema(BaseModel):
-    """Schema for representing a blog post in the API response."""
-
-    id: int
-    title: str
-    excerpt: str
-    content: str
-    image_url: str
-    created_at: datetime
-    updated_at: datetime
-
-    model_config = {"from_attributes": True}
-
-
-class BlogCreateSchema(BaseModel):
-    """Schema for creating a new blog post.
-
-    This represents the data structure required when a client
-    submits a request to create a new blog post.
-    """
+class BlogBase(BaseModel):
+    """Base schema for blog post data."""
 
     title: str = Field(
         ...,
-        min_length=1,
+        min_length=10,
         max_length=255,
         description="Title of the blog post",
     )
     excerpt: str = Field(
         ...,
+        min_length=20,
         max_length=300,
         description="Short excerpt of the blog post",
     )
@@ -47,9 +30,44 @@ class BlogCreateSchema(BaseModel):
         max_length=255,
         description="URL of the blog post image",
     )
+    is_deleted: Optional[bool] = False
+
+
+class BlogCreate(BlogBase):
+    """Schema for creating a new blog post."""
+
+    pass
+
 
+class BlogUpdate(BaseModel):
+    """Schema for updating an existing blog post."""
 
-class BlogListItemResponseSchema(BaseModel):
+    title: Optional[str] = Field(
+        None,
+        min_length=10,
+        max_length=255,
+        description="Title of the blog post",
+    )
+    excerpt: Optional[str] = Field(
+        None,
+        min_length=20,
+        max_length=300,
+        description="Short excerpt of the blog post",
+    )
+    content: Optional[str]
+    image_url: Optional[str]
+    is_deleted: Optional[bool] = False
+
+
+class BlogResponse(BlogBase):
+    """Schema for returning blog post data."""
+
+    id: int
+    created_at: datetime
+    updated_at: datetime
+
+
+class BlogListItemResponse(BaseModel):
     """Schema for representing a blog post item in the list response."""
 
     id: int
@@ -58,24 +76,11 @@ class BlogListItemResponseSchema(BaseModel):
     image_url: str
     created_at: datetime
 
-    model_config = {"from_attributes": True}
-
 
-class BlogListResponseSchema(BaseModel):
+class BlogListResponse(BaseModel):
     """Schema for representing a blog post listing in the API response."""
 
     count: int
     next: Optional[str]
     previous: Optional[str]
-    results: List[BlogListItemResponseSchema]
-
-    model_config = {"from_attributes": True}
-
-
-class BlogUpdateSchema(BaseModel):
-    title: str
-    excerpt: str
-    content: str
-    image_url: str
-
-    model_config = {"from_attributes": True}
+    results: List[BlogListItemResponse]

api/v1/services/blog.py

@@ -3,20 +3,20 @@
 
 from api.v1.models.blog import Blog
 from api.v1.schemas.blog import (
-    BlogCreateSchema,
-    BlogListItemResponseSchema,
-    BlogListResponseSchema,
-    BlogResponseSchema,
-    BlogUpdateSchema,
+    BlogCreate,
+    BlogListItemResponse,
+    BlogListResponse,
+    BlogResponse,
+    BlogUpdate,
 )
 
 
 class BlogService:
     @staticmethod
     def create_blog(
         db: Session,
-        blog: BlogCreateSchema,
-    ) -> BlogResponseSchema:
+        blog: BlogCreate,
+    ) -> BlogResponse:
         existing_blog = db.query(Blog).filter(Blog.title == blog.title).first()
         if existing_blog:
             raise ValueError("A blog post with this title already exists.")
@@ -31,7 +31,7 @@ def create_blog(
         db.commit()
         db.refresh(new_blog)
 
-        return BlogResponseSchema(
+        return BlogResponse(
             id=new_blog.id,
             title=new_blog.title,
             excerpt=new_blog.excerpt,
@@ -46,7 +46,7 @@ def list_blog(
         db: Session,
         page: int,
         page_size: int,
-    ) -> BlogListResponseSchema:
+    ) -> BlogListResponse:
         offset = (page - 1) * page_size
         query = (
             db.query(Blog)
@@ -65,7 +65,7 @@ def list_blog(
             prev_page = f"/api/v1/blogs?page={page - 1}&page_size={page_size}"
 
         results = [
-            BlogListItemResponseSchema(
+            BlogListItemResponse(
                 id=blog.id,
                 title=blog.title,
                 excerpt=blog.excerpt,
@@ -75,7 +75,7 @@ def list_blog(
             for blog in blogs
         ]
 
-        return BlogListResponseSchema(
+        return BlogListResponse(
             count=total_count,
             next=next_page,
             previous=prev_page,
@@ -86,7 +86,7 @@ def list_blog(
     def read_blog(
         db: Session,
         id: int,
-    ) -> BlogResponseSchema:
+    ) -> BlogResponse:
         blog = (
             db.query(Blog)
             .filter(
@@ -98,14 +98,14 @@ def read_blog(
         if not blog:
             raise ValueError("Blog post not found.")
 
-        return BlogResponseSchema.model_validate(blog.__dict__)
+        return BlogResponse.model_validate(blog.__dict__)
 
     @staticmethod
     def update_blog(
         db: Session,
         id: int,
-        blog_update: BlogUpdateSchema,
-    ) -> BlogResponseSchema:
+        blog_update: BlogUpdate,
+    ) -> BlogResponse:
         blog = (
             db.query(Blog)
             .filter(
@@ -137,7 +137,7 @@ def update_blog(
         db.commit()
         db.refresh(blog)
 
-        return BlogResponseSchema(
+        return BlogResponse(
             id=blog.id,
             title=blog.title,
             excerpt=blog.excerpt,

tests/v1/blog/test_update_blog.py

@@ -134,15 +134,32 @@ def test_update_blog_internal_server_error(db_session_mock):
     assert response.json()["detail"] == "Internal server error."
 
 
-def test_update_blog_invalid_data():
-    """Send requests with invalid data and check for validation errors."""
+def test_update_blog_invalid_data(db_session_mock):
+    """Ensure that the endpoint returns a 422 status code when updating with invalid data."""
+    existing_blog = Blog(
+        id=1,
+        title="Existing Blog Post",
+        excerpt="A summary of the blog post...",
+        content="The content of the blog post...",
+        image_url="image-url-link",
+        created_at=datetime.now(timezone.utc),
+        updated_at=datetime.now(timezone.utc),
+    )
     invalid_data = {
+        "title": "Short",
         "excerpt": "An updated summary...",
-        "content": "Updated content...",
-        "image_url": "updated-image-url-link",
     }
 
-    response = client.patch("/api/v1/blogs/1", json=invalid_data)
+    db_session_mock.query.return_value.filter.return_value.first.side_effect = [
+        existing_blog,  # Return the existing blog for the first query
+        None,  # Return None for the second query to simulate no conflict
+    ]
+    db_session_mock.commit.side_effect = lambda: None
+    db_session_mock.refresh.side_effect = lambda blog: setattr(
+        blog, "updated_at", datetime.now(timezone.utc)
+    )
+
+    response = client.patch(f"/api/v1/blogs/{existing_blog.id}", json=invalid_data)
 
     assert response.status_code == 422