Add comprehensive Pydantic models and configure templates/static files
- Create detailed Pydantic models for anime requests and responses - Add AnimeCreateRequest, AnimeUpdateRequest, PaginatedAnimeResponse, etc. - Update route signatures to use proper response models - Convert return values to use Pydantic models instead of raw dicts - Configure Jinja2Templates in FastAPI application - Mount StaticFiles for CSS, JS, images at /static endpoint - Update anime search and list endpoints to use typed responses - Mark completed Pydantic models and template configuration tasks in web_todo.md
This commit is contained in:
@@ -33,6 +33,43 @@ class AnimeResponse(BaseModel):
|
||||
folder: Optional[str] = None
|
||||
episodes: int = 0
|
||||
|
||||
class AnimeCreateRequest(BaseModel):
|
||||
"""Request model for creating anime entries."""
|
||||
name: str = Field(..., min_length=1, max_length=255)
|
||||
folder: str = Field(..., min_length=1)
|
||||
description: Optional[str] = None
|
||||
status: str = Field(default="planned", pattern="^(ongoing|completed|planned|dropped|paused)$")
|
||||
genre: Optional[str] = None
|
||||
year: Optional[int] = Field(None, ge=1900, le=2100)
|
||||
|
||||
class AnimeUpdateRequest(BaseModel):
|
||||
"""Request model for updating anime entries."""
|
||||
name: Optional[str] = Field(None, min_length=1, max_length=255)
|
||||
folder: Optional[str] = None
|
||||
description: Optional[str] = None
|
||||
status: Optional[str] = Field(None, pattern="^(ongoing|completed|planned|dropped|paused)$")
|
||||
genre: Optional[str] = None
|
||||
year: Optional[int] = Field(None, ge=1900, le=2100)
|
||||
|
||||
class PaginatedAnimeResponse(BaseModel):
|
||||
"""Paginated response model for anime lists."""
|
||||
success: bool = True
|
||||
data: List[AnimeResponse]
|
||||
pagination: Dict[str, Any]
|
||||
|
||||
class AnimeSearchResponse(BaseModel):
|
||||
"""Response model for anime search results."""
|
||||
success: bool = True
|
||||
data: List[AnimeResponse]
|
||||
pagination: Dict[str, Any]
|
||||
search: Dict[str, Any]
|
||||
|
||||
class RescanResponse(BaseModel):
|
||||
"""Response model for rescan operations."""
|
||||
success: bool
|
||||
message: str
|
||||
total_series: int
|
||||
|
||||
# Dependency to get SeriesApp instance
|
||||
def get_series_app() -> SeriesApp:
|
||||
"""Get SeriesApp instance for business logic operations."""
|
||||
@@ -47,7 +84,7 @@ def get_series_app() -> SeriesApp:
|
||||
router = APIRouter(prefix='/api/v1/anime', tags=['anime'])
|
||||
|
||||
|
||||
@router.get('', response_model=Dict[str, Any])
|
||||
@router.get('', response_model=PaginatedAnimeResponse)
|
||||
async def list_anime(
|
||||
status: Optional[str] = Query(None, pattern="^(ongoing|completed|planned|dropped|paused)$"),
|
||||
genre: Optional[str] = Query(None),
|
||||
@@ -57,7 +94,7 @@ async def list_anime(
|
||||
per_page: int = Query(50, ge=1, le=1000),
|
||||
current_user: Optional[Dict] = Depends(get_current_user),
|
||||
series_app: SeriesApp = Depends(get_series_app)
|
||||
) -> Dict[str, Any]:
|
||||
) -> PaginatedAnimeResponse:
|
||||
"""
|
||||
Get all anime with optional filtering and pagination.
|
||||
|
||||
@@ -76,35 +113,34 @@ async def list_anime(
|
||||
# Get the series list from SeriesApp
|
||||
anime_list = series_app.series_list
|
||||
|
||||
# Convert to list of dicts for response
|
||||
formatted_anime = []
|
||||
# Convert to list of AnimeResponse objects
|
||||
anime_responses = []
|
||||
for series_item in anime_list:
|
||||
anime_dict = {
|
||||
'id': getattr(series_item, 'id', str(uuid.uuid4())),
|
||||
'title': getattr(series_item, 'name', 'Unknown'),
|
||||
'folder': getattr(series_item, 'folder', ''),
|
||||
'description': getattr(series_item, 'description', ''),
|
||||
'status': 'ongoing', # Default status
|
||||
'episodes': getattr(series_item, 'total_episodes', 0)
|
||||
}
|
||||
anime_response = AnimeResponse(
|
||||
id=getattr(series_item, 'id', str(uuid.uuid4())),
|
||||
title=getattr(series_item, 'name', 'Unknown'),
|
||||
folder=getattr(series_item, 'folder', ''),
|
||||
description=getattr(series_item, 'description', ''),
|
||||
status='ongoing', # Default status
|
||||
episodes=getattr(series_item, 'total_episodes', 0)
|
||||
)
|
||||
|
||||
# Apply search filter if provided
|
||||
if search:
|
||||
if search.lower() not in anime_dict['title'].lower():
|
||||
if search.lower() not in anime_response.title.lower():
|
||||
continue
|
||||
|
||||
formatted_anime.append(anime_dict)
|
||||
anime_responses.append(anime_response)
|
||||
|
||||
# Apply pagination
|
||||
total = len(formatted_anime)
|
||||
total = len(anime_responses)
|
||||
start_idx = (page - 1) * per_page
|
||||
end_idx = start_idx + per_page
|
||||
paginated_anime = formatted_anime[start_idx:end_idx]
|
||||
paginated_anime = anime_responses[start_idx:end_idx]
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"data": paginated_anime,
|
||||
"pagination": {
|
||||
return PaginatedAnimeResponse(
|
||||
data=paginated_anime,
|
||||
pagination={
|
||||
"page": page,
|
||||
"per_page": per_page,
|
||||
"total": total,
|
||||
@@ -112,7 +148,7 @@ async def list_anime(
|
||||
"has_next": end_idx < total,
|
||||
"has_prev": page > 1
|
||||
}
|
||||
}
|
||||
)
|
||||
except Exception as e:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
@@ -370,14 +406,14 @@ def delete_anime(anime_id: int) -> Dict[str, Any]:
|
||||
)
|
||||
|
||||
|
||||
@router.get('/search', response_model=Dict[str, Any])
|
||||
@router.get('/search', response_model=AnimeSearchResponse)
|
||||
async def search_anime(
|
||||
q: str = Query(..., min_length=2, description="Search query"),
|
||||
page: int = Query(1, ge=1),
|
||||
per_page: int = Query(20, ge=1, le=100),
|
||||
current_user: Optional[Dict] = Depends(get_current_user),
|
||||
series_app: SeriesApp = Depends(get_series_app)
|
||||
) -> Dict[str, Any]:
|
||||
) -> AnimeSearchResponse:
|
||||
"""
|
||||
Search anime by name using SeriesApp.
|
||||
|
||||
@@ -393,29 +429,28 @@ async def search_anime(
|
||||
# Use SeriesApp to perform search
|
||||
search_results = series_app.search(q)
|
||||
|
||||
# Convert search results to our response format
|
||||
formatted_results = []
|
||||
# Convert search results to AnimeResponse objects
|
||||
anime_responses = []
|
||||
for result in search_results:
|
||||
anime_dict = {
|
||||
'id': getattr(result, 'id', str(uuid.uuid4())),
|
||||
'title': getattr(result, 'name', getattr(result, 'title', 'Unknown')),
|
||||
'description': getattr(result, 'description', ''),
|
||||
'status': 'available',
|
||||
'episodes': getattr(result, 'episodes', 0),
|
||||
'key': getattr(result, 'key', '')
|
||||
}
|
||||
formatted_results.append(anime_dict)
|
||||
anime_response = AnimeResponse(
|
||||
id=getattr(result, 'id', str(uuid.uuid4())),
|
||||
title=getattr(result, 'name', getattr(result, 'title', 'Unknown')),
|
||||
description=getattr(result, 'description', ''),
|
||||
status='available',
|
||||
episodes=getattr(result, 'episodes', 0),
|
||||
folder=getattr(result, 'key', '')
|
||||
)
|
||||
anime_responses.append(anime_response)
|
||||
|
||||
# Apply pagination
|
||||
total = len(formatted_results)
|
||||
total = len(anime_responses)
|
||||
start_idx = (page - 1) * per_page
|
||||
end_idx = start_idx + per_page
|
||||
paginated_results = formatted_results[start_idx:end_idx]
|
||||
paginated_results = anime_responses[start_idx:end_idx]
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"data": paginated_results,
|
||||
"pagination": {
|
||||
return AnimeSearchResponse(
|
||||
data=paginated_results,
|
||||
pagination={
|
||||
"page": page,
|
||||
"per_page": per_page,
|
||||
"total": total,
|
||||
@@ -423,11 +458,11 @@ async def search_anime(
|
||||
"has_next": end_idx < total,
|
||||
"has_prev": page > 1
|
||||
},
|
||||
"search": {
|
||||
search={
|
||||
"query": q,
|
||||
"total_results": total
|
||||
}
|
||||
}
|
||||
)
|
||||
except Exception as e:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
@@ -633,11 +668,11 @@ def bulk_anime_operation() -> Dict[str, Any]:
|
||||
message=f"Bulk {action} operation completed"
|
||||
)
|
||||
|
||||
@router.post('/rescan', response_model=Dict[str, Any])
|
||||
@router.post('/rescan', response_model=RescanResponse)
|
||||
async def rescan_anime_directory(
|
||||
current_user: Dict = Depends(get_current_user),
|
||||
series_app: SeriesApp = Depends(get_series_app)
|
||||
) -> Dict[str, Any]:
|
||||
) -> RescanResponse:
|
||||
"""
|
||||
Rescan the anime directory for new episodes and series.
|
||||
|
||||
@@ -653,11 +688,11 @@ async def rescan_anime_directory(
|
||||
|
||||
series_app.ReScan(progress_callback)
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"message": "Anime directory rescanned successfully",
|
||||
"total_series": len(series_app.series_list) if hasattr(series_app, 'series_list') else 0
|
||||
}
|
||||
return RescanResponse(
|
||||
success=True,
|
||||
message="Anime directory rescanned successfully",
|
||||
total_series=len(series_app.series_list) if hasattr(series_app, 'series_list') else 0
|
||||
)
|
||||
except Exception as e:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
|
||||
Reference in New Issue
Block a user