Jonathan
486749a890
Full-stack Dutch supermarket price tracker with FastAPI backend, PostgreSQL/SQLAlchemy, Albert Heijn scraper, and Next.js frontend. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
65 lines
2.1 KiB
Python
65 lines
2.1 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session, selectinload
|
|
|
|
from ..database import get_db
|
|
from ..models import PriceSnapshot, Product
|
|
from ..schemas import PriceSnapshot as PriceSnapshotSchema, ProductWithLatestPrice
|
|
|
|
router = APIRouter(prefix="/api/products", tags=["products"])
|
|
|
|
|
|
def _attach_latest_price(product: Product, db: Session) -> ProductWithLatestPrice:
|
|
p = ProductWithLatestPrice.model_validate(product)
|
|
latest = db.scalar(
|
|
select(PriceSnapshot)
|
|
.where(PriceSnapshot.product_id == product.id)
|
|
.order_by(PriceSnapshot.timestamp.desc())
|
|
.limit(1)
|
|
)
|
|
if latest:
|
|
p.latest_price = latest.price
|
|
p.latest_price_timestamp = latest.timestamp
|
|
p.is_on_sale = latest.is_on_sale
|
|
return p
|
|
|
|
|
|
@router.get("", response_model=list[ProductWithLatestPrice])
|
|
def search_products(
|
|
search: str = Query(default=""),
|
|
limit: int = Query(default=20, le=100),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
q = select(Product).options(selectinload(Product.store))
|
|
if search:
|
|
q = q.where(Product.name.ilike(f"%{search}%"))
|
|
q = q.order_by(Product.name).limit(limit)
|
|
products = db.scalars(q).all()
|
|
return [_attach_latest_price(p, db) for p in products]
|
|
|
|
|
|
@router.get("/{product_id}", response_model=ProductWithLatestPrice)
|
|
def get_product(product_id: int, db: Session = Depends(get_db)):
|
|
product = db.scalar(
|
|
select(Product)
|
|
.where(Product.id == product_id)
|
|
.options(selectinload(Product.store))
|
|
)
|
|
if not product:
|
|
raise HTTPException(status_code=404, detail="Product not found")
|
|
return _attach_latest_price(product, db)
|
|
|
|
|
|
@router.get("/{product_id}/prices", response_model=list[PriceSnapshotSchema])
|
|
def get_product_prices(
|
|
product_id: int,
|
|
limit: int = Query(default=200, le=1000),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
return db.scalars(
|
|
select(PriceSnapshot)
|
|
.where(PriceSnapshot.product_id == product_id)
|
|
.order_by(PriceSnapshot.timestamp.asc())
|
|
.limit(limit)
|
|
).all()
|