4d6c2e1778
- Create project structure and dependencies - Set up SQLAlchemy models for anime, genres, and their relationships - Implement CRUD operations for all models - Set up FastAPI endpoints for managing anime and genres - Add health check endpoint - Configure Alembic for database migrations - Add data seeding capability - Update README with project information
117 lines
3.4 KiB
Python
117 lines
3.4 KiB
Python
from typing import List, Optional, Dict, Any, Union
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.crud.base import CRUDBase
|
|
from app.models.anime import Anime
|
|
from app.models.anime_genre import AnimeGenre
|
|
from app.schemas.anime import AnimeCreate, AnimeUpdate
|
|
|
|
|
|
class CRUDAnime(CRUDBase[Anime, AnimeCreate, AnimeUpdate]):
|
|
def create_with_genres(
|
|
self, db: Session, *, obj_in: AnimeCreate
|
|
) -> Anime:
|
|
genre_ids = obj_in.genre_ids
|
|
anime_data = obj_in.dict(exclude={"genre_ids"})
|
|
|
|
db_obj = Anime(**anime_data)
|
|
db.add(db_obj)
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
|
|
# Add genres
|
|
if genre_ids:
|
|
for genre_id in genre_ids:
|
|
anime_genre = AnimeGenre(anime_id=db_obj.id, genre_id=genre_id)
|
|
db.add(anime_genre)
|
|
db.commit()
|
|
|
|
return db_obj
|
|
|
|
def update_with_genres(
|
|
self,
|
|
db: Session,
|
|
*,
|
|
db_obj: Anime,
|
|
obj_in: Union[AnimeUpdate, Dict[str, Any]]
|
|
) -> Anime:
|
|
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 anime attributes
|
|
for field in update_data:
|
|
setattr(db_obj, field, update_data[field])
|
|
|
|
# Update genres if provided
|
|
if genre_ids is not None:
|
|
# Remove existing genre links
|
|
db.query(AnimeGenre).filter(AnimeGenre.anime_id == db_obj.id).delete()
|
|
|
|
# Add new genre links
|
|
for genre_id in genre_ids:
|
|
anime_genre = AnimeGenre(anime_id=db_obj.id, genre_id=genre_id)
|
|
db.add(anime_genre)
|
|
|
|
db.add(db_obj)
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
return db_obj
|
|
|
|
def get_with_genres(self, db: Session, id: int):
|
|
"""Get anime with its genres populated"""
|
|
anime = db.query(Anime).filter(Anime.id == id).first()
|
|
if anime:
|
|
# Load genres (to be implemented with relationships)
|
|
pass
|
|
return anime
|
|
|
|
def search(
|
|
self,
|
|
db: Session,
|
|
*,
|
|
title: Optional[str] = None,
|
|
genre_id: Optional[int] = None,
|
|
status: Optional[str] = None,
|
|
skip: int = 0,
|
|
limit: int = 100
|
|
) -> List[Anime]:
|
|
query = db.query(Anime)
|
|
|
|
if title:
|
|
query = query.filter(Anime.title.ilike(f"%{title}%"))
|
|
|
|
if genre_id:
|
|
query = query.join(AnimeGenre).filter(AnimeGenre.genre_id == genre_id)
|
|
|
|
if status:
|
|
query = query.filter(Anime.status == status)
|
|
|
|
return query.offset(skip).limit(limit).all()
|
|
|
|
def search_count(
|
|
self,
|
|
db: Session,
|
|
*,
|
|
title: Optional[str] = None,
|
|
genre_id: Optional[int] = None,
|
|
status: Optional[str] = None
|
|
) -> int:
|
|
query = db.query(Anime)
|
|
|
|
if title:
|
|
query = query.filter(Anime.title.ilike(f"%{title}%"))
|
|
|
|
if genre_id:
|
|
query = query.join(AnimeGenre).filter(AnimeGenre.genre_id == genre_id)
|
|
|
|
if status:
|
|
query = query.filter(Anime.status == status)
|
|
|
|
return query.count()
|
|
|
|
|
|
anime = CRUDAnime(Anime) |