added enums for status constants/fixed status not updating

src/monitor_utils/config.py

@@ -17,11 +17,13 @@ class DevelopmentMode(str, Enum):
 
 def is_production(development_mode: DevelopmentMode) -> bool:
     """Return True when the mode is PRODUCTION."""
+
     return development_mode == DevelopmentMode.PRODUCTION
 
 
 def get_bot_token(is_production: bool) -> str:
     """Return the production or development bot token for the mode."""
+
     if is_production:
         return PRODUCTION_BOT_TOKEN
 
@@ -45,6 +47,7 @@ def get_config(
     Returns:
         Tuple of the parsed config dict and the resolved config path.
     """
+
     monitor_name = monitor_name.replace(".json", "")
     base = config_dir if use_production_path else test_config_dir
     config_path = pathlib.Path(base) / f"{monitor_name}.json"
@@ -54,5 +57,6 @@ def get_config(
 
 def set_logger_levels(loggers: list[str] = LOGGERS, level: int = logging.WARNING):
     """Set the given loggers to the given level to quiet noisy libraries."""
+
     for name in loggers:
         logging.getLogger(name).setLevel(level)

src/monitor_utils/constants.py

@@ -6,6 +6,7 @@ and the expected error tuple used for known-failure handling.
 """
 
 import asyncio
+from enum import Enum
 from json import JSONDecodeError
 
 import aiohttp
@@ -58,10 +59,6 @@ BUCKET_WEBHOOK_MAP = {
     90: "90-100",
     "price_drop": "40-49",
 }
-STATUS_PENDING = "pending"
-STATUS_SENT = "sent"
-STATUS_SKIPPED = "skipped"
-STATUS_FAILED = "failed"
 
 EXPECTED_TASK_ERRORS = (
     RequestsError,
@@ -72,3 +69,12 @@ EXPECTED_TASK_ERRORS = (
     PyMongoError,
     JSONDecodeError,
 )
+
+
+class ProductStatus(str, Enum):
+    """Lifecycle status for a monitored product document."""
+
+    PENDING = "pending"
+    SENT = "sent"
+    SKIPPED = "skipped"
+    FAILED = "failed"

src/monitor_utils/database.py

@@ -1,3 +1,10 @@
+"""Database helpers for upserting monitored products and tracking price history.
+
+Provides the shared Motor client, collection access, and the upsert path that
+rewrites current product fields while preserving prior prices in a pastPrices
+map and stamping lastUpdated on every write.
+"""
+
 import logging
 from dataclasses import asdict
 from datetime import datetime, timezone
@@ -7,18 +14,33 @@ from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorCollection
 from pymongo import UpdateOne
 from pymongo.errors import PyMongoError
 
-from monitor_utils.constants import MONGO_DATABASE_NAME, MONGO_URL, STATUS_PENDING
+from monitor_utils.constants import ProductStatus, MONGO_DATABASE_NAME, MONGO_URL
 from monitor_utils.models import Product
 
 LOGGER = logging.getLogger(__name__)
 
 
 def get_collection(collection_name: str) -> AsyncIOMotorCollection:
+    """Return the named collection from the shared client's database.
+
+    Args:
+        collection_name: Name of the collection to return.
+
+    Returns:
+        The requested collection.
+    """
+
     return get_mongo_client()[MONGO_DATABASE_NAME][collection_name]
 
 
 @cache
 def get_mongo_client() -> AsyncIOMotorClient:
+    """Return a cached Motor client configured for the shared cluster.
+
+    Returns:
+        The pooled, timezone-aware Motor client.
+    """
+
     return AsyncIOMotorClient(
         MONGO_URL,
         maxPoolSize=200,
@@ -34,6 +56,8 @@ async def get_product_from_database(
     collection: AsyncIOMotorCollection,
     product_id: str,
 ) -> dict:
+    """Fetch a single product document by its _id."""
+
     return await collection.find_one({"_id": product_id})
 
 
@@ -41,18 +65,38 @@ async def add_products_to_database(
     products: list[Product],
     collection: AsyncIOMotorCollection,
     category_identifier: str,
+    use_composite: bool = False,
 ):
+    """Upsert products, tracking price history and last-updated time.
+
+    Rewrites current fields on each upsert while preserving prior prices in a
+    pastPrices map keyed by epoch-millisecond timestamps, appending only when
+    the price differs from the stored value. Flags dropped prices as pending
+    and records the drop's prior price.
+
+    Args:
+        products: Products to upsert.
+        collection: Target MongoDB collection.
+        category_identifier: Category suffix appended to composite ids.
+        use_composite: Build composite ids so multi-variant monitors (Woot)
+            keep one doc per variant; other monitors collapse to one doc per
+            product_id.
+    """
     if not products:
         return
 
-    composite_ids = [f"{p.product_id}_{category_identifier}" for p in products]
+    doc_ids = [
-    cursor = collection.find({"_id": {"$in": composite_ids}}, {"currentPrice": 1})
+        build_doc_id(p.product_id, category_identifier, use_composite) for p in products
+    ]
+    cursor = collection.find({"_id": {"$in": doc_ids}}, {"currentPrice": 1})
     existing_prices = {str(doc["_id"]): doc.get("currentPrice") async for doc in cursor}
 
     ops = []
-    for product in products:
+    now = datetime.now(timezone.utc)
+    price_key = str(int(now.timestamp() * 1000))
+    for product, doc_id in zip(products, doc_ids):
         document = convert_product_to_mongo(product, category_identifier)
-        doc_id = document["_id"]
+        document["_id"] = doc_id
         old_price = existing_prices.get(doc_id)
         new_price = document.get("currentPrice")
 
@@ -60,20 +104,18 @@ async def add_products_to_database(
             new_price is not None and new_price < old_price
         )
 
-        set_on_insert = {"discoveredDate": datetime.now(timezone.utc)}
+        set_on_insert = {"discoveredDate": now}
         if price_dropped:
-            document["status"] = STATUS_PENDING
+            document["status"] = ProductStatus.PENDING
             document["priceBeforeDrop"] = old_price
         else:
-            set_on_insert["status"] = STATUS_PENDING
+            set_on_insert["status"] = ProductStatus.PENDING
 
-        ops.append(
+        update = {"$set": document, "$setOnInsert": set_on_insert}
-            UpdateOne(
+        if new_price is not None and new_price != old_price:
-                {"_id": doc_id},
+            update["$set"][f"pastPrices.{price_key}"] = new_price
-                {"$set": document, "$setOnInsert": set_on_insert},
+
-                upsert=True,
+        ops.append(UpdateOne({"_id": doc_id}, update, upsert=True))
-            )
-        )
 
     try:
         await collection.bulk_write(ops, ordered=False)
@@ -85,7 +127,35 @@ async def add_products_to_database(
         raise
 
 
-def convert_product_to_mongo(product: Product, category_identifier: str) -> dict:
+def build_doc_id(
+    product_id: str, category_identifier: str, use_composite: bool = False
+) -> str:
+    """Return the composite id when use_composite is set, else the bare product id.
+
+    Args:
+        product_id: Retailer-specific product identifier.
+        category_identifier: Category suffix for multi-variant monitors.
+        use_composite: Append category_identifier to disambiguate variants.
+
+    Returns:
+        The composite id when use_composite is set, otherwise product_id.
+    """
+    return f"{product_id}_{category_identifier}" if use_composite else product_id
+
+
+def convert_product_to_mongo(product: Product) -> dict:
+    """Flatten a Product into its MongoDB document representation.
+
+    Coerces string prices to floats and stamps lastUpdated with the current
+    UTC time. The _id is assigned by the caller via build_doc_id.
+
+    Args:
+        product: Product to convert.
+
+    Returns:
+        The MongoDB document dict for the product.
+    """
+
     data = asdict(product)
     metadata = data["metadata"]
     timing = data["timing"]
@@ -98,7 +168,6 @@ def convert_product_to_mongo(product: Product, category_identifier: str) -> dict
         original_price = float(original_price.replace(",", ""))
 
     return {
-        "_id": f"{data['product_id']}_{category_identifier}",
         "title": data["title"],
         "url": data["url"],
         "currentPrice": current_price,
@@ -117,4 +186,5 @@ def convert_product_to_mongo(product: Product, category_identifier: str) -> dict
         "appExclusive": availability["app_exclusive"],
         "memberPick": availability["member_pick"],
         "memberDeal": availability["member_deal"],
+        "lastUpdated": datetime.now(timezone.utc),
     }

src/monitor_utils/runner.py

@@ -58,6 +58,7 @@ def init_monitor(
     Returns:
         Tuple of category names, Mongo collection, and loaded proxies.
     """
+
     use_production = is_production(development_mode)
     setup_logging(
         name="logs/latest.log",
@@ -90,6 +91,7 @@ def setup_config(
     Returns:
         Tuple of category names, channel id mapping, and Mongo collection.
     """
+
     use_production = is_production(development_mode)
     config, _ = get_config(
         monitor_name=monitor_name,
@@ -122,6 +124,7 @@ def get_channel_id(
     Returns:
         The resolved channel id.
     """
+
     if product.availability.app_exclusive:
         return get_single_channel_id(channel_ids["appExclusive"])
 
@@ -147,6 +150,7 @@ def get_channel_id(
 
 def get_single_channel_id(value: str | list[str]) -> str:
     """Return the first channel id when given a list, else the value itself."""
+
     if isinstance(value, list):
         return value[0]
 
@@ -174,6 +178,7 @@ def build_product_details(
     Returns:
         The built Product, or None when skipped or below thresholds.
     """
+
     product_id = product["_id"]
     current_price = product["currentPrice"]
     original_price = product["originalPrice"]
@@ -259,6 +264,7 @@ async def execute_monitor_logic(
         use_tls: Use the TLS fingerprint session instead of the base session.
         **kwargs: Forwarded verbatim to pre_task_func and every task_func call.
     """
+
     while True:
         session_ctx = (
             get_tls_session(
@@ -344,6 +350,7 @@ async def execute_pagination(
             query_category_id) must be supplied here, since naming varies
             per monitor.
     """
+
     LOGGER.info("executing pagination", extra={"category": category_identifier})
     total_pages = (int(total_products) + products_per_page - 1) // products_per_page
     semaphore = asyncio.Semaphore(max_concurrent_tasks)
@@ -410,6 +417,7 @@ async def fetch_all_pages(
     Returns:
         Page numbers whose fetch raised an expected error.
     """
+
     tasks = [fetch_func(page) for page in page_range]
     results = await asyncio.gather(*tasks, return_exceptions=True)
     return handle_page_results(
@@ -440,8 +448,8 @@ def handle_page_results(
     Returns:
         Page numbers whose fetch raised an expected error.
     """
-    failed_pages = []
 
+    failed_pages = []
     for page_number, result in zip(page_numbers, results):
         if isinstance(result, EXPECTED_TASK_ERRORS):
             LOGGER.warning(
@@ -503,6 +511,7 @@ async def retry_failed_pages(
     Returns:
         Page numbers still failing after all retries.
     """
+
     retry_attempt = 1
     while failed_pages and retry_attempt <= max_retries:
         await asyncio.sleep(2**retry_attempt)

src/monitor_utils/sessions.py

@@ -63,6 +63,7 @@ async def get_base_session(
     Yields:
         The configured aiohttp session.
     """
+
     headers = headers if headers else {}
     headers.setdefault("User-Agent", UA.random)
     session = aiohttp.ClientSession(
@@ -95,6 +96,7 @@ async def get_tls_session(
     Yields:
         The configured curl_cffi session.
     """
+
     headers = headers if headers else {}
     headers.setdefault("User-Agent", UA.chrome)
     async with AsyncSession(
@@ -115,6 +117,7 @@ def build_session_details(
         session: Session whose headers are updated in place.
         headers: Headers to merge; User-Agent is defaulted if absent.
     """
+
     headers = headers or {}
     headers.setdefault("User-Agent", UA.random)
     session.headers.update(headers)
@@ -174,6 +177,7 @@ async def send_request(
         json.JSONDecodeError: When parsing fails and reraise_json_error is
             True.
     """
+
     verify = True
     if development_proxy:
         proxy = development_proxy