33import warnings
44from datetime import timedelta
55from types import MethodType
6- from typing import Dict , List , Optional , Type , Union
6+ from typing import Dict , List , Optional , Union
77
88import dill
99from google .protobuf .duration_pb2 import Duration
1010
1111from feast import utils
1212from feast .aggregation import Aggregation
13- from feast .data_source import DataSource , KafkaSource
13+ from feast .data_source import DataSource , KafkaSource , PushSource
1414from feast .entity import Entity
1515from feast .feature_view import FeatureView
1616from feast .field import Field
@@ -106,7 +106,9 @@ def __init__(
106106 self .mode = mode or ""
107107 self .timestamp_field = timestamp_field or ""
108108 self .udf = udf
109- _batch_source = source .batch_source if source .batch_source else None
109+ _batch_source = None
110+ if isinstance (source , KafkaSource ) or isinstance (source , PushSource ):
111+ _batch_source = source .batch_source if source .batch_source else None
110112 _ttl = ttl
111113 if not _ttl :
112114 _ttl = timedelta (days = 0 )
@@ -124,17 +126,20 @@ def __init__(
124126 source = source ,
125127 )
126128
127- def __eq__ (self , other ) -> bool :
129+ def __eq__ (self , other ):
128130 if not isinstance (other , StreamFeatureView ):
129131 raise TypeError ("Comparisons should only involve StreamFeatureViews" )
130132
131133 if not super ().__eq__ (other ):
132134 return False
133-
135+ if not self .udf :
136+ return not other .udf
137+ if not other .udf :
138+ return False
134139 if (
135140 self .mode != other .mode
136141 or self .timestamp_field != other .timestamp_field
137- or ( self .udf and self . udf . __code__ .co_code != other .udf .__code__ .co_code )
142+ or self .udf . __code__ .co_code != other .udf .__code__ .co_code
138143 or self .aggregations != other .aggregations
139144 ):
140145 return False
@@ -144,7 +149,7 @@ def __eq__(self, other) -> bool:
144149 def __hash__ (self ) -> int :
145150 return super ().__hash__ ()
146151
147- def to_proto (self ) -> StreamFeatureViewProto :
152+ def to_proto (self ):
148153 meta = StreamFeatureViewMetaProto (materialization_intervals = [])
149154 if self .created_timestamp :
150155 meta .created_timestamp .FromDatetime (self .created_timestamp )
@@ -270,10 +275,6 @@ def from_proto(cls, sfv_proto):
270275
271276 return sfv_feature_view
272277
273- @property
274- def proto_class (self ) -> Type [StreamFeatureViewProto ]:
275- return StreamFeatureViewProto
276-
277278 def __copy__ (self ):
278279 fv = StreamFeatureView (
279280 name = self .name ,
0 commit comments