2025-06-02 12:04:09 +00:00

159 lines
4.1 KiB
Python

from typing import Any, List
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from app import crud, schemas
from app.db.session import get_db
router = APIRouter()
@router.get("/categories/", response_model=List[schemas.Category])
def read_categories(
db: Session = Depends(get_db),
skip: int = 0,
limit: int = 100,
) -> Any:
"""
Retrieve all categories.
"""
categories = crud.category.get_multi(db, skip=skip, limit=limit)
return categories
@router.get("/categories/{category_id}", response_model=schemas.Category)
def read_category(
*,
db: Session = Depends(get_db),
category_id: int,
) -> Any:
"""
Get category by ID.
"""
category = crud.category.get(db, id=category_id)
if not category:
raise HTTPException(status_code=404, detail="Category not found")
return category
@router.get("/items/", response_model=List[schemas.RentalItem])
def read_rental_items(
db: Session = Depends(get_db),
skip: int = 0,
limit: int = 100,
category_id: int = None,
available_only: bool = False,
) -> Any:
"""
Retrieve rental items with optional filtering.
"""
if category_id:
items = crud.rental_item.get_by_category(
db, category_id=category_id, skip=skip, limit=limit
)
elif available_only:
items = crud.rental_item.get_available(db, skip=skip, limit=limit)
else:
items = crud.rental_item.get_multi(db, skip=skip, limit=limit)
return items
@router.get("/items/{item_id}", response_model=schemas.RentalItem)
def read_rental_item(
*,
db: Session = Depends(get_db),
item_id: int,
) -> Any:
"""
Get rental item by ID.
"""
item = crud.rental_item.get(db, id=item_id)
if not item:
raise HTTPException(status_code=404, detail="Rental item not found")
return item
@router.get("/customers/", response_model=List[schemas.Customer])
def read_customers(
db: Session = Depends(get_db),
skip: int = 0,
limit: int = 100,
) -> Any:
"""
Retrieve all customers.
"""
customers = crud.customer.get_multi(db, skip=skip, limit=limit)
return customers
@router.get("/customers/{customer_id}", response_model=schemas.Customer)
def read_customer(
*,
db: Session = Depends(get_db),
customer_id: int,
) -> Any:
"""
Get customer by ID.
"""
customer = crud.customer.get(db, id=customer_id)
if not customer:
raise HTTPException(status_code=404, detail="Customer not found")
return customer
@router.get("/customers/email/{email}", response_model=schemas.Customer)
def read_customer_by_email(
*,
db: Session = Depends(get_db),
email: str,
) -> Any:
"""
Get customer by email.
"""
customer = crud.customer.get_by_email(db, email=email)
if not customer:
raise HTTPException(status_code=404, detail="Customer not found")
return customer
@router.get("/records/", response_model=List[schemas.RentalRecord])
def read_rental_records(
db: Session = Depends(get_db),
skip: int = 0,
limit: int = 100,
customer_id: int = None,
item_id: int = None,
active_only: bool = False,
) -> Any:
"""
Retrieve rental records with optional filtering.
"""
if customer_id:
records = crud.rental_record.get_by_customer(
db, customer_id=customer_id, skip=skip, limit=limit
)
elif item_id:
records = crud.rental_record.get_by_item(
db, item_id=item_id, skip=skip, limit=limit
)
elif active_only:
records = crud.rental_record.get_active(db, skip=skip, limit=limit)
else:
records = crud.rental_record.get_multi(db, skip=skip, limit=limit)
return records
@router.get("/records/{record_id}", response_model=schemas.RentalRecord)
def read_rental_record(
*,
db: Session = Depends(get_db),
record_id: int,
) -> Any:
"""
Get rental record by ID.
"""
record = crud.rental_record.get(db, id=record_id)
if not record:
raise HTTPException(status_code=404, detail="Rental record not found")
return record