feat: Blog Update API Endpoint [Issue #5]

Author: Pythonian

.coveragerc

@@ -0,0 +1,6 @@
+[run]
+branch = True
+omit = */__init__.py
+
+[report]
+show_missing = True

Makefile

@@ -20,9 +20,6 @@ TOX = $(VENV_DIR)/bin/tox
 # Define the pre-commit executable within the virtual environment
 PRE_COMMIT = $(VENV_DIR)/bin/pre-commit
 
-# Define the uvicorn executable within the virtual environment
-UVICORN = $(VENV_DIR)/bin/uvicorn
-
 # Define the alembic executable within the virtual environment
 ALEMBIC = $(VENV_DIR)/bin/alembic
 
@@ -53,7 +50,7 @@ check: ## Run all checks using tox and pre-commit
 	@echo "All checks passed"
 
 run: ## Run the development server
-	$(UVICORN) main:app --reload
+	@fastapi dev
 
 clean: ## Clean up the project
 	@echo "Cleaning up the project of temporary files and directories..."

api/v1/routes/blog.py

@@ -12,6 +12,7 @@
     BlogListItemResponseSchema,
     BlogListResponseSchema,
     BlogResponseSchema,
+    BlogUpdateSchema,
 )
 
 blog = APIRouter(prefix="/blogs", tags=["Blog"])
@@ -48,7 +49,15 @@ async def create_blog(
         db.refresh(new_blog)
         logger.info(f"Blog post '{new_blog.title}' created successfully.")
 
-        return BlogResponseSchema.model_validate(new_blog.__dict__)
+        return BlogResponseSchema(
+            id=new_blog.id,
+            title=new_blog.title,
+            excerpt=new_blog.excerpt,
+            content=new_blog.content,
+            image_url=new_blog.image_url,
+            created_at=new_blog.created_at,
+            updated_at=new_blog.updated_at,
+        )
 
     except HTTPException as http_err:
         logger.warning(f"HTTP error occurred: {http_err.detail}")
@@ -171,6 +180,96 @@ async def read_blog(
         )
 
 
+@blog.patch(
+    "/{id}",
+    response_model=BlogResponseSchema,
+    status_code=status.HTTP_200_OK,
+)
+def update_blog(
+    id: int,
+    blog_update: BlogUpdateSchema,
+    db: Session = Depends(get_db),
+) -> BlogResponseSchema:
+    try:
+        # Fetch the blog post to be updated
+        blog = (
+            db.query(Blog)
+            .filter(
+                Blog.id == id,
+                not_(Blog.is_deleted),
+            )
+            .first()
+        )
+        if not blog:
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail="Blog post not found.",
+            )
+
+        # Get the updated data
+        update_data = blog_update.model_dump(exclude_unset=True)
+
+        # Check for title uniqueness
+        if "title" in update_data and update_data["title"] != blog.title:
+            existing_blog = (
+                db.query(Blog)
+                .filter(
+                    Blog.title == update_data["title"],
+                    not_(Blog.is_deleted),
+                )
+                .first()
+            )
+            if existing_blog:
+                logger.warning(
+                    f"Blog post with title '{update_data['title']}' already exists."
+                )
+                raise HTTPException(
+                    status_code=status.HTTP_409_CONFLICT,
+                    detail="A blog post with this title already exists.",
+                )
+
+        # Update fields if they are provided
+        for field, value in update_data.items():
+            setattr(blog, field, value)
+
+        db.commit()
+        db.refresh(blog)
+        logger.info(f"Blog post '{blog.title}' updated successfully.")
+
+        return BlogResponseSchema(
+            id=blog.id,
+            title=blog.title,
+            excerpt=blog.excerpt,
+            content=blog.content,
+            image_url=blog.image_url,
+            created_at=blog.created_at,
+            updated_at=blog.updated_at,
+        )
+
+    except HTTPException as http_err:
+        # Log the HTTP exception
+        logger.warning(f"HTTP error occurred: {http_err.detail}")
+        raise http_err
+
+    except SQLAlchemyError as sql_err:
+        # Log SQLAlchemy errors
+        logger.error(f"Database error occurred: {sql_err}")
+        db.rollback()
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Database error occurred.",
+        )
+
+    except Exception as e:
+        # Log other unexpected errors
+        logger.error(f"Unexpected error occurred: {e}")
+        db.rollback()
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Internal server error.",
+        )
+
+
 @blog.delete(
     "/{id}",
     status_code=status.HTTP_204_NO_CONTENT,

api/v1/schemas/blog.py

@@ -70,3 +70,12 @@ class BlogListResponseSchema(BaseModel):
     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}

requirements.txt

@@ -4,6 +4,7 @@ fastapi==0.111.1
 psycopg2-binary==2.9.9
 pydantic-settings==2.3.4
 pytest==8.3.1
+pytest-cov==5.0.0
 pytest-mock==3.14.0
 python-decouple==3.8
 SQLAlchemy==2.0.31

tests/v1/blog/test_update_blog.py

