GeezerVison is showing github.com — feat: Add registry list feature views by updated since (#6092) by nquinn408 · Pull Request #6093 · feast-dev/feast
feat: Add registry list feature views by updated since (#6092)
#6093
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement . We’ll occasionally send you account related emails.
Already on GitHub?
Sign in
to your account
Show file tree
Hide file tree
Changes from all commits
Commits
File filter
Filter by extension
.md
(1)
.proto
(1)
.py
(14)
.pyi
(1)
All 4 file types selected
Viewed files
Conversations
Failed to load comments.
Loading
Uh oh!
There was an error while loading. Please reload this page.
Jump to
Jump to file
Failed to load files.
Loading
Uh oh!
There was an error while loading. Please reload this page.
Diff view
Hide whitespace
Diff view
Hide whitespace
There are no files selected for viewing
Original file line number
Diff line number
Diff line change
Expand Up
@@ -214,6 +214,7 @@ Most endpoints support these common query parameters:
- `feature` (optional): Filter feature views by feature name
- `feature_service` (optional): Filter feature views by feature service name
- `data_source` (optional): Filter feature views by data source name
- `updated_since` (optional): Only return feature views updated at or after this ISO-8601 UTC timestamp (e.g. `2024-01-01T00:00:00Z`)
- `page` (optional): Page number for pagination
- `limit` (optional): Number of items per page
- `sort_by` (optional): Field to sort by
Expand All
@@ -223,27 +224,31 @@ Most endpoints support these common query parameters:
# Basic list
curl -H "Authorization: Bearer <token>" \
"http://localhost:6572/api/v1/feature_views?project=my_project"
# With pagination and relationships
curl -H "Authorization: Bearer <token>" \
"http://localhost:6572/api/v1/feature_views?project=my_project&include_relationships=true&page=1&limit=5&sort_by=name"
# Filter by entity
curl -H "Authorization: Bearer <token>" \
"http://localhost:6572/api/v1/feature_views?project=my_project&entity=user"
# Filter by feature
curl -H "Authorization: Bearer <token>" \
"http://localhost:6572/api/v1/feature_views?project=my_project&feature=age"
# Filter by data source
curl -H "Authorization: Bearer <token>" \
"http://localhost:6572/api/v1/feature_views?project=my_project&data_source=user_profile_source"
# Filter by feature service
curl -H "Authorization: Bearer <token>" \
"http://localhost:6572/api/v1/feature_views?project=my_project&feature_service=user_service"
# Filter by last-updated timestamp
curl -H "Authorization: Bearer <token>" \
"http://localhost:6572/api/v1/feature_views?project=my_project&updated_since=2024-06-01T00:00:00Z"
# Multiple filters combined
curl -H "Authorization: Bearer <token>" \
"http://localhost:6572/api/v1/feature_views?project=my_project&entity=user&feature=age"
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -291,6 +291,7 @@ message ListAllFeatureViewsRequest {
string data_source = 7;
PaginationParams pagination = 8;
SortingParams sorting = 9;
google.protobuf.Timestamp updated_since = 10;
}
message ListAllFeatureViewsResponse {
Expand Down
Original file line number
Diff line number
Diff line change
@@ -1,8 +1,10 @@
import logging
from datetime import timezone
from typing import Dict, List, Optional
from fastapi import APIRouter, Depends, Query
from fastapi import APIRouter, Depends, HTTPException, Query
from fastapi.responses import JSONResponse
from google.protobuf import timestamp_pb2
from google.protobuf.duration_pb2 import Duration
from pydantic import BaseModel
Expand Down
Expand Up
@@ -266,10 +268,34 @@ def list_all_feature_views(
data_source: str = Query(
None, description="Filter feature views by data source name"
),
updated_since: Optional[str] = Query(
None,
description="Only return feature views updated at or after this ISO-8601 UTC timestamp (e.g. 2024-01-01T00:00:00Z)",
),
tags: Dict[str, str] = Depends(parse_tags),
pagination_params: dict = Depends(get_pagination_params),
sorting_params: dict = Depends(get_sorting_params),
):
updated_since_proto = None
if updated_since is not None:
from datetime import datetime
try:
dt = datetime.fromisoformat(updated_since.replace("Z", "+00:00"))
except ValueError:
raise HTTPException(
status_code=400,
detail=(
f"Invalid 'updated_since' value '{updated_since}'; expected an "
"ISO-8601 timestamp (e.g. 2024-01-01T00:00:00Z)."
),
)
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
ts = timestamp_pb2.Timestamp()
ts.FromDatetime(dt.astimezone(timezone.utc))
updated_since_proto = ts
req = RegistryServer_pb2.ListAllFeatureViewsRequest(
project=project,
allow_cache=allow_cache,
Expand All
@@ -280,6 +306,7 @@ def list_all_feature_views(
data_source=data_source,
pagination=create_grpc_pagination_params(pagination_params),
sorting=create_grpc_sorting_params(sorting_params),
updated_since=updated_since_proto,
)
response = grpc_call(grpc_handler.ListAllFeatureViews, req)
any_feature_views = response.get("featureViews", [])
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -567,6 +567,7 @@ def list_all_feature_views(
allow_cache: bool = False,
tags: Optional[dict[str, str]] = None,
skip_udf: bool = False,
updated_since: Optional[datetime] = None,
) -> List[BaseFeatureView]:
"""
Retrieve a list of feature views of all types from the registry
Expand All
@@ -576,6 +577,7 @@ def list_all_feature_views(
project: Filter feature views based on project name
tags: Filter by tags
skip_udf: Skip deserializing UDFs (for metadata-only operations)
updated_since: Only return feature views updated at or after this timestamp
Returns:
List of feature views
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -3,7 +3,7 @@
import threading
import warnings
from abc import abstractmethod
from datetime import timedelta
from datetime import datetime, timedelta
from threading import Lock
from typing import Any, Dict, List, Optional
Expand All
@@ -23,7 +23,7 @@
from feast.protos.feast.core.Registry_pb2 import Registry as RegistryProto
from feast.saved_dataset import SavedDataset, ValidationReference
from feast.stream_feature_view import StreamFeatureView
from feast.utils import _utc_now
from feast.utils import _utc_now, to_naive_utc
logger = logging.getLogger(__name__)
Expand Down
Expand Up
@@ -122,7 +122,11 @@ def get_any_feature_view(
@abstractmethod
def _list_all_feature_views(
self, project: str, tags: Optional[dict[str, str]], **kwargs: Any
self,
project: str,
tags: Optional[dict[str, str]],
updated_since: Optional[datetime] = None,
**kwargs: Any,
) -> List[BaseFeatureView]:
pass
Expand All
@@ -132,13 +136,27 @@ def list_all_feature_views(
allow_cache: bool = False,
tags: Optional[dict[str, str]] = None,
skip_udf: bool = False,
updated_since: Optional[datetime] = None,
) -> List[BaseFeatureView]:
if allow_cache:
self._refresh_cached_registry_if_necessary()
return proto_registry_utils.list_all_feature_views(
feature_views = proto_registry_utils.list_all_feature_views(
self.cached_registry_proto, project, tags, skip_udf=skip_udf
)
return self._list_all_feature_views(project, tags, skip_udf=skip_udf)
if updated_since is not None:
# last_updated_timestamp from proto is offset-naive UTC; normalise for comparison
cutoff = to_naive_utc(updated_since)
feature_views = [
fv
for fv in feature_views
if fv.last_updated_timestamp is not None
and fv.last_updated_timestamp >= cutoff
]
else:
feature_views = self._list_all_feature_views(
project, tags, updated_since=updated_since, skip_udf=skip_udf
)
return feature_views
@abstractmethod
def _get_feature_view(self, name: str, project: str) -> FeatureView:
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -58,7 +58,7 @@
from feast.repo_contents import RepoContents
from feast.saved_dataset import SavedDataset, ValidationReference
from feast.stream_feature_view import StreamFeatureView
from feast.utils import _utc_now
from feast.utils import _utc_now, to_naive_utc
from feast.version_utils import (
generate_version_id,
parse_version,
Expand Down
Expand Up
@@ -1083,13 +1083,24 @@ def list_all_feature_views(
allow_cache: bool = False,
tags: Optional[dict[str, str]] = None,
skip_udf: bool = False,
updated_since: Optional[datetime] = None,
) -> List[BaseFeatureView]:
registry_proto = self._get_registry_proto(
project=project, allow_cache=allow_cache
)
return proto_registry_utils.list_all_feature_views(
feature_views = proto_registry_utils.list_all_feature_views(
registry_proto, project, tags, skip_udf=skip_udf
)
if updated_since is not None:
# last_updated_timestamp from proto is offset-naive UTC; normalise for comparison
cutoff = to_naive_utc(updated_since)
feature_views = [
fv
for fv in feature_views
if fv.last_updated_timestamp is not None
and fv.last_updated_timestamp >= cutoff
]
return feature_views
Comment thread
nquinn408 marked this conversation as resolved.
Resolved
Uh oh!
There was an error while loading. Please reload this page.
def get_any_feature_view(
self, name: str, project: str, allow_cache: bool = False
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -403,9 +403,18 @@ def list_all_feature_views(
allow_cache: bool = False,
tags: Optional[dict[str, str]] = None,
skip_udf: bool = False,
updated_since: Optional[datetime] = None,
) -> List[BaseFeatureView]:
updated_since_proto = None
if updated_since is not None:
ts = Timestamp()
ts.FromDatetime(updated_since)
updated_since_proto = ts
request = RegistryServer_pb2.ListAllFeatureViewsRequest(
project=project, allow_cache=allow_cache, tags=tags
project=project,
allow_cache=allow_cache,
tags=tags,
updated_since=updated_since_proto,
)
response: RegistryServer_pb2.ListAllFeatureViewsResponse = (
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -62,7 +62,7 @@
from feast.repo_config import RegistryConfig
from feast.saved_dataset import SavedDataset, ValidationReference
from feast.stream_feature_view import StreamFeatureView
from feast.utils import _utc_now, has_all_tags
from feast.utils import _utc_now, has_all_tags, to_naive_utc
logger = logging.getLogger(__name__)
Expand Down
Expand Up
@@ -656,14 +656,24 @@ def list_all_feature_views(
allow_cache: bool = False,
tags: Optional[dict[str, str]] = None,
skip_udf: bool = False,
updated_since: Optional[datetime] = None,
) -> List[BaseFeatureView]:
if allow_cache:
registry_proto = self._refresh_cached_registry_if_necessary()
return proto_registry_utils.list_all_feature_views(
feature_views = proto_registry_utils.list_all_feature_views(
registry_proto, project, tags, skip_udf=skip_udf
)
if updated_since is not None:
cutoff = to_naive_utc(updated_since)
feature_views = [
fv
for fv in feature_views
if fv.last_updated_timestamp is not None
and fv.last_updated_timestamp >= cutoff
]
return feature_views
return (
feature_views = (
cast(
list[BaseFeatureView],
self.list_feature_views(project, allow_cache, tags, skip_udf=skip_udf),
Expand All
@@ -686,6 +696,17 @@ def list_all_feature_views(
)
)
if updated_since is not None:
cutoff = to_naive_utc(updated_since)
feature_views = [
fv
for fv in feature_views
if fv.last_updated_timestamp is not None
and fv.last_updated_timestamp >= cutoff
]
return feature_views
Comment thread
nquinn408 marked this conversation as resolved.
Resolved
Uh oh!
There was an error while loading. Please reload this page.
def get_infra(self, project: str, allow_cache: bool = False) -> Infra:
infra_object = self._get_object(
"MANAGED_INFRA",
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -485,26 +485,36 @@ def _get_any_feature_view(self, name: str, project: str) -> BaseFeatureView:
return fv
def _list_all_feature_views(
self, project: str, tags: Optional[dict[str, str]], **kwargs
self,
project: str,
tags: Optional[dict[str, str]],
updated_since: Optional[datetime] = None,
**kwargs,
) -> List[BaseFeatureView]:
return (
cast(
list[BaseFeatureView],
self._list_feature_views(project=project, tags=tags, **kwargs),
self._list_feature_views(
project=project, tags=tags, updated_since=updated_since, **kwargs
),
)
+ cast(
list[BaseFeatureView],
self._list_stream_feature_views(project=project, tags=tags, **kwargs),
self._list_stream_feature_views(
project=project, tags=tags, updated_since=updated_since, **kwargs
),
)
+ cast(
list[BaseFeatureView],
self._list_on_demand_feature_views(
project=project, tags=tags, **kwargs
project=project, tags=tags, updated_since=updated_since, **kwargs
),
)
+ cast(
list[BaseFeatureView],
self._list_label_views(project=project, tags=tags, **kwargs),
self._list_label_views(
project=project, tags=tags, updated_since=updated_since, **kwargs
),
)
)
Expand Down
Expand Up
@@ -1597,6 +1607,7 @@ def _list_objects(
tags: Optional[dict[str, str]] = None,
proto_only: bool = False,
skip_udf: bool = False,
updated_since: Optional[datetime] = None,
):
"""
Args:
Expand All
@@ -1618,6 +1629,17 @@ def _list_objects(
with self.read_engine.begin() as conn:
stmt = select(table).where(table.c.project_id == project)
if updated_since is not None:
# Ensure naive datetimes are treated as UTC, consistent with
# the Python-side filters that compare against offset-naive UTC
# last_updated_timestamp values from protobuf.
if updated_since.tzinfo is None:
updated_since_utc = updated_since.replace(tzinfo=timezone.utc)
else:
updated_since_utc = updated_since.astimezone(timezone.utc)
stmt = stmt.where(
table.c.last_updated_timestamp >= int(updated_since_utc.timestamp())
)
Comment thread
nquinn408 marked this conversation as resolved.
Resolved
Uh oh!
There was an error while loading. Please reload this page.
rows = conn.execute(stmt).all()
if rows:
objects = []
Expand Down
Large diffs are not rendered by default.
Oops, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
Original file line number
Diff line number
Diff line change
Expand Up
@@ -704,6 +704,7 @@ class ListAllFeatureViewsRequest(google.protobuf.message.Message):
DATA_SOURCE_FIELD_NUMBER: builtins.int
PAGINATION_FIELD_NUMBER: builtins.int
SORTING_FIELD_NUMBER: builtins.int
UPDATED_SINCE_FIELD_NUMBER: builtins.int
project: builtins.str
allow_cache: builtins.bool
@property
Expand All
@@ -716,6 +717,8 @@ class ListAllFeatureViewsRequest(google.protobuf.message.Message):
def pagination(self) -> global___PaginationParams: ...
@property
def sorting(self) -> global___SortingParams: ...
@property
def updated_since(self) -> google.protobuf.timestamp_pb2.Timestamp: ...
def __init__(
self,
*,
Expand All
@@ -728,9 +731,10 @@ class ListAllFeatureViewsRequest(google.protobuf.message.Message):
data_source: builtins.str = ...,
pagination: global___PaginationParams | None = ...,
sorting: global___SortingParams | None = ...,
updated_since: google.protobuf.timestamp_pb2.Timestamp | None = ...,
) -> None: ...
def HasField(self, field_name: typing_extensions.Literal["pagination", b"pagination", "sorting", b"sorting"]) -> builtins.bool: ...
def ClearField(self, field_name: typing_extensions.Literal["allow_cache", b"allow_cache", "data_source", b"data_source", "entity", b"entity", "feature", b"feature", "feature_service", b"feature_service", "pagination", b"pagination", "project", b"project", "sorting", b"sorting", "tags", b"tags"]) -> None: ...
def HasField(self, field_name: typing_extensions.Literal["pagination", b"pagination", "sorting", b"sorting", "updated_since", b"updated_since" ]) -> builtins.bool: ...
def ClearField(self, field_name: typing_extensions.Literal["allow_cache", b"allow_cache", "data_source", b"data_source", "entity", b"entity", "feature", b"feature", "feature_service", b"feature_service", "pagination", b"pagination", "project", b"project", "sorting", b"sorting", "tags", b"tags", "updated_since", b"updated_since" ]) -> None: ...
global___ListAllFeatureViewsRequest = ListAllFeatureViewsRequest
Expand Down
Loading
Oops, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
Loading
Oops, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.