Refactor item and lifecycle models to use Optional types and add validators for empty strings

- Updated the LifecycleUpdate and ItemType models to use Optional types for various fields, improving data handling.
- Added a validator to convert empty strings to None for specific fields in both models, enhancing input validation.
- Removed unnecessary checks in the benchmark API to streamline the upsert_benchmark function.

Author: maplethorpej

app/api/benchmark.py

@@ -7,7 +7,7 @@
 
 from models.base import User, CategoryBenchmark
 from utils.auth import authenticate
-from utils.gear_lifecycle import get_all_benchmarks, DEFAULT_BENCHMARKS
+from utils.gear_lifecycle import get_all_benchmarks
 
 logger = logging.getLogger(__name__)
 
@@ -28,9 +28,6 @@ class BenchmarkUpdate(BaseModel):
 
 @route.put("/{category_name}")
 def upsert_benchmark(category_name: str, payload: BenchmarkUpdate, user: User = Depends(authenticate)):
-    if category_name not in DEFAULT_BENCHMARKS:
-        raise HTTPException(400, f"Unknown category: {category_name}")
-
     override = db.session.query(CategoryBenchmark).filter_by(
         user_id=user.id, category_name=category_name
     ).first()

app/api/item.py

@@ -3,8 +3,8 @@
 
 from fastapi import APIRouter, Depends, HTTPException, File, UploadFile
 from fastapi_sqlalchemy import db
-from pydantic import BaseModel
-from typing import List
+from pydantic import BaseModel, validator
+from typing import List, Optional
 from io import StringIO
 from sqlalchemy import or_, func
 
@@ -38,15 +38,26 @@ class ItemType(BaseModel):
     product_url: str = None
     notes: str = None
 
-    acquired_date: str = None
-    acquisition_type: str = None
-    purchase_retailer: str = None
-    condition: str = None
-    status: str = None
-    retired_date: str = None
-    retired_reason: str = None
+    acquired_date: Optional[str] = None
+    acquisition_type: Optional[str] = None
+    purchase_retailer: Optional[str] = None
+    condition: Optional[str] = None
+    status: Optional[str] = None
+    retired_date: Optional[str] = None
+    retired_reason: Optional[str] = None
     replaced_by_id: int = None
 
+    @validator(
+        "acquired_date", "acquisition_type", "purchase_retailer",
+        "condition", "status", "retired_date", "retired_reason",
+        "product_url", "notes",
+        pre=True, always=True,
+    )
+    def empty_str_to_none(cls, v):
+        if isinstance(v, str) and v.strip() == "":
+            return None
+        return v
+
 
 def _find_catalog_product(session, brand_id: int, product_id: int, product_variant_id: int | None):
     q = session.query(CatalogProduct).filter(

app/api/item_lifecycle.py

@@ -3,7 +3,7 @@
 
 from fastapi import APIRouter, Depends, HTTPException
 from fastapi_sqlalchemy import db
-from pydantic import BaseModel
+from pydantic import BaseModel, validator
 from typing import Optional
 
 from models.base import User, Item, ItemLog
@@ -21,15 +21,25 @@
 
 
 class LifecycleUpdate(BaseModel):
-    acquired_date: str = None
-    acquisition_type: str = None
-    purchase_retailer: str = None
-    condition: str = None
-    status: str = None
-    retired_date: str = None
-    retired_reason: str = None
+    acquired_date: Optional[str] = None
+    acquisition_type: Optional[str] = None
+    purchase_retailer: Optional[str] = None
+    condition: Optional[str] = None
+    status: Optional[str] = None
+    retired_date: Optional[str] = None
+    retired_reason: Optional[str] = None
     replaced_by_id: int = None
 
+    @validator(
+        "acquired_date", "acquisition_type", "purchase_retailer",
+        "condition", "status", "retired_date", "retired_reason",
+        pre=True, always=True,
+    )
+    def empty_str_to_none(cls, v):
+        if isinstance(v, str) and v.strip() == "":
+            return None
+        return v
+
 
 @route.put("/{item_id}/lifecycle")
 def update_lifecycle(item_id: int, payload: LifecycleUpdate, user: User = Depends(authenticate)):
@@ -152,13 +162,15 @@ def get_replacement_score(item_id: int, user: User = Depends(authenticate)):
     )
 
     benchmark = get_benchmark(db.session, user.id, category_name)
