feat: Search within date range in pipeline run API (#130)

### TL;DR

Implemented search within date range in Pipeline Runs.

### What changed?

#### Functional

- API `GET /api/pipeline_runs/`
  - Search by date range in `filter_query`. Example query (not URL encoded for example):
    ```
      /api?filter_query={
        "and": [{
          "time_range": {
            "key": "system/pipeline_run.date.created_at",
            "start_time": "2024-01-01T00:00:00Z",
            "end_time": "2024-02-01T00:00:00Z"
          }
        }]
      }
    ```
  - Valid time ranges:
    - Start time only
    - End time only
    - (Start + End) time
    

#### Other

- Added timezone handling to convert aware datetimes to naive UTC for database compatibility (i.e. date range search will convert API timezone to UTC because DB datetime is only UTC)

### How to test?

```
uv run pytest tests/test_api_server_sql.py tests/test_filter_query_sql.py
```

The implementation includes comprehensive tests covering:
- Basic time range filtering with start and end times
- Start-only time ranges (no end time)
- Boundary conditions (inclusive start, exclusive end)
- Timezone offset handling (e.g., `+05:30` converted to UTC)
- Negation with `NOT` operator
- Combination with annotation-based filters
- Nested logical operations (`AND`/`OR`)
- Multiple time ranges in the same query

### Why make this change?

- This change enables users to filter pipeline runs by creation date range.
- The implementation properly handles timezone conversions to ensure consistent behavior across different client timezones while maintaining compatibility with the database's naive UTC storage format.

Author: yuechao-qin

cloud_pipelines_backend/filter_query_models.py

@@ -99,9 +99,13 @@ def key(self) -> str:
         return self.value_equals.key
 
 
-class TimeRangePredicate(_BaseModel):
+class TimeRangePredicate(KeyPredicateBase):
     time_range: TimeRange
 
+    @property
+    def key(self) -> str:
+        return self.time_range.key
+
 
 LeafPredicate = (
     KeyExistsPredicate

cloud_pipelines_backend/filter_query_sql.py

@@ -1,4 +1,5 @@
 import base64
+import datetime
 import json
 import enum
 from typing import Any, Final
@@ -16,6 +17,7 @@
 class PipelineRunAnnotationSystemKey(enum.StrEnum):
     CREATED_BY = f"{_PIPELINE_RUN_KEY_PREFIX}created_by"
     PIPELINE_NAME = f"{_PIPELINE_RUN_KEY_PREFIX}name"
+    CREATED_AT = f"{_PIPELINE_RUN_KEY_PREFIX}date.created_at"
 
 
 SYSTEM_KEY_SUPPORTED_PREDICATES: dict[PipelineRunAnnotationSystemKey, set[type]] = {
@@ -30,6 +32,9 @@ class PipelineRunAnnotationSystemKey(enum.StrEnum):
         filter_query_models.ValueContainsPredicate,
         filter_query_models.ValueInPredicate,
     },
+    PipelineRunAnnotationSystemKey.CREATED_AT: {
+        filter_query_models.TimeRangePredicate,
+    },
 }
 
 # ---------------------------------------------------------------------------
@@ -301,8 +306,9 @@ def _predicate_to_clause(
             return _value_contains_to_clause(predicate=predicate)
         case filter_query_models.ValueInPredicate():
             return _value_in_to_clause(predicate=predicate)
+        case filter_query_models.TimeRangePredicate():
+            return _time_range_to_clause(predicate=predicate)
         case _:
-            # TODO: TimeRangePredicate -- not supported currently, will be supported in the future.
             raise NotImplementedError(
                 f"Predicate type {type(predicate).__name__} is not yet implemented."
             )
@@ -363,3 +369,59 @@ def _value_in_to_clause(
             bts.PipelineRunAnnotation.value.in_(predicate.value_in.values),
         ],
     )
+
+
+# ---------------------------------------------------------------------------
+# Column-based predicates (bypass annotation table)
+# ---------------------------------------------------------------------------
+
+
+def _time_range_to_clause(
+    *, predicate: filter_query_models.TimeRangePredicate
+) -> sql.ColumnElement:
+    """Build a WHERE clause for pipeline_run.created_at from a time range.
+
+    Pydantic's AwareDatetime preserves the original timezone offset, so we
+    must normalize to naive UTC before comparing against the DB column.
+
+    The DB stores "naive UTC" datetimes -- the values represent UTC but carry
+    no timezone label. For example, the DB stores '2024-01-01 02:30:00', not
+    '2024-01-01 02:30:00+00:00'. The UtcDateTime type decorator (in
+    backend_types_sql.py) strips tzinfo on write and re-attaches UTC on read.
+
+    Conversion pipeline for input '2024-01-01T08:00:00+05:30':
+
+      API request (JSON string)
+        '2024-01-01T08:00:00+05:30'
+              |
+              v
+      Pydantic AwareDatetime (preserves offset)
+        datetime(2024, 1, 1, 8, 0, 0, tzinfo=+05:30)
+              |
+              v  .astimezone(utc) -- converts 08:00 - 05:30 = 02:30
+      UTC-aware datetime
+        datetime(2024, 1, 1, 2, 30, 0, tzinfo=UTC)
+              |
+              v  .replace(tzinfo=None) -- strips timezone label
+      Naive datetime
+        datetime(2024, 1, 1, 2, 30, 0)
+              |
+              v  SQLAlchemy literal_binds -- adds microsecond precision
+      SQL string
+        '2024-01-01 02:30:00.000000'  <-- matches DB storage format
+    """
+    tr = predicate.time_range
+    if tr.key != PipelineRunAnnotationSystemKey.CREATED_AT:
+        raise errors.ApiValidationError(
+            "time_range only supports key "
+            f"{PipelineRunAnnotationSystemKey.CREATED_AT!r}, got {tr.key!r}"
+        )
+    # Convert aware datetimes to naive UTC to match DB storage format.
+    clauses: list[sql.ColumnElement] = []
+    if tr.start_time is not None:
+        start_utc = tr.start_time.astimezone(datetime.timezone.utc).replace(tzinfo=None)
+        clauses.append(bts.PipelineRun.created_at >= start_utc)
+    if tr.end_time is not None:
+        end_utc = tr.end_time.astimezone(datetime.timezone.utc).replace(tzinfo=None)
+        clauses.append(bts.PipelineRun.created_at < end_utc)
+    return sql.and_(*clauses)