a9210ca8ed
- Implemented CRUD operations for manga, authors, publishers, and genres - Added search and filtering functionality - Set up SQLAlchemy ORM with SQLite database - Configured Alembic for database migrations - Implemented logging with Loguru - Added comprehensive API documentation - Set up error handling and validation - Added ruff for linting and formatting
145 lines
4.3 KiB
Python
145 lines
4.3 KiB
Python
from typing import Any
|
|
|
|
from fastapi.encoders import jsonable_encoder
|
|
from sqlalchemy.orm import Session, joinedload
|
|
|
|
from app.crud.base import CRUDBase
|
|
from app.models.manga import Manga, MangaGenre
|
|
from app.schemas.manga import MangaCreate, MangaUpdate
|
|
|
|
|
|
class CRUDManga(CRUDBase[Manga, MangaCreate, MangaUpdate]):
|
|
def get(self, db: Session, id: Any) -> Manga | None:
|
|
"""
|
|
Get a manga by ID with all relationships loaded.
|
|
"""
|
|
return (
|
|
db.query(Manga)
|
|
.filter(Manga.id == id)
|
|
.options(
|
|
joinedload(Manga.author),
|
|
joinedload(Manga.publisher),
|
|
joinedload(Manga.genres).joinedload(MangaGenre.genre),
|
|
)
|
|
.first()
|
|
)
|
|
|
|
def get_by_isbn(self, db: Session, *, isbn: str) -> Manga | None:
|
|
"""
|
|
Get a manga by ISBN.
|
|
"""
|
|
return db.query(Manga).filter(Manga.isbn == isbn).first()
|
|
|
|
def get_multi(self, db: Session, *, skip: int = 0, limit: int = 100) -> list[Manga]:
|
|
"""
|
|
Get multiple manga with all relationships loaded.
|
|
"""
|
|
return (
|
|
db.query(Manga)
|
|
.options(
|
|
joinedload(Manga.author),
|
|
joinedload(Manga.publisher),
|
|
joinedload(Manga.genres).joinedload(MangaGenre.genre),
|
|
)
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
def create(self, db: Session, *, obj_in: MangaCreate) -> Manga:
|
|
"""
|
|
Create a new manga with genre relationships.
|
|
"""
|
|
obj_in_data = jsonable_encoder(obj_in, exclude={"genre_ids"})
|
|
db_obj = Manga(**obj_in_data)
|
|
db.add(db_obj)
|
|
db.flush() # Flush to get the ID
|
|
|
|
# Add genres if provided
|
|
if obj_in.genre_ids:
|
|
for genre_id in obj_in.genre_ids:
|
|
manga_genre = MangaGenre(manga_id=db_obj.id, genre_id=genre_id)
|
|
db.add(manga_genre)
|
|
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
return db_obj
|
|
|
|
def update(self, db: Session, *, db_obj: Manga, obj_in: MangaUpdate | dict[str, Any]) -> Manga:
|
|
"""
|
|
Update a manga with genre relationships.
|
|
"""
|
|
if isinstance(obj_in, dict):
|
|
update_data = obj_in
|
|
genre_ids = update_data.pop("genre_ids", None)
|
|
else:
|
|
update_data = obj_in.dict(exclude_unset=True)
|
|
genre_ids = update_data.pop("genre_ids", None) if "genre_ids" in update_data else None
|
|
|
|
# Update the manga object
|
|
obj_data = jsonable_encoder(db_obj)
|
|
for field in obj_data:
|
|
if field in update_data:
|
|
setattr(db_obj, field, update_data[field])
|
|
|
|
# Update genres if provided
|
|
if genre_ids is not None:
|
|
# Remove existing genres
|
|
db.query(MangaGenre).filter(MangaGenre.manga_id == db_obj.id).delete()
|
|
|
|
# Add new genres
|
|
for genre_id in genre_ids:
|
|
manga_genre = MangaGenre(manga_id=db_obj.id, genre_id=genre_id)
|
|
db.add(manga_genre)
|
|
|
|
db.add(db_obj)
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
return db_obj
|
|
|
|
def search(
|
|
self,
|
|
db: Session,
|
|
*,
|
|
title: str | None = None,
|
|
author_id: int | None = None,
|
|
publisher_id: int | None = None,
|
|
genre_id: int | None = None,
|
|
in_stock: bool | None = None,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
) -> list[Manga]:
|
|
"""
|
|
Search manga by various criteria.
|
|
"""
|
|
query = db.query(Manga)
|
|
|
|
if title:
|
|
query = query.filter(Manga.title.ilike(f"%{title}%"))
|
|
|
|
if author_id:
|
|
query = query.filter(Manga.author_id == author_id)
|
|
|
|
if publisher_id:
|
|
query = query.filter(Manga.publisher_id == publisher_id)
|
|
|
|
if genre_id:
|
|
query = query.join(Manga.genres).filter(MangaGenre.genre_id == genre_id)
|
|
|
|
if in_stock is not None:
|
|
query = query.filter(Manga.in_stock == in_stock)
|
|
|
|
return (
|
|
query.options(
|
|
joinedload(Manga.author),
|
|
joinedload(Manga.publisher),
|
|
joinedload(Manga.genres).joinedload(MangaGenre.genre),
|
|
)
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
|
|
manga = CRUDManga(Manga)
|