testing price tracking/composite key changes

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/database.py

@@ -14,11 +14,26 @@ 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 +49,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 +58,42 @@ 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})
+        id(p): build_doc_id(p.product_id, category_identifier, use_composite)
+        for p in products
+    }
+    cursor = collection.find(
+        {"_id": {"$in": list(doc_ids.values())}}, {"currentPrice": 1}
+    )
     existing_prices = {str(doc["_id"]): doc.get("currentPrice") async for doc in cursor}
 
     ops = []
+    now = datetime.now(timezone.utc)
+    price_key = str(int(now.timestamp() * 1000))
     for product in products:
+        doc_id = doc_ids[id(product)]
         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 +101,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["priceBeforeDrop"] = old_price
         else:
             set_on_insert["status"] = STATUS_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 +124,36 @@ async def add_products_to_database(
         raise
 
 
+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, category_identifier: str) -> dict:
+    """Flatten a Product into its MongoDB document representation.
+
+    Coerces string prices to floats and stamps lastUpdated with the current
+    UTC time.
+
+    Args:
+        product: Product to convert.
+        category_identifier: Category suffix appended to the composite _id.
+
+    Returns:
+        The MongoDB document dict for the product.
+    """
+
     data = asdict(product)
     metadata = data["metadata"]
     timing = data["timing"]
@@ -117,4 +185,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/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