43 lines
1.3 KiB
Python
43 lines
1.3 KiB
Python
"""Ranking endpoints – listings sorted by ELO."""
|
||
|
||
from fastapi import APIRouter, Depends
|
||
from sqlalchemy import text
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.config import settings
|
||
from app.database import get_db
|
||
from app.queries import LISTING_SELECT, row_to_listing
|
||
from app.schemas import RankingResponse
|
||
|
||
router = APIRouter()
|
||
|
||
SAMPLE_JOIN = (
|
||
f" inner join {settings.ELO_SCHEMA}.sample_listings as s"
|
||
f" on l.global_id = s.global_id"
|
||
)
|
||
|
||
|
||
@router.get("/rankings", response_model=list[RankingResponse])
|
||
def get_rankings(
|
||
status: str | None = None,
|
||
limit: int = 100,
|
||
offset: int = 0,
|
||
db: Session = Depends(get_db),
|
||
):
|
||
"""Return listings ranked by ELO rating (highest first)."""
|
||
query = LISTING_SELECT + SAMPLE_JOIN
|
||
params: dict = {"limit": limit, "offset": offset, "default_elo": settings.DEFAULT_ELO}
|
||
|
||
if status and status != "all":
|
||
query += " where l.status = :status"
|
||
params["status"] = status
|
||
|
||
query += " order by elo_rating desc, l.current_price desc limit :limit offset :offset"
|
||
result = db.execute(text(query), params)
|
||
listings = [row_to_listing(row) for row in result]
|
||
|
||
return [
|
||
RankingResponse(rank=offset + i + 1, listing=listing)
|
||
for i, listing in enumerate(listings)
|
||
]
|