@@ -0,0 +1,186 @@
+from datetime import datetime, timezone
+from unittest.mock import MagicMock
+
+import pytest
+from fastapi.testclient import TestClient
+
+from api.db.database import get_db
+from api.v1.models.blog import Blog
+from main import app
+
+client = TestClient(app)
+
+
+@pytest.fixture
+def db_session_mock():
+    return MagicMock()
+
+
+@pytest.fixture(autouse=True)
+def override_get_db(db_session_mock):
+    app.dependency_overrides[get_db] = lambda: db_session_mock
+    yield
+    app.dependency_overrides[get_db] = None
+
+
+def test_update_blog_success(db_session_mock):
+    """Ensure that the endpoint successfully updates the data of an existing blog post."""
+    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),
+    )
+    updated_data = {
+        "title": "Updated Blog Post",
+        "excerpt": "An updated summary...",
+        "content": "Updated content...",
+        "image_url": "updated-image-url-link",
+    }
+
+    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=updated_data)
+
+    assert response.status_code == 200
+    response_data = response.json()
+    assert response_data["title"] == updated_data["title"]
+    assert response_data["excerpt"] == updated_data["excerpt"]
+    assert response_data["content"] == updated_data["content"]
+    assert response_data["image_url"] == updated_data["image_url"]
+    assert "updated_at" in response_data
+
+
+def test_update_blog_conflict(db_session_mock):
+    """Simulate a request to update a blog post with a title that already exists."""
+    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),
+    )
+    conflicting_blog = Blog(
+        id=2,
+        title="Updated Blog Post",
+        excerpt="Another summary...",
+        content="Another content...",
+        image_url="another-image-url-link",
+        created_at=datetime.now(timezone.utc),
+        updated_at=datetime.now(timezone.utc),
+    )
+    updated_data = {
+        "title": "Updated Blog Post",
+        "excerpt": "An updated summary...",
+        "content": "Updated content...",
+        "image_url": "updated-image-url-link",
+    }
+
+    db_session_mock.query.return_value.filter.return_value.first.side_effect = [
+        existing_blog,  # Return the existing blog for the first query
+        conflicting_blog,  # Return the conflicting blog for the second query
+    ]
+
+    response = client.patch(f"/api/v1/blogs/{existing_blog.id}", json=updated_data)
+
+    assert response.status_code == 409
+    assert response.json()["detail"] == "A blog post with this title already exists."
+
+
+def test_update_blog_not_found(db_session_mock):
+    """Simulate a request to update a blog post that does not exist."""
+    updated_data = {
+        "title": "Updated Blog Post",
+        "excerpt": "An updated summary...",
+        "content": "Updated content...",
+        "image_url": "updated-image-url-link",
+    }
+
+    db_session_mock.query.return_value.filter.return_value.first.return_value = None
+
+    response = client.patch("/api/v1/blogs/999", json=updated_data)
+
+    assert response.status_code == 404
+    assert response.json()["detail"] == "Blog post not found."
+
+
+def test_update_blog_internal_server_error(db_session_mock):
+    """Simulate an internal server error to raise an exception."""
+    updated_data = {
+        "title": "Updated Blog Post",
+        "excerpt": "An updated summary...",
+        "content": "Updated content...",
+        "image_url": "updated-image-url-link",
+    }
+
+    db_session_mock.query.side_effect = Exception("Unexpected error")
+
+    response = client.patch("/api/v1/blogs/1", json=updated_data)
+
+    assert response.status_code == 500
+    assert response.json()["detail"] == "Internal server error."
+
+
+def test_update_blog_invalid_data():
+    """Send requests with invalid data and check for validation errors."""
+    invalid_data = {
+        "excerpt": "An updated summary...",
+        "content": "Updated content...",
+        "image_url": "updated-image-url-link",
+    }
+
+    response = client.patch("/api/v1/blogs/1", json=invalid_data)
+
+    assert response.status_code == 422
+
+
+def test_update_blog_boundary_testing(db_session_mock):
+    """Test the maximum length constraints for the title and excerpt fields."""
+    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),
+    )
+    boundary_blog_data = {
+        "title": "T" * 255,  # Maximum allowed length for title
+        "excerpt": "E" * 300,  # Maximum allowed length for excerpt
+        "content": "Content of the blog post...",
+        "image_url": "image-url-link",
+    }
+
+    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=boundary_blog_data
+    )
+
+    assert response.status_code == 200
+    response_data = response.json()
+    assert response_data["title"] == boundary_blog_data["title"]
+    assert response_data["excerpt"] == boundary_blog_data["excerpt"]
+    assert response_data["content"] == boundary_blog_data["content"]
+    assert response_data["image_url"] == boundary_blog_data["image_url"]
+    assert "updated_at" in response_data

tox.ini

@@ -14,7 +14,7 @@ basepython = python3.10
 deps =
     -r requirements.txt
 commands =
-    pytest
+    pytest --cov-config=.coveragerc --cov=api --cov-report=html:htmlcov --cov-report=term-missing tests/
 
 [testenv:type]
 description = run type checks