Skip to content

Commit e733c89

Browse files
felixwang9817felixwang9817
committed
Add tests
Signed-off-by: Felix Wang <[email protected]>
1 parent b0dc405 commit e733c89

7 files changed

Lines changed: 397 additions & 162 deletions

‎sdk/python/feast/feature_store.py‎

Lines changed: 19 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,13 @@
4343
from feast import feature_server, flags_helper, ui_server, utils
4444
from feast.base_feature_view import BaseFeatureView
4545
from feast.batch_feature_view import BatchFeatureView
46-
from feast.data_source import DataSource, PushMode
46+
from feast.data_source import (
47+
DataSource,
48+
KafkaSource,
49+
KinesisSource,
50+
PushMode,
51+
PushSource,
52+
)
4753
from feast.diff.infra_diff import InfraDiff, diff_infra_protos
4854
from feast.diff.registry_diff import RegistryDiff, apply_diff_to_registry, diff_between
4955
from feast.dqm.errors import ValidationFailed
@@ -827,6 +833,18 @@ def apply(
827833
ob for ob in objects if isinstance(ob, ValidationReference)
828834
]
829835

836+
batch_sources_to_add: List[DataSource] = []
837+
for data_source in data_sources_set_to_update:
838+
if (
839+
isinstance(data_source, PushSource)
840+
or isinstance(data_source, KafkaSource)
841+
or isinstance(data_source, KinesisSource)
842+
):
843+
assert data_source.batch_source
844+
batch_sources_to_add.append(data_source.batch_source)
845+
for batch_source in batch_sources_to_add:
846+
data_sources_set_to_update.add(batch_source)
847+
830848
for fv in itertools.chain(views_to_update, sfvs_to_update):
831849
data_sources_set_to_update.add(fv.batch_source)
832850
if fv.stream_source:

‎sdk/python/tests/example_repos/example_feature_repo_with_inline_batch_source.py‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
from datetime import timedelta
2+
3+
from feast import Entity, FeatureView, Field, FileSource
4+
from feast.types import Float32, Int32, Int64
5+
6+
driver = Entity(
7+
name="driver_id",
8+
description="driver id",
9+
)
10+
11+
driver_hourly_stats_view = FeatureView(
12+
name="driver_hourly_stats",
13+
entities=[driver],
14+
ttl=timedelta(days=1),
15+
schema=[
16+
Field(name="conv_rate", dtype=Float32),
17+
Field(name="acc_rate", dtype=Float32),
18+
Field(name="avg_daily_trips", dtype=Int64),
19+
Field(name="driver_id", dtype=Int32),
20+
],
21+
online=True,
22+
source=FileSource(
23+
path="data/driver_stats.parquet", # Fake path
24+
timestamp_field="event_timestamp",
25+
created_timestamp_column="created",
26+
),
27+
tags={},
28+
)

‎sdk/python/tests/example_repos/example_feature_repo_with_inline_stream_source.py‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,37 @@
1+
from datetime import timedelta
2+
3+
from feast import Entity, FeatureView, Field, FileSource, KafkaSource
4+
from feast.data_format import AvroFormat
5+
from feast.types import Float32, Int32, Int64
6+
7+
driver = Entity(
8+
name="driver_id",
9+
description="driver id",
10+
)
11+
12+
driver_hourly_stats_view = FeatureView(
13+
name="driver_hourly_stats",
14+
entities=[driver],
15+
ttl=timedelta(days=1),
16+
schema=[
17+
Field(name="conv_rate", dtype=Float32),
18+
Field(name="acc_rate", dtype=Float32),
19+
Field(name="avg_daily_trips", dtype=Int64),
20+
Field(name="driver_id", dtype=Int32),
21+
],
22+
online=True,
23+
source=KafkaSource(
24+
name="kafka",
25+
timestamp_field="event_timestamp",
26+
kafka_bootstrap_servers="",
27+
message_format=AvroFormat(""),
28+
topic="topic",
29+
batch_source=FileSource(
30+
path="data/driver_stats.parquet", # Fake path
31+
timestamp_field="event_timestamp",
32+
created_timestamp_column="created",
33+
),
34+
watermark_delay_threshold=timedelta(days=1),
35+
),
36+
tags={},
37+
)

‎sdk/python/tests/example_repos/example_feature_repo_with_stream_source.py‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,18 @@
1+
from datetime import timedelta
2+
3+
from feast import FileSource, KafkaSource
4+
from feast.data_format import AvroFormat
5+
6+
stream_source = KafkaSource(
7+
name="kafka",
8+
timestamp_field="event_timestamp",
9+
kafka_bootstrap_servers="",
10+
message_format=AvroFormat(""),
11+
topic="topic",
12+
batch_source=FileSource(
13+
path="data/driver_stats.parquet", # Fake path
14+
timestamp_field="event_timestamp",
15+
created_timestamp_column="created",
16+
),
17+
watermark_delay_threshold=timedelta(days=1),
18+
)

0 commit comments

Comments
 (0)