ows_ep_dev / main.py
bagustyo's picture
Update main.py
2df1afe verified
from fastapi import FastAPI, HTTPException, Request, Form, Query
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.templating import Jinja2Templates
from pydantic import BaseModel
from typing import List, Optional
import json
import os
app = FastAPI()
templates = Jinja2Templates(directory="templates")
# Path to the JSON file that serves as a simple database
DB_FILE = "books.json"
# Define the Book data model using Pydantic
class Book(BaseModel):
id: int
book_name: str
year: int
the_author: str
def read_books_file() -> List[dict]:
"""Read books from the JSON file."""
if os.path.exists(DB_FILE):
with open(DB_FILE, "r") as f:
try:
data = json.load(f)
return data
except json.JSONDecodeError:
# Return an empty list if the file is empty or corrupted
return []
return []
def write_books_file(books: List[dict]):
"""Write books to the JSON file."""
with open(DB_FILE, "w") as f:
json.dump(books, f, indent=4)
def get_next_id(books: List[dict]) -> int:
"""Generate the next ID for a new book."""
if books:
return max(book["id"] for book in books) + 1
return 1
# ---------------------------
# JSON API Endpoints
# ---------------------------
@app.get("/api/books", response_model=List[Book])
def get_books_api():
"""
Retrieve all books as JSON.
"""
books = read_books_file()
return books
@app.post("/api/books", response_model=Book)
def create_book_api(book: Book):
"""
Create a new book record via JSON API.
"""
books = read_books_file()
book.id = get_next_id(books)
books.append(book.dict())
write_books_file(books)
return book
@app.get("/api/books/search", response_model=List[Book])
def search_books_api(
book_name: Optional[str] = Query(None),
year: Optional[int] = Query(None),
the_author: Optional[str] = Query(None)
):
"""
Search for books by book name, year, or author.
"""
books = read_books_file()
filtered_books = [
book for book in books
if (book_name and book_name.lower() in book["book_name"].lower()) or
(year and book["year"] == year) or
(the_author and the_author.lower() in book["the_author"].lower())
]
return filtered_books
# ---------------------------
# HTML Endpoints using Jinja2 & Bootstrap
# ---------------------------
@app.get("/", response_class=HTMLResponse)
def read_books_page(request: Request):
"""
Render the homepage with a list of books.
"""
books = read_books_file()
return templates.TemplateResponse("index.html", {"request": request, "books": books})
@app.get("/add", response_class=HTMLResponse)
def add_book_page(request: Request):
"""
Render a page with a form to add a new book.
"""
return templates.TemplateResponse("add_book.html", {"request": request, "error": None})
@app.post("/add")
def add_book(
request: Request,
book_name: str = Form(...),
year: int = Form(...),
the_author: str = Form(...)
):
"""
Process the form data and add a new book.
"""
books = read_books_file()
new_book = {
"id": get_next_id(books),
"book_name": book_name,
"year": year,
"the_author": the_author
}
books.append(new_book)
write_books_file(books)
# Redirect back to the homepage after a successful submission
return RedirectResponse("/", status_code=303)