Minor version bump (1.x.0) is appropriate because:

 New functionality added (soft delete system)
 Backward compatible (existing features unchanged)
 Significant enhancement (complete temporal tracking system)
 API additions (new endpoints, parameters)
 UI enhancements (new components, visual indicators)
This commit is contained in:
2025-05-30 09:49:26 +02:00
parent 56c3c16f6d
commit 0b42a74fe9
16 changed files with 1438 additions and 237 deletions

View File

@@ -1,7 +1,7 @@
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.middleware.cors import CORSMiddleware
from sqlalchemy.orm import Session
from sqlalchemy import text
from sqlalchemy import text, func
from typing import List
import models, schemas
from database import engine, get_db
@@ -95,6 +95,34 @@ def build_shopping_event_response(event: models.ShoppingEvent, db: Session) -> s
def read_root():
return {"message": __app_name__, "version": __version__, "name": "Groceries Tracker Backend"}
# Utility endpoints
@app.get("/current-date")
def get_current_date():
"""Get current date for use as default in valid_from fields"""
from datetime import date
return {"current_date": date.today().isoformat()}
@app.get("/products/available-for-shopping/{shopping_date}", response_model=List[schemas.Product])
def get_products_available_for_shopping(shopping_date: str, db: Session = Depends(get_db)):
"""Get products that were available (not deleted) on a specific shopping date"""
from datetime import datetime
try:
# Parse the shopping date
target_date = datetime.strptime(shopping_date, '%Y-%m-%d').date()
except ValueError:
raise HTTPException(status_code=400, detail="Invalid date format. Use YYYY-MM-DD format")
# Get products that were either:
# 1. Never deleted (deleted=False)
# 2. Deleted after the shopping date (valid_from > shopping_date for deleted=True products)
products = db.query(models.Product).filter(
(models.Product.deleted == False) |
((models.Product.deleted == True) & (models.Product.valid_from > target_date))
).all()
return products
# Product endpoints
@app.post("/products/", response_model=schemas.Product)
def create_product(product: schemas.ProductCreate, db: Session = Depends(get_db)):
@@ -109,15 +137,31 @@ def create_product(product: schemas.ProductCreate, db: Session = Depends(get_db)
if brand is None:
raise HTTPException(status_code=404, detail="Brand not found")
db_product = models.Product(**product.dict())
# Validate valid_from date if provided
if product.valid_from is not None:
from datetime import date
if product.valid_from > date.today():
raise HTTPException(status_code=400, detail="Valid from date cannot be in the future")
# Create product data
product_data = product.dict(exclude={'valid_from'})
db_product = models.Product(**product_data)
# Set valid_from if provided, otherwise let database default handle it
if product.valid_from is not None:
db_product.valid_from = product.valid_from
db.add(db_product)
db.commit()
db.refresh(db_product)
return db_product
@app.get("/products/", response_model=List[schemas.Product])
def read_products(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
products = db.query(models.Product).offset(skip).limit(limit).all()
def read_products(skip: int = 0, limit: int = 100, show_deleted: bool = False, db: Session = Depends(get_db)):
query = db.query(models.Product)
if not show_deleted:
query = query.filter(models.Product.deleted == False)
products = query.offset(skip).limit(limit).all()
return products
@app.get("/products/{product_id}", response_model=schemas.Product)
@@ -127,13 +171,21 @@ def read_product(product_id: int, db: Session = Depends(get_db)):
raise HTTPException(status_code=404, detail="Product not found")
return product
@app.get("/products/{product_id}/valid-from")
def get_product_valid_from(product_id: int, db: Session = Depends(get_db)):
"""Get the current valid_from date for a product (used for validation when editing)"""
product = db.query(models.Product).filter(models.Product.id == product_id).first()
if product is None:
raise HTTPException(status_code=404, detail="Product not found")
return {"valid_from": product.valid_from.isoformat()}
@app.put("/products/{product_id}", response_model=schemas.Product)
def update_product(product_id: int, product_update: schemas.ProductUpdate, db: Session = Depends(get_db)):
product = db.query(models.Product).filter(models.Product.id == product_id).first()
if product is None:
raise HTTPException(status_code=404, detail="Product not found")
update_data = product_update.dict(exclude_unset=True)
update_data = product_update.dict(exclude_unset=True, exclude={'valid_from'})
# Validate category exists if category_id is being updated
if 'category_id' in update_data:
@@ -147,9 +199,32 @@ def update_product(product_id: int, product_update: schemas.ProductUpdate, db: S
if brand is None:
raise HTTPException(status_code=404, detail="Brand not found")
# Validate valid_from date if provided
if product_update.valid_from is not None:
from datetime import date
if product_update.valid_from > date.today():
raise HTTPException(status_code=400, detail="Valid from date cannot be in the future")
if product_update.valid_from <= product.valid_from:
raise HTTPException(
status_code=400,
detail=f"Valid from date must be after the current product's valid from date ({product.valid_from})"
)
# Check if any versioned fields are actually changing
versioned_fields = ['name', 'category_id', 'brand_id', 'organic', 'weight', 'weight_unit']
has_changes = any(
field in update_data and getattr(product, field) != update_data[field]
for field in versioned_fields
)
# Apply the updates - trigger will handle history creation automatically
for field, value in update_data.items():
setattr(product, field, value)
# Set valid_from if provided for manual versioning
if product_update.valid_from is not None:
product.valid_from = product_update.valid_from
db.commit()
db.refresh(product)
return product
@@ -160,10 +235,163 @@ def delete_product(product_id: int, db: Session = Depends(get_db)):
if product is None:
raise HTTPException(status_code=404, detail="Product not found")
db.delete(product)
if product.deleted:
raise HTTPException(status_code=400, detail="Product is already deleted")
from datetime import date
# Simply mark as deleted and set valid_from to today
# The trigger will automatically create the history record
product.deleted = True
product.valid_from = date.today()
product.updated_at = func.now()
db.commit()
return {"message": "Product deleted successfully"}
# Historical Product endpoints
@app.get("/products/{product_id}/history", response_model=List[schemas.ProductHistory])
def get_product_history(product_id: int, db: Session = Depends(get_db)):
"""Get all historical versions of a product"""
# Check if product exists
product = db.query(models.Product).filter(models.Product.id == product_id).first()
if product is None:
raise HTTPException(status_code=404, detail="Product not found")
# Get history from history table
history = db.query(models.ProductHistory).filter(
models.ProductHistory.id == product_id
).order_by(models.ProductHistory.valid_from.desc()).all()
return history
@app.get("/products/{product_id}/at/{date}", response_model=schemas.ProductAtDate)
def get_product_at_date(product_id: int, date: str, db: Session = Depends(get_db)):
"""Get product as it existed at a specific date - CRUCIAL for shopping events"""
from datetime import datetime, date as date_type
try:
# Parse the date string (accept YYYY-MM-DD format)
target_date = datetime.strptime(date, '%Y-%m-%d').date()
except ValueError:
raise HTTPException(status_code=400, detail="Invalid date format. Use YYYY-MM-DD format")
# First try current products table
current_product = db.query(models.Product).filter(
models.Product.id == product_id,
models.Product.valid_from <= target_date,
models.Product.valid_to >= target_date
).first()
if current_product:
# Get related data
category = db.query(models.GroceryCategory).filter(
models.GroceryCategory.id == current_product.category_id
).first()
brand = None
if current_product.brand_id:
brand = db.query(models.Brand).filter(
models.Brand.id == current_product.brand_id
).first()
return schemas.ProductAtDate(
id=current_product.id,
name=current_product.name,
category_id=current_product.category_id,
category=category,
brand_id=current_product.brand_id,
brand=brand,
organic=current_product.organic,
weight=current_product.weight,
weight_unit=current_product.weight_unit,
valid_from=current_product.valid_from,
valid_to=current_product.valid_to,
deleted=current_product.deleted,
was_current=True
)
# Try history table
historical_product = db.query(models.ProductHistory).filter(
models.ProductHistory.id == product_id,
models.ProductHistory.valid_from <= target_date,
models.ProductHistory.valid_to >= target_date
).first()
if historical_product:
# Get related data (note: these might have changed too, but we'll use current versions)
category = db.query(models.GroceryCategory).filter(
models.GroceryCategory.id == historical_product.category_id
).first()
brand = None
if historical_product.brand_id:
brand = db.query(models.Brand).filter(
models.Brand.id == historical_product.brand_id
).first()
return schemas.ProductAtDate(
id=historical_product.id,
name=historical_product.name,
category_id=historical_product.category_id,
category=category,
brand_id=historical_product.brand_id,
brand=brand,
organic=historical_product.organic,
weight=historical_product.weight,
weight_unit=historical_product.weight_unit,
valid_from=historical_product.valid_from,
valid_to=historical_product.valid_to,
deleted=historical_product.deleted,
was_current=False
)
# Product didn't exist at that date
raise HTTPException(
status_code=404,
detail=f"Product {product_id} did not exist on {date}"
)
@app.get("/shopping-events/{event_id}/products-as-purchased", response_model=List[schemas.ProductAtPurchase])
def get_shopping_event_products_as_purchased(event_id: int, db: Session = Depends(get_db)):
"""Get products as they were when purchased - shows historical product data"""
# Get the shopping event
event = db.query(models.ShoppingEvent).filter(models.ShoppingEvent.id == event_id).first()
if event is None:
raise HTTPException(status_code=404, detail="Shopping event not found")
# Get products from association table
products_data = db.execute(
text("""
SELECT sep.product_id, sep.amount, sep.price, sep.discount
FROM shopping_event_products sep
WHERE sep.shopping_event_id = :event_id
"""),
{"event_id": event_id}
).fetchall()
result = []
for product_data in products_data:
# Get product as it was at the time of purchase
try:
# Extract just the date from the shopping event datetime
purchase_date = event.date.date().strftime('%Y-%m-%d')
product_at_purchase = get_product_at_date(
product_data.product_id,
purchase_date,
db
)
result.append(schemas.ProductAtPurchase(
product=product_at_purchase,
amount=product_data.amount,
price=product_data.price,
discount=product_data.discount
))
except HTTPException:
# Product didn't exist at purchase time (shouldn't happen, but handle gracefully)
continue
return result
# Shop endpoints
@app.post("/shops/", response_model=schemas.Shop)
def create_shop(shop: schemas.ShopCreate, db: Session = Depends(get_db)):