Attach live warehouse event stats to a flagsmith connection. No-op for non-flagsmith connections or when no warehouse is configured; leaves stats unset when the warehouse is unreachable. Read-only: never changes status.
(
connection: WarehouseConnection,
environment_key: str,
)
| 748 | |
| 749 | |
| 750 | def annotate_warehouse_event_stats( |
| 751 | connection: WarehouseConnection, |
| 752 | environment_key: str, |
| 753 | ) -> None: |
| 754 | """Attach live warehouse event stats to a flagsmith connection. No-op for |
| 755 | non-flagsmith connections or when no warehouse is configured; leaves stats |
| 756 | unset when the warehouse is unreachable. Read-only: never changes status.""" |
| 757 | if ( |
| 758 | connection.warehouse_type != WarehouseType.FLAGSMITH |
| 759 | or not settings.EXPERIMENTATION_CLICKHOUSE_URL |
| 760 | ): |
| 761 | return |
| 762 | try: |
| 763 | connection.event_stats = get_warehouse_event_stats(environment_key) |
| 764 | except Exception: |
| 765 | return |
searching dependent graphs…