feat: Blog List API Endpoint [Issue #3]

Author: Pythonian

api/v1/routes/blog.py

@@ -1,12 +1,14 @@
-from fastapi import APIRouter, Depends, HTTPException, status
+from typing import Any
+from fastapi import APIRouter, Depends, HTTPException, Query, status
 from sqlalchemy.orm import Session
 from sqlalchemy.exc import SQLAlchemyError
+from sqlalchemy import not_
 from api.v1.models.blog import Blog
 from api.v1.schemas.blog import BlogCreateSchema, BlogResponseSchema
 from api.db.database import get_db
 import logging
 
-blog = APIRouter(prefix="/blogs", tags=["blog"])
+blog = APIRouter(prefix="/blogs", tags=["Blog"])
 
 logger = logging.getLogger("api")
 
@@ -16,7 +18,7 @@
     response_model=BlogResponseSchema,
     status_code=status.HTTP_201_CREATED,
 )
-def create_blog(blog: BlogCreateSchema, db: Session = Depends(get_db)):
+async def create_blog(blog: BlogCreateSchema, db: Session = Depends(get_db)):
     try:
         existing_blog = db.query(Blog).filter(Blog.title == blog.title).first()
         if existing_blog:
@@ -55,3 +57,59 @@ def create_blog(blog: BlogCreateSchema, db: Session = Depends(get_db)):
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
             detail="Internal server error.",
         )
+
+
+@blog.get("", 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),
+) -> dict[str, Any]:
+    try:
+        offset = (page - 1) * page_size
+        query = (
+            db.query(Blog)
+            .filter(not_(Blog.is_deleted))
+            .order_by(Blog.created_at.desc())
+        )
+        total_count = query.count()
+        blogs = query.offset(offset).limit(page_size).all()
+
+        next_page = None
+        if offset + page_size < total_count:
+            next_page = f"/api/v1/blogs?page={page + 1}&page_size={page_size}"
+
+        prev_page = None
+        if page > 1:
+            prev_page = f"/api/v1/blogs?page={page - 1}&page_size={page_size}"
+
+        results = [
+            {
+                "id": blog.id,
+                "title": blog.title,
+                "excerpt": blog.excerpt,
+                "image_url": blog.image_url,
+                "created_at": blog.created_at,
+            }
+            for blog in blogs
+        ]
+
+        return {
+            "count": total_count,
+            "next": next_page,
+            "previous": prev_page,
+            "results": results,
+        }
+
+    except SQLAlchemyError as e:
+        logger.error(f"Database error occurred: {e}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Database error occurred.",
+        )
+    except Exception as e:
+        logger.error(f"Unexpected error occurred: {e}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Internal server error.",
+        )

tests/v1/blog/test_list_blog.py

@@ -0,0 +1,120 @@
+import pytest
+from fastapi.testclient import TestClient
+from unittest.mock import MagicMock
+from main import app
+from api.v1.models.blog import Blog
+from api.db.database import get_db
+
+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_successful_retrieval_of_paginated_blog_posts(db_session_mock):
+    blog1 = Blog(
+        id=1,
+        title="My First Blog",
+        excerpt="This is an excerpt from my first blog.",
+        content="Content of the first blog",
+        image_url="https://example.com/image1.jpg",
+    )
+    _ = Blog(
+        id=2,
+        title="My Second Blog",
+        excerpt="This is an excerpt from my second blog.",
+        content="Content of the second blog",
+        image_url="https://example.com/image2.jpg",
+    )
+
+    db_session_mock.query().filter().order_by().offset().limit().all.return_value = [
+        blog1
+    ]
+    db_session_mock.query().filter().order_by().count.return_value = 2
+
+    response = client.get("/api/v1/blogs?page=1&page_size=1")
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["count"] == 2
+    assert data["next"] == "/api/v1/blogs?page=2&page_size=1"
+    assert data["previous"] is None
+    assert len(data["results"]) == 1
+    assert data["results"][0]["title"] == "My First Blog"
+
+
+def test_no_blog_posts_present(db_session_mock):
+    db_session_mock.query().filter().order_by().offset().limit().all.return_value = []
+    db_session_mock.query().filter().order_by().count.return_value = 0
+
+    response = client.get("/api/v1/blogs?page=1&page_size=10")
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["count"] == 0
+    assert data["next"] is None
+    assert data["previous"] is None
+    assert len(data["results"]) == 0
+
+
+def test_internal_server_error(mocker):
+    mocker.patch("api.v1.routes.blog", side_effect=Exception("Test exception"))
+
+    response = client.get("/api/v1/blogs?page=1&page_size=10")
+
+    assert response.status_code == 500
+    data = response.json()
+    assert data["detail"] == "Internal server error."
+
+
+def test_invalid_page_or_page_size_parameters():
+    # Test invalid page parameter
+    response = client.get("/api/v1/blogs?page=-1&page_size=10")
+    assert response.status_code == 422
+
+    # Test invalid page_size parameter
+    response = client.get("/api/v1/blogs?page=1&page_size=-1")
+    assert response.status_code == 422
+
+    # Test non-integer page parameter
+    response = client.get("/api/v1/blogs?page=abc&page_size=10")
+    assert response.status_code == 422
+
+
+def test_invalid_method():
+    response = client.delete("/api/v1/blogs")
+
+    assert response.status_code == 405
+    data = response.json()
+    assert data["detail"] == "Method Not Allowed"
+
+
+def test_soft_deleted_blog_post_access_control(db_session_mock):
+    _ = Blog(
+        id=1,
+        title="Soft Deleted Blog",
+        excerpt="This is an excerpt from a soft deleted blog.",
+        content="Content of the soft deleted blog",
+        image_url="https://example.com/image1.jpg",
+        is_deleted=True,
+    )
+    db_session_mock.query().filter().order_by().offset().limit().all.return_value = []
+    db_session_mock.query().filter().order_by().count.return_value = 0
+
+    response = client.get("/api/v1/blogs?page=1&page_size=10")
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["count"] == 0
+    assert data["next"] is None
+    assert data["previous"] is None
+    assert len(data["results"]) == 0