Merge pull request #1542 from PolicyEngine/codex/add-analytics-routing-version

Track Modal routing version in calculate analytics

Author: anth-volk

alembic/versions/20260519_0004_calculate_analytics_routing_version.py

@@ -0,0 +1,77 @@
+"""add calculate analytics routing version
+
+Revision ID: 20260519_0004
+Revises: 20260512_0003
+Create Date: 2026-05-19 22:05:42.821147
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "20260519_0004"
+down_revision = "20260512_0003"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table(
+        "calculate_request_variables", schema=None
+    ) as batch_op:
+        batch_op.add_column(
+            sa.Column("requested_version", sa.String(length=64), nullable=True)
+        )
+        batch_op.add_column(
+            sa.Column("resolved_channel", sa.String(length=16), nullable=True)
+        )
+        batch_op.create_index(
+            "ix_calc_vars_channel_created",
+            ["resolved_channel", "created_at"],
+            unique=False,
+        )
+        batch_op.create_index(
+            "ix_calc_vars_requested_created",
+            ["requested_version", "created_at"],
+            unique=False,
+        )
+
+    with op.batch_alter_table("calculate_requests", schema=None) as batch_op:
+        batch_op.add_column(
+            sa.Column("requested_version", sa.String(length=64), nullable=True)
+        )
+        batch_op.add_column(
+            sa.Column("resolved_channel", sa.String(length=16), nullable=True)
+        )
+        batch_op.create_index(
+            "ix_calculate_requests_channel_created",
+            ["resolved_channel", "created_at"],
+            unique=False,
+        )
+        batch_op.create_index(
+            "ix_calculate_requests_requested_created",
+            ["requested_version", "created_at"],
+            unique=False,
+        )
+
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table("calculate_requests", schema=None) as batch_op:
+        batch_op.drop_index("ix_calculate_requests_requested_created")
+        batch_op.drop_index("ix_calculate_requests_channel_created")
+        batch_op.drop_column("resolved_channel")
+        batch_op.drop_column("requested_version")
+
+    with op.batch_alter_table(
+        "calculate_request_variables", schema=None
+    ) as batch_op:
+        batch_op.drop_index("ix_calc_vars_requested_created")
+        batch_op.drop_index("ix_calc_vars_channel_created")
+        batch_op.drop_column("resolved_channel")
+        batch_op.drop_column("requested_version")
+
+    # ### end Alembic commands ###

changelog.d/1541.added.md

@@ -0,0 +1 @@
+Added Modal requested-version and resolved-channel metadata to calculate analytics.

policyengine_household_api/data/analytics_migration.py

@@ -1 +1 @@
-ANALYTICS_ALEMBIC_MINIMUM_REVISION = "20260512_0003"
+ANALYTICS_ALEMBIC_MINIMUM_REVISION = "20260519_0004"

policyengine_household_api/data/analytics_setup.py

@@ -46,6 +46,8 @@
         "api_version",
         "country_id",
         "model_version",
+        "requested_version",
+        "resolved_channel",
         "endpoint",
         "method",
         "content_length_bytes",
@@ -63,6 +65,8 @@
         "country_id",
         "api_version",
         "model_version",
+        "requested_version",
+        "resolved_channel",
         "response_status_code",
         "variable_name",
         "variable_name_truncated",

policyengine_household_api/data/models.py

@@ -2,6 +2,7 @@
 from policyengine_household_api.models.analytics import (
     AnalyticsHttpMethod,
     AvailabilityStatus,
+    ModalResolvedChannel,
     PeriodGranularity,
     VariableSource,
 )
@@ -52,6 +53,12 @@ class CalculateRequest(db.Model):
     api_version = mapped_column(String(32), nullable=True)
     country_id = mapped_column(String(16), nullable=False)
     model_version = mapped_column(String(64), nullable=True)
+    requested_version = mapped_column(String(64), nullable=True)
+    resolved_channel = mapped_column(
+        String(16),
+        nullable=True,
+        info={"options": _enum_values(ModalResolvedChannel)},
+    )
     endpoint = mapped_column(String(64), nullable=True)
     method = mapped_column(
         String(16),
@@ -79,6 +86,16 @@ class CalculateRequest(db.Model):
             "country_id",
             "created_at",
         ),
+        Index(
+            "ix_calculate_requests_channel_created",
+            "resolved_channel",
+            "created_at",
+        ),
+        Index(
+            "ix_calculate_requests_requested_created",
+            "requested_version",
+            "created_at",
+        ),
     )
 
 
@@ -98,6 +115,12 @@ class CalculateRequestVariable(db.Model):
     country_id = mapped_column(String(16), nullable=False)
     api_version = mapped_column(String(32), nullable=True)
     model_version = mapped_column(String(64), nullable=True)
+    requested_version = mapped_column(String(64), nullable=True)
+    resolved_channel = mapped_column(
+        String(16),
+        nullable=True,
+        info={"options": _enum_values(ModalResolvedChannel)},
+    )
     response_status_code = mapped_column(Integer, nullable=True)
     variable_name = mapped_column(String(255), nullable=False)
     variable_name_truncated = mapped_column(
@@ -147,4 +170,14 @@ class CalculateRequestVariable(db.Model):
             "model_version",
             "variable_name",
         ),
+        Index(
+            "ix_calc_vars_channel_created",
+            "resolved_channel",
+            "created_at",
+        ),
+        Index(
+            "ix_calc_vars_requested_created",
+            "requested_version",
+            "created_at",
+        ),
     )

policyengine_household_api/decorators/analytics.py

@@ -22,8 +22,13 @@
 )
 from policyengine_household_api.models.analytics import (
     AnalyticsContext,
+    ModalResolvedChannel,
     VariableUsageSummary,
 )
+from policyengine_household_api.utils.modal_routing_metadata import (
+    REQUESTED_VERSION_ENVIRON_KEY,
+    RESOLVED_CHANNEL_ENVIRON_KEY,
+)
 from policyengine_household_api.utils.variable_usage_analytics import (
     extract_variable_usage,
     stored_variable_name,
@@ -120,6 +125,7 @@ def _build_analytics_context(args, kwargs) -> AnalyticsContext:
     client_id = _client_id_from_request()
     country_id = _country_id_from_route_args(args, kwargs)
     payload = _request_json()
+    requested_version, resolved_channel = _routing_metadata_from_request()
 
     context = AnalyticsContext(
         client_id=client_id,
@@ -129,6 +135,8 @@ def _build_analytics_context(args, kwargs) -> AnalyticsContext:
         content_length_bytes=request.content_length,
         created_at=now,
         country_id=country_id,
+        requested_version=requested_version,
+        resolved_channel=resolved_channel,
     )
 
     if country_id is None or not _collect_variable_usage():
@@ -199,6 +207,22 @@ def _collect_variable_usage() -> bool:
     return bool(value)
 
 
+def _routing_metadata_from_request() -> tuple[
+    str | None, ModalResolvedChannel | None
+]:
+    requested_version = request.environ.get(REQUESTED_VERSION_ENVIRON_KEY)
+    resolved_channel = request.environ.get(RESOLVED_CHANNEL_ENVIRON_KEY)
+    if not isinstance(requested_version, str) or not requested_version:
+        return None, None
+
+    try:
+        channel = ModalResolvedChannel(resolved_channel)
+    except ValueError:
+        return None, None
+
+    return requested_version, channel
+
+
 def _record_analytics(
     context: AnalyticsContext | None,
     response_status_code: int | None,
@@ -281,6 +305,10 @@ def _build_calculate_request(
     calculate_request.api_version = context.api_version
     calculate_request.country_id = context.country_id
     calculate_request.model_version = context.model_version
+    calculate_request.requested_version = context.requested_version
+    calculate_request.resolved_channel = (
+        context.resolved_channel.value if context.resolved_channel else None
+    )
     calculate_request.endpoint = context.endpoint
     calculate_request.method = context.method.value
     calculate_request.content_length_bytes = context.content_length_bytes
@@ -307,6 +335,8 @@ def _build_calculate_request_variable(
     variable.country_id = calculate_request.country_id
     variable.api_version = calculate_request.api_version
     variable.model_version = calculate_request.model_version
+    variable.requested_version = calculate_request.requested_version
+    variable.resolved_channel = calculate_request.resolved_channel
     variable.response_status_code = calculate_request.response_status_code
     (
         variable.variable_name,

policyengine_household_api/endpoints/analytics.py

@@ -16,6 +16,7 @@
     CalculateRequest,
     CalculateRequestVariable,
 )
+from policyengine_household_api.models.analytics import ModalResolvedChannel
 
 
 DEFAULT_REQUEST_LIMIT = 1_000
@@ -28,6 +29,8 @@
 class CalculateAnalyticsQuery:
     start_time: datetime | None
     end_time: datetime | None
+    requested_version: str | None
+    resolved_channel: str | None
     unique: bool
     limit: int
 
@@ -50,6 +53,8 @@ def get_calculate_analytics_requests() -> Response:
         "message": None,
         "start_time": _datetime_to_json(query.start_time),
         "end_time": _datetime_to_json(query.end_time),
+        "requested_version": query.requested_version,
+        "resolved_channel": query.resolved_channel,
         "unique": query.unique,
     }
     if query.unique:
@@ -97,6 +102,12 @@ def _parse_query_args() -> CalculateAnalyticsQuery:
     return CalculateAnalyticsQuery(
         start_time=start_time,
         end_time=end_time,
+        requested_version=_parse_optional_string(
+            request.args.get("requested_version")
+        ),
+        resolved_channel=_parse_resolved_channel(
+            request.args.get("resolved_channel")
+        ),
         unique=_parse_bool(request.args.get("unique"), default=False),
         limit=_parse_limit(request.args.get("limit")),
     )
@@ -139,6 +150,23 @@ def _parse_bool(value: str | None, *, default: bool) -> bool:
     raise ValueError("`unique` must be true or false")
 
 
+def _parse_optional_string(value: str | None) -> str | None:
+    if value is None:
+        return None
+    value = value.strip()
+    return value or None
+
+
+def _parse_resolved_channel(value: str | None) -> str | None:
+    value = _parse_optional_string(value)
+    if value is None:
+        return None
+    if value not in {channel.value for channel in ModalResolvedChannel}:
+        allowed = ", ".join(channel.value for channel in ModalResolvedChannel)
+        raise ValueError(f"`resolved_channel` must be one of: {allowed}")
+    return value
+
+
 def _parse_limit(value: str | None) -> int:
     if value is None:
         return DEFAULT_REQUEST_LIMIT
@@ -156,7 +184,7 @@ def _parse_limit(value: str | None) -> int:
 def _calculate_requests(
     query: CalculateAnalyticsQuery,
 ) -> list[dict[str, Any]]:
-    request_query = _apply_time_filters(CalculateRequest.query, query)
+    request_query = _apply_filters(CalculateRequest.query, query)
     calculate_requests = (
         request_query.order_by(CalculateRequest.created_at.desc())
         .limit(query.limit)
@@ -204,7 +232,7 @@ def _variables_by_request_id(
 def _unique_variable_keys(
     query: CalculateAnalyticsQuery,
 ) -> list[dict[str, Any]]:
-    variable_query = _apply_time_filters(CalculateRequestVariable.query, query)
+    variable_query = _apply_filters(CalculateRequestVariable.query, query)
     rows = (
         variable_query.with_entities(
             CalculateRequestVariable.variable_name,
@@ -251,13 +279,21 @@ def _unique_variable_keys(
     ]
 
 
-def _apply_time_filters(query, filters: CalculateAnalyticsQuery):
+def _apply_filters(query, filters: CalculateAnalyticsQuery):
     model = query.column_descriptions[0]["entity"]
     created_at = model.created_at
     if filters.start_time:
         query = query.filter(created_at >= filters.start_time)
     if filters.end_time:
         query = query.filter(created_at <= filters.end_time)
+    if filters.requested_version:
+        query = query.filter(
+            model.requested_version == filters.requested_version
+        )
+    if filters.resolved_channel:
+        query = query.filter(
+            model.resolved_channel == filters.resolved_channel
+        )
     return query
 
 
@@ -271,6 +307,8 @@ def _request_to_dict(
         "api_version": calculate_request.api_version,
         "country_id": calculate_request.country_id,
         "model_version": calculate_request.model_version,
+        "requested_version": calculate_request.requested_version,
+        "resolved_channel": calculate_request.resolved_channel,
         "endpoint": calculate_request.endpoint,
         "method": calculate_request.method,
         "response_status_code": calculate_request.response_status_code,

policyengine_household_api/modal_release/gateway.py

@@ -13,6 +13,10 @@
     empty_manifest,
     validate_manifest,
 )
+from policyengine_household_api.utils.modal_routing_metadata import (
+    MODAL_ROUTING_PAYLOAD_KEY,
+    modal_routing_payload,
+)
 
 
 VERSIONED_ENDPOINTS = {"calculate", "calculate_demo"}
@@ -100,7 +104,7 @@ def route_request(path: str) -> Response:
             )
             return route_to_worker_function(
                 resolved_app.app_name,
-                _request_payload(path, body),
+                _request_payload(path, body, resolved_app),
             )
         except GatewayResolutionError as e:
             return _json_error(str(e), 400)
@@ -212,13 +216,21 @@ def _country_and_endpoint(path: str) -> tuple[str | None, str | None]:
     return parts[0], parts[1]
 
 
-def _request_payload(path: str, body: bytes) -> dict[str, Any]:
+def _request_payload(
+    path: str,
+    body: bytes,
+    resolved_app: ResolvedApp,
+) -> dict[str, Any]:
     return {
         "method": request.method,
         "path": path,
         "query_string": request.query_string.decode("utf-8"),
         "headers": _forward_headers(),
         "body": body,
+        MODAL_ROUTING_PAYLOAD_KEY: modal_routing_payload(
+            requested_version=resolved_app.requested_version,
+            resolved_channel=resolved_app.channel,
+        ),
     }