You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
166 lines
4.3 KiB
166 lines
4.3 KiB
from fastapi import FastAPI, HTTPException
|
|
from pydantic import BaseModel
|
|
from typing import List
|
|
from datetime import date
|
|
from uuid import uuid4
|
|
|
|
app = FastAPI()
|
|
|
|
|
|
|
|
class Book(BaseModel):
|
|
title: str
|
|
author: str
|
|
isbn: str
|
|
copies: int
|
|
|
|
class BookResponse(Book):
|
|
available_copies: int
|
|
|
|
class Customer(BaseModel):
|
|
name: str
|
|
email: str
|
|
customer_id: str
|
|
|
|
class CheckoutRequest(BaseModel):
|
|
isbn: str
|
|
customer_id: str
|
|
due_date: str
|
|
|
|
class CheckoutResponse(BaseModel):
|
|
checkout_id: str
|
|
isbn: str
|
|
title: str
|
|
customer_id: str
|
|
checkout_date: str
|
|
due_date: str
|
|
|
|
class ReturnRequest(BaseModel):
|
|
isbn: str
|
|
customer_id: str
|
|
|
|
|
|
|
|
books = {}
|
|
customers = {}
|
|
checkouts = []
|
|
|
|
|
|
|
|
def get_book(isbn: str):
|
|
if isbn not in books:
|
|
raise HTTPException(status_code=404, detail="Book not found")
|
|
return books[isbn]
|
|
|
|
def get_customer(customer_id: str):
|
|
if customer_id not in customers:
|
|
raise HTTPException(status_code=404, detail="Customer not found")
|
|
return customers[customer_id]
|
|
|
|
def enforce_business_rules(customer_id: str, isbn: str):
|
|
customer_books = [c for c in checkouts if c["customer_id"] == customer_id]
|
|
if len(customer_books) >= 5:
|
|
raise HTTPException(status_code=400, detail="Customer has too many books checked out")
|
|
if books[isbn]["available_copies"] <= 0:
|
|
raise HTTPException(status_code=400, detail="No copies available")
|
|
|
|
|
|
|
|
@app.post("/api/books", response_model=BookResponse, status_code=201)
|
|
def add_book(book: Book):
|
|
if book.isbn in books:
|
|
raise HTTPException(status_code=400, detail="Book already exists")
|
|
books[book.isbn] = book.dict()
|
|
books[book.isbn]["available_copies"] = book.copies
|
|
return books[book.isbn]
|
|
|
|
@app.get("/api/books/{isbn}", response_model=BookResponse)
|
|
def get_book_details(isbn: str):
|
|
return get_book(isbn)
|
|
|
|
|
|
|
|
@app.post("/api/customers", response_model=Customer, status_code=201)
|
|
def create_customer(customer: Customer):
|
|
if customer.customer_id in customers:
|
|
raise HTTPException(status_code=400, detail="Customer already exists")
|
|
customers[customer.customer_id] = customer.dict()
|
|
return customer
|
|
|
|
|
|
@app.get("/api/customers/{customer_id}", response_model=Customer)
|
|
def get_customer_details(customer_id: str):
|
|
return get_customer(customer_id)
|
|
|
|
|
|
|
|
@app.post("/api/checkouts", response_model=CheckoutResponse, status_code=201)
|
|
def checkout_book(request: CheckoutRequest):
|
|
book = get_book(request.isbn)
|
|
get_customer(request.customer_id)
|
|
enforce_business_rules(request.customer_id, request.isbn)
|
|
|
|
checkout_id = "CKO" + uuid4().hex[:6].upper()
|
|
checkout_date = date.today().isoformat()
|
|
|
|
books[request.isbn]["available_copies"] -= 1
|
|
checkout = {
|
|
"checkout_id": checkout_id,
|
|
"isbn": request.isbn,
|
|
"title": book["title"],
|
|
"customer_id": request.customer_id,
|
|
"checkout_date": checkout_date,
|
|
"due_date": request.due_date
|
|
}
|
|
checkouts.append(checkout)
|
|
return checkout
|
|
|
|
|
|
|
|
|
|
|
|
@app.post("/api/returns")
|
|
def return_book(request: ReturnRequest):
|
|
for i, c in enumerate(checkouts):
|
|
if c["isbn"] == request.isbn and c["customer_id"] == request.customer_id:
|
|
checkouts.pop(i)
|
|
books[request.isbn]["available_copies"] += 1
|
|
return {
|
|
"message": "Book returned successfully",
|
|
"isbn": request.isbn,
|
|
"customer_id": request.customer_id,
|
|
"return_date": date.today().isoformat()
|
|
}
|
|
raise HTTPException(status_code=404, detail="Borrow record not found")
|
|
|
|
|
|
|
|
|
|
@app.get("/api/customers/{customer_id}/books")
|
|
def get_customer_books(customer_id: str):
|
|
get_customer(customer_id)
|
|
return [
|
|
{
|
|
"isbn": c["isbn"],
|
|
"title": c["title"],
|
|
"author": books[c["isbn"]]["author"],
|
|
"checkout_date": c["checkout_date"],
|
|
"due_date": c["due_date"]
|
|
}
|
|
for c in checkouts if c["customer_id"] == customer_id
|
|
]
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@app.post("/api/reset")
|
|
def reset_system():
|
|
books.clear()
|
|
customers.clear()
|
|
checkouts.clear()
|
|
return {"message": "System reset successful"}
|