added resolving alerts collection

This commit is contained in:
TPV committed 2026-07-12 15:27:59 -07:00
1 parent 8270c9d7fd
commit 9bc34d4dc9
1 file changed
+28 -5
+28 -5
View File
@@ -34,6 +34,29 @@ def get_collection(collection_name: str) -> AsyncIOMotorCollection:
return get_mongo_client()[MONGO_DATABASE_NAME][collection_name] 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 <name>_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 @cache
def get_mongo_client() -> AsyncIOMotorClient: def get_mongo_client() -> AsyncIOMotorClient:
"""Return a cached Motor client configured for the shared cluster. """Return a cached Motor client configured for the shared cluster.
@@ -132,7 +155,7 @@ async def add_products_to_database(
try: try:
await collection.bulk_write(ops, ordered=False) await collection.bulk_write(ops, ordered=False)
if alert_ops: if alert_ops:
await write_alert_ops(alert_ops) await write_alert_ops(alert_ops, resolve_alerts_collection(collection))
except PyMongoError: except PyMongoError:
LOGGER.exception( LOGGER.exception(
"failed to add products to database", "failed to add products to database",
@@ -142,7 +165,7 @@ async def add_products_to_database(
async def write_alert_ops( 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. """Write alert upserts, swallowing duplicate-key races.
@@ -154,12 +177,12 @@ async def write_alert_ops(
Args: Args:
alert_ops: Alert upsert operations to write. 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: try:
await get_collection(alerts_collection_name).bulk_write( await alerts_collection.bulk_write(alert_ops, ordered=False)
alert_ops, ordered=False
)
except BulkWriteError as exc: except BulkWriteError as exc:
real_errors = [ real_errors = [
write_error write_error