234 lines
7.2 KiB
Python
234 lines
7.2 KiB
Python
import sqlite3
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
|
|
from app.database import get_db
|
|
from app.models import Review, ReviewListResponse
|
|
|
|
router = APIRouter(prefix="/reviews", tags=["Reviews"])
|
|
|
|
@router.get(
|
|
"",
|
|
response_model=ReviewListResponse,
|
|
summary="List reviews",
|
|
description="Retrieve a paginated list of reviews with optional filters. All filters are combined with AND logic, where only reviews matching every specified filter are returned.",
|
|
)
|
|
def list_reviews(
|
|
category: str | None = Query(
|
|
None, description="Filter by media category (exact match).", examples=["Movies"]
|
|
),
|
|
creator: str | None = Query(
|
|
None, description="Filter by creator name (exact match).", examples=["Christopher Nolan"]
|
|
),
|
|
genre: str | None = Query(
|
|
None, description="Filter by genre (exact match).", examples=["Science Fiction"]
|
|
),
|
|
min_rating: float | None = Query(
|
|
None, description="Minimum rating (inclusive).", ge=-3.0, le=3.0, examples=[2.0]
|
|
),
|
|
max_rating: float | None = Query(
|
|
None, description="Maximum rating (inclusive).", ge=-3.0, le=3.0, examples=[3.0]
|
|
),
|
|
date_from: str | None = Query(
|
|
None, description="Earliest review date (inclusive, YYYY-MM-DD).", examples=["2020-01-01"]
|
|
),
|
|
date_to: str | None = Query(
|
|
None, description="Latest review date (inclusive, YYYY-MM-DD).", examples=["2025-12-31"]
|
|
),
|
|
year_from: int | None = Query(
|
|
None, description="Earliest release year (inclusive).", examples=[1990]
|
|
),
|
|
year_to: int | None = Query(
|
|
None, description="Latest release year (inclusive).", examples=[2000]
|
|
),
|
|
limit: int = Query(
|
|
20, description="Number of results per page.", ge=1, le=100
|
|
),
|
|
offset: int = Query(
|
|
0, description="Number of results to skip.", ge=0
|
|
),
|
|
db: sqlite3.Connection = Depends(get_db),
|
|
):
|
|
conditions = []
|
|
params = []
|
|
|
|
if category:
|
|
conditions.append("category = ?")
|
|
params.append(category)
|
|
if creator:
|
|
conditions.append("creator = ?")
|
|
params.append(creator)
|
|
if genre:
|
|
conditions.append("genre = ?")
|
|
params.append(genre)
|
|
if min_rating is not None:
|
|
conditions.append("rating >= ?")
|
|
params.append(min_rating)
|
|
if max_rating is not None:
|
|
conditions.append("rating <= ?")
|
|
params.append(max_rating)
|
|
if date_from:
|
|
conditions.append("date >= ?")
|
|
params.append(date_from)
|
|
if date_to:
|
|
conditions.append("date <= ?")
|
|
params.append(date_to)
|
|
if year_from is not None:
|
|
conditions.append("year >= ?")
|
|
params.append(year_from)
|
|
if year_to is not None:
|
|
conditions.append("year <= ?")
|
|
params.append(year_to)
|
|
|
|
where = ""
|
|
if conditions:
|
|
where = "WHERE " + " AND ".join(conditions)
|
|
|
|
total = db.execute(
|
|
f"SELECT COUNT(*) FROM reviews {where}", params
|
|
).fetchone()[0]
|
|
|
|
rows = db.execute(
|
|
f"SELECT * FROM reviews {where} ORDER BY date DESC LIMIT ? OFFSET ?",
|
|
params + [limit, offset],
|
|
).fetchall()
|
|
|
|
return ReviewListResponse(
|
|
total=total,
|
|
limit=limit,
|
|
offset=offset,
|
|
results=[dict(row) for row in rows],
|
|
)
|
|
|
|
@router.get(
|
|
"/random",
|
|
response_model=Review,
|
|
summary="Get a random review",
|
|
description="Returns a single randomly selected review. Optionally filter by category to get a random review from a specific media type.",
|
|
)
|
|
def random_review(
|
|
category: str | None = Query(
|
|
None, description="Limit random selection to this category.", examples=["Books"]
|
|
),
|
|
db: sqlite3.Connection = Depends(get_db),
|
|
):
|
|
conditions = []
|
|
params = []
|
|
|
|
if category:
|
|
conditions.append("category = ?")
|
|
params.append(category)
|
|
|
|
where = ""
|
|
if conditions:
|
|
where = "WHERE " + " AND ".join(conditions)
|
|
|
|
row = db.execute(
|
|
f"SELECT * FROM reviews {where} ORDER BY RANDOM() LIMIT 1", params
|
|
).fetchone()
|
|
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="No reviews found.")
|
|
|
|
return dict(row)
|
|
|
|
@router.get(
|
|
"/top",
|
|
response_model=list[Review],
|
|
summary="Get top-rated reviews",
|
|
description="Returns the highest-rated reviews, sorted by rating descending.",
|
|
)
|
|
def top_reviews(
|
|
category: str | None = Query(None, description="Limit to this category.", examples=["Movies"]),
|
|
year_from: int | None = Query(None, description="Earliest release year (inclusive).", examples=[1990]),
|
|
year_to: int | None = Query(None, description="Latest release year (inclusive).", examples=[2000]),
|
|
limit: int = Query(10, description="Number of results to return.", ge=1, le=50),
|
|
db: sqlite3.Connection = Depends(get_db),
|
|
):
|
|
conditions = []
|
|
params = []
|
|
|
|
if category:
|
|
conditions.append("category = ?")
|
|
params.append(category)
|
|
if year_from is not None:
|
|
conditions.append("year >= ?")
|
|
params.append(year_from)
|
|
if year_to is not None:
|
|
conditions.append("year <= ?")
|
|
params.append(year_to)
|
|
|
|
where = ""
|
|
if conditions:
|
|
where = "WHERE " + " AND ".join(conditions)
|
|
|
|
rows = db.execute(
|
|
f"SELECT * FROM reviews {where} ORDER BY rating DESC LIMIT ?",
|
|
params + [limit],
|
|
).fetchall()
|
|
|
|
return [dict(row) for row in rows]
|
|
|
|
@router.get(
|
|
"/bottom",
|
|
response_model=list[Review],
|
|
summary="Get lowest-rated reviews",
|
|
description="Returns the lowest-rated reviews, sorted by rating ascending. "
|
|
"These are the harshest takes in the collection.",
|
|
)
|
|
def bottom_reviews(
|
|
category: str | None = Query(None, description="Limit to this category.", examples=["Movies"]),
|
|
year_from: int | None = Query(None, description="Earliest release year (inclusive).", examples=[1990]),
|
|
year_to: int | None = Query(None, description="Latest release year (inclusive).", examples=[2000]),
|
|
limit: int = Query(10, description="Number of results to return.", ge=1, le=50),
|
|
db: sqlite3.Connection = Depends(get_db),
|
|
):
|
|
conditions = []
|
|
params = []
|
|
|
|
if category:
|
|
conditions.append("category = ?")
|
|
params.append(category)
|
|
if year_from is not None:
|
|
conditions.append("year >= ?")
|
|
params.append(year_from)
|
|
if year_to is not None:
|
|
conditions.append("year <= ?")
|
|
params.append(year_to)
|
|
|
|
where = ""
|
|
if conditions:
|
|
where = "WHERE " + " AND ".join(conditions)
|
|
|
|
rows = db.execute(
|
|
f"SELECT * FROM reviews {where} ORDER BY rating ASC LIMIT ?",
|
|
params + [limit],
|
|
).fetchall()
|
|
|
|
return [dict(row) for row in rows]
|
|
|
|
@router.get(
|
|
"/{category}/{title}/{creator}",
|
|
response_model=Review,
|
|
summary="Get a specific review",
|
|
description="Retrieve a single review by its unique combination of "
|
|
"category, title, and creator.",
|
|
)
|
|
def get_review(
|
|
category: str,
|
|
title: str,
|
|
creator: str,
|
|
db: sqlite3.Connection = Depends(get_db),
|
|
):
|
|
row = db.execute(
|
|
"SELECT * FROM reviews WHERE category = ? AND title = ? AND creator = ?",
|
|
[category, title, creator],
|
|
).fetchone()
|
|
|
|
if not row:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f"No review found for '{title}' by {creator} in {category}.",
|
|
)
|
|
|
|
return dict(row)
|