From 9bc34d4dc9e23d536a33b72403dd9fc3bcb13814 Mon Sep 17 00:00:00 2001 From: TPV Date: Sun, 12 Jul 2026 15:27:59 -0700 Subject: [PATCH] added resolving alerts collection --- src/monitor_utils/database.py | 33 ++++++++++++++++++++++++++++----- 1 file changed, 28 insertions(+), 5 deletions(-) diff --git a/src/monitor_utils/database.py b/src/monitor_utils/database.py index 69c58e2..5224fa3 100644 --- a/src/monitor_utils/database.py +++ b/src/monitor_utils/database.py @@ -34,6 +34,29 @@ def get_collection(collection_name: str) -> AsyncIOMotorCollection: return get_mongo_client()[MONGO_DATABASE_NAME][collection_name] +def resolve_alerts_collection( + product_collection: AsyncIOMotorCollection, +) -> AsyncIOMotorCollection: + """Return the alerts collection beside a product collection. + + Alerts co-locate with the products they reference: the same database and + the same development mode. The mode is read from the product collection's + own name (setup_config appends a _testing suffix outside production), so a + monitor writing products to _testing also writes alerts to + alerts_testing and the development bot draining alerts_testing sees them. + + Args: + product_collection: The mode-resolved product collection the monitor + already writes to. + + Returns: + The matching alerts collection in the same database and mode. + """ + + suffix = "_testing" if product_collection.name.endswith("_testing") else "" + return product_collection.database[f"alerts{suffix}"] + + @cache def get_mongo_client() -> AsyncIOMotorClient: """Return a cached Motor client configured for the shared cluster. @@ -132,7 +155,7 @@ async def add_products_to_database( try: await collection.bulk_write(ops, ordered=False) if alert_ops: - await write_alert_ops(alert_ops) + await write_alert_ops(alert_ops, resolve_alerts_collection(collection)) except PyMongoError: LOGGER.exception( "failed to add products to database", @@ -142,7 +165,7 @@ async def add_products_to_database( async def write_alert_ops( - alert_ops: list[UpdateOne], alerts_collection_name: AsyncIOMotorCollection + alert_ops: list[UpdateOne], alerts_collection: AsyncIOMotorCollection ): """Write alert upserts, swallowing duplicate-key races. @@ -154,12 +177,12 @@ async def write_alert_ops( Args: alert_ops: Alert upsert operations to write. + alerts_collection: The mode-resolved alerts collection to write to, + supplied by the caller so the writer stays mode-agnostic. """ try: - await get_collection(alerts_collection_name).bulk_write( - alert_ops, ordered=False - ) + await alerts_collection.bulk_write(alert_ops, ordered=False) except BulkWriteError as exc: real_errors = [ write_error