Validate event search parameters in FastAPI query models

server/polar/event/endpoints.py

@@ -4,7 +4,7 @@
 
 from fastapi import Depends, Query
 from fastapi.exceptions import RequestValidationError
-from pydantic import UUID4, ValidationError
+from pydantic import UUID4
 
 from polar.exceptions import (
     NotPermitted,
@@ -19,7 +19,6 @@
 )
 from polar.kit.time_queries import is_under_limits
 from polar.kit.versioning import version
-from polar.meter.filter import Filter
 from polar.models import Event
 from polar.openapi import APITag
 from polar.postgres import (
@@ -61,43 +60,6 @@
 router = APIRouter(prefix="/events", tags=["events", APITag.public])
 
 
-def _parse_filter(filter: str | None) -> Filter | None:
-    # FastAPI does not support complex schemas in query parameters.
-    if filter is None:
-        return None
-    try:
-        return Filter.model_validate_json(filter)
-    except ValidationError as e:
-        raise RequestValidationError(e.errors()) from e
-
-
-def _validate_query(query: str | None, organization_id: Sequence[UUID4] | None) -> None:
-    if query is None:
-        return
-    if organization_id is None:
-        raise RequestValidationError(
-            [
-                {
-                    "type": "value_error",
-                    "msg": "Query is only supported when organization_id is provided.",
-                    "loc": ("query", "query"),
-                    "input": query,
-                }
-            ]
-        )
-    if len(query) < 3:
-        raise RequestValidationError(
-            [
-                {
-                    "type": "query",
-                    "msg": "Query must be at least 3 characters.",
-                    "loc": ("query", "query"),
-                    "input": query,
-                }
-            ]
-        )
-
-
 EventNotFound = {"description": "Event not found.", "model": ResourceNotFound.schema()}
 
 
@@ -135,13 +97,10 @@ async def list(
 ) -> ListResource[EventSchema] | ListResourceWithCursorPagination[EventSchema]:
     """List events."""
 
-    parsed_filter = _parse_filter(filters.filter)
-    _validate_query(filters.query, filters.organization_id)
-
     result = await event_service.list(
         session,
         auth_subject,
-        filter=parsed_filter,
+        filter=filters.filter,
         start_timestamp=filters.start_timestamp,
         end_timestamp=filters.end_timestamp,
         organization_id=filters.organization_id,
@@ -195,13 +154,11 @@ async def get_statistics_by_type(
     Accepts the same filters as the event list, except the event name, so every
     event type gets a count under the other filters.
     """
-    parsed_filter = _parse_filter(filters.filter)
-    _validate_query(filters.query, filters.organization_id)
 
     items = await event_service.list_event_type_stats(
         session,
         auth_subject,
-        filter=parsed_filter,
+        filter=filters.filter,
         start_timestamp=filters.start_timestamp,
         end_timestamp=filters.end_timestamp,
         organization_id=filters.organization_id,
@@ -244,15 +201,13 @@ async def get_histogram(
                 }
             ]
         )
-    parsed_filter = _parse_filter(filters.filter)
-    _validate_query(filters.query, filters.organization_id)
 
     return await event_service.get_histogram(
         session,
         auth_subject,
         start_timestamp=filters.start_timestamp,
         end_timestamp=filters.end_timestamp,
-        filter=parsed_filter,
+        filter=filters.filter,
         organization_id=filters.organization_id,
         customer_id=filters.customer_id,
         external_customer_id=filters.external_customer_id,
@@ -394,17 +349,14 @@ async def list_statistics_timeseries(
             ]
         )
 
-    parsed_filter = _parse_filter(filters.filter)
-    _validate_query(filters.query, filters.organization_id)
-
     return await event_service.list_statistics_timeseries(
         session,
         auth_subject,
         start_date=filters.start_date,
         end_date=filters.end_date,
         timezone=ZoneInfo(filters.timezone),
         interval=filters.interval,
-        filter=parsed_filter,
+        filter=filters.filter,
         organization_id=filters.organization_id,
         customer_id=filters.customer_id,
         external_customer_id=filters.external_customer_id,

server/polar/event/schemas.py

@@ -2,7 +2,7 @@
 from collections.abc import Sequence
 from datetime import UTC, date, datetime
 from decimal import Decimal
-from typing import Annotated, Literal, NotRequired
+from typing import Annotated, Literal, NotRequired, Self
 
 from annotated_types import Len, Predicate
 from fastapi import Path
@@ -13,8 +13,12 @@
     AwareDatetime,
     Discriminator,
     Field,
+    Json,
+    WithJsonSchema,
+    model_validator,
 )
 from pydantic.type_adapter import TypeAdapter
+from pydantic_core import PydanticCustomError
 from pydantic_extra_types.timezone_name import TimeZoneName
 from typing_extensions import TypedDict
 
@@ -70,6 +74,7 @@
 )
 from polar.kit.time_queries import TimeInterval
 from polar.kit.versioning import QuerySchema
+from polar.meter.filter import Filter
 from polar.meter.schemas import MeterID
 from polar.models.event import EventSource
 from polar.organization.schemas import OrganizationID
@@ -918,8 +923,32 @@ def _validate_aggregate_field(value: str) -> bool:
 ]
 
 
-class EventListParams(QuerySchema):
-    filter: str | None = Field(
+# Keep the JSON-encoded query parameter a string in generated clients.
+EventQueryFilter = Annotated[Json[Filter], WithJsonSchema({"type": "string"})]
+
+
+class EventSearchParams(QuerySchema):
+    organization_id: MultipleQueryFilter[OrganizationID] | None = Field(
+        None, title="OrganizationID Filter", description="Filter by organization ID."
+    )
+    query: str | None = Field(
+        None, min_length=3, title="Query", description="Query to filter events."
+    )
+
+    @model_validator(mode="after")
+    def validate_search_query(self) -> Self:
+        if self.query is None:
+            return self
+        if self.organization_id is None:
+            raise PydanticCustomError(
+                "value_error",
+                "Query is only supported when organization_id is provided.",
+            )
+        return self
+
+
+class EventListParams(EventSearchParams):
+    filter: EventQueryFilter | None = Field(
         None,
         description=(
             "Filter events following filter clauses. "
@@ -932,9 +961,6 @@ class EventListParams(QuerySchema):
     end_timestamp: AwareDatetime | None = Field(
         None, description="Filter events before this timestamp."
     )
-    organization_id: MultipleQueryFilter[OrganizationID] | None = Field(
-        None, title="OrganizationID Filter", description="Filter by organization ID."
-    )
     customer_id: MultipleQueryFilter[CustomerID] | None = Field(
         None, title="CustomerID Filter", description="Filter by customer ID."
     )
@@ -952,9 +978,6 @@ class EventListParams(QuerySchema):
     source: MultipleQueryFilter[EventSource] | None = Field(
         None, title="Source Filter", description="Filter by event source."
     )
-    query: str | None = Field(
-        None, title="Query", description="Query to filter events."
-    )
     parent_id: EventID | None = Field(
         None,
         description="When combined with depth, use this event as the anchor instead of root events.",
@@ -967,8 +990,8 @@ class EventListParams(QuerySchema):
     )
 
 
-class EventTypeStatisticsParams(QuerySchema):
-    filter: str | None = Field(
+class EventTypeStatisticsParams(EventSearchParams):
+    filter: EventQueryFilter | None = Field(
         None,
         description=(
             "Filter events following filter clauses. "
@@ -981,9 +1004,6 @@ class EventTypeStatisticsParams(QuerySchema):
     end_timestamp: AwareDatetime | None = Field(
         None, description="Filter events before this timestamp."
     )
-    organization_id: MultipleQueryFilter[OrganizationID] | None = Field(
-        None, title="OrganizationID Filter", description="Filter by organization ID."
-    )
     customer_id: MultipleQueryFilter[CustomerID] | None = Field(
         None, title="CustomerID Filter", description="Filter by customer ID."
     )
@@ -995,9 +1015,6 @@ class EventTypeStatisticsParams(QuerySchema):
     source: MultipleQueryFilter[EventSource] | None = Field(
         None, title="Source Filter", description="Filter by event source."
     )
-    query: str | None = Field(
-        None, title="Query", description="Query to filter events."
-    )
 
 
 class EventHistogramParams(EventTypeStatisticsParams):
@@ -1070,7 +1087,7 @@ class EventStatisticsByVarianceParams(EventStatisticsParams):
     limit: int = Field(default=100, le=1000)
 
 
-class EventListStatisticsTimeseriesParams(QuerySchema):
+class EventListStatisticsTimeseriesParams(EventSearchParams):
     start_date: date = Field(
         ...,
         description="Start date.",
@@ -1081,16 +1098,13 @@ class EventListStatisticsTimeseriesParams(QuerySchema):
         description="Timezone to use for the dates. Default is UTC.",
     )
     interval: TimeInterval = Field(..., description="Interval between two dates.")
-    filter: str | None = Field(
+    filter: EventQueryFilter | None = Field(
         None,
         description=(
             "Filter events following filter clauses. "
             "JSON string following the same schema a meter filter clause. "
         ),
     )
-    organization_id: MultipleQueryFilter[OrganizationID] | None = Field(
-        None, title="OrganizationID Filter", description="Filter by organization ID."
-    )
     customer_id: MultipleQueryFilter[CustomerID] | None = Field(
         None, title="CustomerID Filter", description="Filter by customer ID."
     )
@@ -1113,9 +1127,6 @@ class EventListStatisticsTimeseriesParams(QuerySchema):
         title="Event Type ID Filter",
         description="Filter by event type ID.",
     )
-    query: str | None = Field(
-        None, title="Query", description="Query to filter events."
-    )
     aggregate_fields: Sequence[AggregateField] = Field(
         default=["_cost.amount"],
         description="Metadata field paths to aggregate (e.g., '_cost.amount', 'duration_ns'). Use dot notation for nested fields.",