implement CRUD API for books

Author: sis0k0

main.py

@@ -1,11 +1,20 @@
 from fastapi import FastAPI
 from dotenv import dotenv_values
+from pymongo import MongoClient
+from routes import router as book_router
+
 config = dotenv_values(".env")
 
 app = FastAPI()
 
-@app.get("/")
-async def root():
-    return {"message": "Hello World"}
+@app.on_event("startup")
+def startup_db_client():
+    app.mongodb_client = MongoClient(config["ATLAS_URI"])
+    app.database = app.mongodb_client[config["DB_NAME"]]
+
+@app.on_event("shutdown")
+def shutdown_db_client():
+    app.mongodb_client.close()
 
+app.include_router(book_router, tags=["books"], prefix="/book")
 

models.py

@@ -0,0 +1,34 @@
+import uuid
+from typing import Optional
+from pydantic import BaseModel, Field
+
+class Book(BaseModel):
+    id: str = Field(default_factory=uuid.uuid4, alias="_id")
+    title: str = Field(...)
+    author: str = Field(...)
+    synopsis: str = Field(...)
+
+    class Config:
+        allow_population_by_field_name = True
+        schema_extra = {
+            "example": {
+                "title": "Don Quixote",
+                "author": "Miguel de Cervantes",
+                "synopsis": "..."
+            }
+        }
+
+
+class BookUpdate(BaseModel):
+    title: str = Field(...)
+    author: str = Field(...)
+    synopsis: str = Field(...)
+
+    class Config:
+        schema_extra = {
+            "example": {
+                "title": "Don Quixote",
+                "author": "Miguel de Cervantes",
+                "synopsis": "Don Quixote is a Spanish novel by Miguel de Cervantes..."
+            }
+        }

routes.py

@@ -0,0 +1,63 @@
+from fastapi import APIRouter, Body, Request, Response, HTTPException, status
+from fastapi.encoders import jsonable_encoder
+
+from models import Book, BookUpdate
+
+router = APIRouter()
+
+@router.post("/", response_description="Create a new book", status_code=status.HTTP_201_CREATED)
+def create_book(request: Request, book: Book = Body(...)):
+    book = jsonable_encoder(book)
+    new_book = request.app.database["books"].insert_one(book)
+    created_book = request.app.database["books"].find_one(
+        {"_id": new_book.inserted_id}
+    )
+
+    return created_book
+
+
+@router.get("/", response_description="List all books")
+def list_books(request: Request):
+    books = list(request.app.database["books"].find(limit=100))
+    return books
+
+@router.get("/{id}", response_description="Get a single book by id")
+def find_book(id: str, request: Request):
+    if (book := request.app.database["books"].find_one({"_id": id})) is not None:
+        return book
+
+    raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Book with ID {id} not found")
+
+
+@router.put("/{id}", response_description="Update a book")
+def update_book(id: str, request: Request, book: BookUpdate = Body(...)):
+    book = {k: v for k, v in book.dict().items() if v is not None}
+
+    if len(book) >= 1:
+        update_result = request.app.database["books"].update_one(
+            {"_id": id}, {"$set": book}
+        )
+
+        if update_result.modified_count == 1:
+            if (
+                updated_book := request.app.database["books"].find_one({"_id": id})
+            ) is not None:
+                return updated_book
+
+    if (
+        existing_book := request.app.database["books"].find_one({"_id": id})
+    ) is not None:
+        return existing_book
+
+    raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Book with ID {id} not found")
+
+
+@router.delete("/{id}", response_description="Delete a book")
+def delete_book(id: str, request: Request, response: Response):
+    delete_result = request.app.database["books"].delete_one({"_id": id})
+
+    if delete_result.deleted_count == 1:
+        response.status_code = status.HTTP_204_NO_CONTENT
+        return response
+
+    raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Book with ID {id} not found")