gilles
4ff5d3ee79
- Add automatic scraping when creating a new product
- Update product title and image from scraped data
- Add GET /products/{id}/snapshots endpoint for price history
- Add list_snapshots and get_latest_snapshot to CRUD
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
65 lines
1.8 KiB
Python
65 lines
1.8 KiB
Python
from __future__ import annotations
|
|
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from backend.app.db import models, schemas
|
|
|
|
|
|
"""
|
|
CRUD minimal pour manipuler les produits dans la base SQLite.
|
|
"""
|
|
|
|
|
|
def get_product(db: Session, product_id: int) -> models.Product | None:
|
|
return db.query(models.Product).filter(models.Product.id == product_id).first()
|
|
|
|
|
|
def list_products(db: Session, skip: int = 0, limit: int = 100) -> list[models.Product]:
|
|
return db.query(models.Product).offset(skip).limit(limit).all()
|
|
|
|
|
|
def create_product(db: Session, data: schemas.ProductCreate) -> models.Product:
|
|
product = models.Product(**data.dict())
|
|
db.add(product)
|
|
try:
|
|
db.commit()
|
|
except IntegrityError:
|
|
db.rollback()
|
|
raise
|
|
db.refresh(product)
|
|
return product
|
|
|
|
|
|
def update_product(db: Session, product: models.Product, changes: schemas.ProductUpdate) -> models.Product:
|
|
for field, value in changes.dict(exclude_unset=True).items():
|
|
setattr(product, field, value)
|
|
db.add(product)
|
|
db.commit()
|
|
db.refresh(product)
|
|
return product
|
|
|
|
|
|
def remove_product(db: Session, product: models.Product) -> None:
|
|
db.delete(product)
|
|
db.commit()
|
|
|
|
|
|
def list_snapshots(db: Session, product_id: int, limit: int = 30) -> list[models.ProductSnapshot]:
|
|
return (
|
|
db.query(models.ProductSnapshot)
|
|
.filter(models.ProductSnapshot.produit_id == product_id)
|
|
.order_by(models.ProductSnapshot.scrape_le.desc())
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
|
|
def get_latest_snapshot(db: Session, product_id: int) -> models.ProductSnapshot | None:
|
|
return (
|
|
db.query(models.ProductSnapshot)
|
|
.filter(models.ProductSnapshot.produit_id == product_id)
|
|
.order_by(models.ProductSnapshot.scrape_le.desc())
|
|
.first()
|
|
)
|