+    is_default_fallback = benchmark.pop("is_default_fallback", False)
     score = replacement_score(item.acquired_date, item.condition, benchmark)
 
     return {
         "item_id": item.id,
         "score": score,
         "category": category_name,
         "benchmark": benchmark,
+        "is_default_fallback": is_default_fallback,
         "acquired_date": str(item.acquired_date) if item.acquired_date else None,
         "condition": item.condition,
     }

app/utils/gear_lifecycle.py

@@ -1,6 +1,8 @@
 import datetime
 
-from models.base import CategoryBenchmark
+from sqlalchemy import or_
+
+from models.base import Category, CategoryBenchmark
 
 DEFAULT_BENCHMARKS = {
     "Shelter":       {"lifespan_years": 5,  "expected_nights": 300},
@@ -31,38 +33,72 @@
 }
 
 
+def _apply_override(merged: dict, override, fields=_BENCHMARK_FIELDS) -> dict:
+    for field in fields:
+        val = getattr(override, field, None)
+        if val is not None:
+            merged[field] = float(val) if field != "distance_unit" else val
+    return merged
+
+
 def get_benchmark(session, user_id: int, category_name: str) -> dict:
+    is_known = category_name in DEFAULT_BENCHMARKS
     defaults = DEFAULT_BENCHMARKS.get(category_name, DEFAULT_BENCHMARKS["Miscellaneous"])
+
     override = session.query(CategoryBenchmark).filter_by(
         user_id=user_id, category_name=category_name
     ).first()
-    if not override:
-        return dict(defaults)
+
     merged = dict(defaults)
-    for field in _BENCHMARK_FIELDS:
-        val = getattr(override, field, None)
-        if val is not None:
-            merged[field] = float(val) if field != "distance_unit" else val
+    if override:
+        _apply_override(merged, override)
+        merged["is_default_fallback"] = False
+    else:
+        merged["is_default_fallback"] = not is_known
     return merged
 
 
+def _user_category_names(session, user_id: int) -> set[str]:
+    rows = session.query(Category.name).filter(
+        or_(Category.user_id == user_id, Category.user_id.is_(None))
+    ).all()
+    return {r.name for r in rows if r.name}
+
+
 def get_all_benchmarks(session, user_id: int) -> dict[str, dict]:
     overrides = session.query(CategoryBenchmark).filter_by(user_id=user_id).all()
     override_map = {o.category_name: o for o in overrides}
 
+    user_cats = _user_category_names(session, user_id)
+
     result = {}
+
     for cat_name, defaults in DEFAULT_BENCHMARKS.items():
         merged = dict(defaults)
+        merged["is_default_fallback"] = False
+        override = override_map.get(cat_name)
+        if override:
+            _apply_override(merged, override)
+            merged["has_override"] = True
+        else:
+            merged["has_override"] = False
+        result[cat_name] = merged
+
+    misc_defaults = DEFAULT_BENCHMARKS["Miscellaneous"]
+    for cat_name in user_cats:
+        if cat_name in result:
+            continue
+        merged = dict(misc_defaults)
+        merged["is_default_fallback"] = True
         override = override_map.get(cat_name)
         if override:
-            for field in _BENCHMARK_FIELDS:
-                val = getattr(override, field, None)
-                if val is not None:
-                    merged[field] = float(val) if field != "distance_unit" else val
+            _apply_override(merged, override)
             merged["has_override"] = True
+            merged["is_default_fallback"] = False
         else:
             merged["has_override"] = False
         result[cat_name] = merged
+
     return result