Skip to content

Commit 9f3b207

Browse files
felixwang9817achals
felixwang9817
authored and
achals
committed
Format and lint
Signed-off-by: Felix Wang <wangfelix98@gmail.com> Signed-off-by: Achal Shah <achals@gmail.com>
1 parent 467beb2 commit 9f3b207

9 files changed

Lines changed: 81 additions & 80 deletions

File tree

‎sdk/python/feast/errors.py‎

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
from typing import List, Set, Any
1+
from typing import Any, List, Set
22

33
from colorama import Fore, Style
44

@@ -113,11 +113,10 @@ def __init__(self):
113113
"Failed to connect to the Go subprocess (locally running feature server)"
114114
)
115115

116+
116117
class GoServerError(Exception):
117118
def __init__(self, error_message):
118-
super().__init__(
119-
f"Go server raises exception: {error_message}"
120-
)
119+
super().__init__(f"Go server raises exception: {error_message}")
121120

122121

123122
class FeastModuleImportError(Exception):
@@ -347,4 +346,6 @@ def __init__(self, query: str):
347346

348347
class InvalidFeaturesParameterType(Exception):
349348
def __init__(self, features: Any):
350-
super().__init__(f"Invalid `features` parameter type {type(features)}. Expected one of List[str] and FeatureService.")
349+
super().__init__(
350+
f"Invalid `features` parameter type {type(features)}. Expected one of List[str] and FeatureService."
351+
)

‎sdk/python/feast/feature_store.py‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,6 @@
3434

3535
import pandas as pd
3636
from colorama import Fore, Style
37-
from feast.flags_helper import enable_go_feature_server
3837
from google.protobuf.timestamp_pb2 import Timestamp
3938
from tqdm import tqdm
4039

@@ -60,6 +59,8 @@
6059
DUMMY_ENTITY_VAL,
6160
FeatureView,
6261
)
62+
from feast.flags_helper import enable_go_feature_server
63+
from feast.go_server import GoServer
6364
from feast.inference import (
6465
update_data_sources_with_inferred_event_timestamp_col,
6566
update_entities_with_inferred_types_from_feature_views,
@@ -85,7 +86,6 @@
8586
from feast.usage import log_exceptions, log_exceptions_and_usage, set_usage_attribute
8687
from feast.value_type import ValueType
8788
from feast.version import get_version
88-
from feast.go_server import GoServer
8989

9090
warnings.simplefilter("once", DeprecationWarning)
9191

@@ -1167,7 +1167,9 @@ def get_online_features(
11671167
# Lazily start the go server on the first request
11681168
if self._go_server is None:
11691169
self._go_server = GoServer(str(self.repo_path.absolute()), self.config)
1170-
return self._go_server.get_online_features(features, columnar, full_feature_names)
1170+
return self._go_server.get_online_features(
1171+
features, columnar, full_feature_names
1172+
)
11711173

11721174
return self._get_online_features(
11731175
features=features,

‎sdk/python/feast/go_server.py‎

Lines changed: 39 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -1,20 +1,24 @@
1+
import atexit
12
import os
2-
import signal
33
import platform
44
import socket
55
import subprocess
6-
import atexit
76
import time
8-
from typing import Union, List, Dict, Any
7+
from typing import Any, Dict, List, Union
8+
99
import grpc
10+
11+
import feast
1012
from feast import errors
11-
from feast.type_map import python_values_to_proto_values
1213
from feast.feature_service import FeatureService
13-
from feast.repo_config import RepoConfig
1414
from feast.online_response import OnlineResponse
15-
from feast.protos.feast.serving.ServingService_pb2 import GetOnlineFeaturesRequest, GetFeastServingInfoRequest
15+
from feast.protos.feast.serving.ServingService_pb2 import (
16+
GetFeastServingInfoRequest,
17+
GetOnlineFeaturesRequest,
18+
)
1619
from feast.protos.feast.serving.ServingService_pb2_grpc import ServingServiceStub
17-
import feast
20+
from feast.repo_config import RepoConfig
21+
from feast.type_map import python_values_to_proto_values
1822

1923

2024
class GoServer:
@@ -30,10 +34,10 @@ def __init__(self, repo_path: str, config: RepoConfig):
3034
self._connect()
3135

3236
def get_online_features(
33-
self,
34-
features: Union[List[str], FeatureService],
35-
entities: Dict[str, List[Any]],
36-
full_feature_names: bool = False,
37+
self,
38+
features: Union[List[str], FeatureService],
39+
entities: Dict[str, List[Any]],
40+
full_feature_names: bool = False,
3741
) -> OnlineResponse:
3842

3943
if not self.grpcServerStarted:
@@ -49,7 +53,7 @@ def get_online_features(
4953

5054
for key, values in entities.items():
5155
request.entities[key].val.extend(python_values_to_proto_values(values))
52-
56+
5357
try:
5458
response = self.client.GetOnlineFeatures(request=request)
5559
except grpc.RpcError as rpc_error:
@@ -69,7 +73,9 @@ def get_online_features(
6973
parsed_error_message = error_message.split(": ")[1].split("; ")
7074
collided_feature_refs = parsed_error_message[0].split(", ")
7175
full_feature_names = parsed_error_message[1] == "true"
72-
raise errors.FeatureNameCollisionError(collided_feature_refs, full_feature_names)
76+
raise errors.FeatureNameCollisionError(
77+
collided_feature_refs, full_feature_names
78+
)
7379
elif error_message.startswith(self.ValueError_STRING):
7480
parsed_error_message = error_message.split(": ")[1]
7581
raise ValueError(parsed_error_message)
@@ -88,34 +94,39 @@ def _connect(self):
8894
# pass a random unused port to go subprocess, so that there's no conflicts
8995
# if multiple Python processes start Go subprocess on the same host
9096
"FEAST_GRPC_PORT": unused_port,
91-
**os.environ
97+
**os.environ,
9298
}
9399
cwd = feast.__path__[0]
94100

95101
if "dev" in feast.__version__:
96-
self.process = subprocess.Popen(["go", "run", "github.com/feast-dev/feast/go/server"],
97-
cwd=cwd, env=env,
98-
stdin=subprocess.PIPE )
102+
self.process = subprocess.Popen(
103+
["go", "run", "github.com/feast-dev/feast/go/server"],
104+
cwd=cwd,
105+
env=env,
106+
stdin=subprocess.PIPE,
107+
)
99108
else:
100109
goos = platform.system().lower()
101110
goarch = "amd64" if platform.machine() == "x86_64" else "arm64"
102111
executable = feast.__path__[0] + f"/binaries/go_server_{goos}_{goarch}"
103-
self.process = subprocess.Popen([executable], cwd=cwd, env=env, stdin=subprocess.PIPE)
112+
self.process = subprocess.Popen(
113+
[executable], cwd=cwd, env=env, stdin=subprocess.PIPE
114+
)
104115

105116
# Make sure the subprocess is terminated when the parent process dies
106117
# Note: this doesn't handle cases where the parent process is abruptly killed (e.g. with SIGKILL)
107-
atexit.register(lambda: self.stop() )
118+
atexit.register(lambda: self.stop())
108119
self.start_grpc_server()
109120
self.pipeClosed = False
110121

111122
def start_grpc_server(self):
112123
if self.grpcServerStarted:
113124
return
114125
# Try connecting to the go server using a gPRC client
115-
126+
116127
for i in range(5):
117128
try:
118-
self.process.stdin.write(b'startGrpc\n')
129+
self.process.stdin.write(b"startGrpc\n")
119130
self.process.stdin.flush()
120131
self.grpcServerStarted = True
121132
break
@@ -146,7 +157,9 @@ def start_http_server(self, host: str, port: int):
146157
return
147158
for i in range(10):
148159
try:
149-
self.process.stdin.write(bytes(f"startHttp {host}:{port}\n", encoding='utf8'))
160+
self.process.stdin.write(
161+
bytes(f"startHttp {host}:{port}\n", encoding="utf8")
162+
)
150163
self.process.stdin.flush()
151164
self.httpServerStarted = True
152165
break
@@ -163,7 +176,7 @@ def stop(self):
163176
# Otherwise, let go subprocess clean up and shut down itself
164177
if not self.pipeClosed:
165178
try:
166-
self.process.stdin.write(bytes(f"stop\n", encoding='utf8'))
179+
self.process.stdin.write(bytes("stop\n", encoding="utf8"))
167180
self.process.stdin.flush()
168181
# TODO (Ly): Review: We don't close stdin here
169182
# since if the call succeeds go process closes
@@ -172,11 +185,12 @@ def stop(self):
172185
except subprocess.CalledProcessError as error:
173186
self.process.terminate()
174187
raise errors.GoSubprocessConnectionFailed() from error
175-
188+
176189
self.grpcServerStarted = False
177190
self.httpServerStarted = False
178191
self.pipeClosed = True
179-
192+
193+
180194
def _get_unused_port() -> str:
181195
sock = socket.socket()
182196
# binding port 0 means os will choose an unused port for us

‎sdk/python/feast/infra/online_stores/connector.py‎

Lines changed: 7 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -12,33 +12,20 @@
1212
# See the License for the specific language governing permissions and
1313
# limitations under the License.
1414

15-
from pydantic import StrictStr
16-
from pydantic.typing import Literal
1715
from datetime import datetime
18-
from typing import (
19-
Any,
20-
ByteString,
21-
Callable,
22-
Dict,
23-
List,
24-
Optional,
25-
Sequence,
26-
Tuple,
27-
Union,
28-
)
29-
30-
from google.protobuf.timestamp_pb2 import Timestamp
16+
from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple
17+
3118
from pydantic import StrictStr
3219
from pydantic.typing import Literal
3320

3421
from feast import Entity, FeatureView, RepoConfig
35-
from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto
36-
from feast.protos.feast.types.Value_pb2 import Value as ValueProto
3722
from feast.infra.infra_object import InfraObject
23+
from feast.infra.online_stores.online_store import OnlineStore
3824
from feast.protos.feast.core.Registry_pb2 import Registry as RegistryProto
39-
25+
from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto
26+
from feast.protos.feast.types.Value_pb2 import Value as ValueProto
4027
from feast.repo_config import FeastConfigBaseModel
41-
from feast.infra.online_stores.online_store import OnlineStore
28+
4229

4330
class ConnectorOnlineStoreConfig(FeastConfigBaseModel):
4431
"""Online store config for Connector store"""
@@ -56,6 +43,7 @@ class ConnectorOnlineStore(OnlineStore):
5643
OnlineStore is an object used for all interaction between Feast and the service used for online storage of
5744
features.
5845
"""
46+
5947
def online_write_batch(
6048
self,
6149
config: RepoConfig,
@@ -121,4 +109,3 @@ def teardown(
121109
entities: Sequence[Entity],
122110
):
123111
pass
124-

‎sdk/python/tests/conftest.py‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -73,7 +73,10 @@ def pytest_addoption(parser):
7373
"--universal", action="store_true", default=False, help="Run universal tests",
7474
)
7575
parser.addoption(
76-
"--noodfv", action="store_true", default=False, help="Run tests without on demand transforms",
76+
"--noodfv",
77+
action="store_true",
78+
default=False,
79+
help="Run tests without on demand transforms",
7780
)
7881

7982

@@ -106,7 +109,7 @@ def pytest_collection_modifyitems(config, items: List[Item]):
106109
items.clear()
107110
for t in universal_tests:
108111
items.append(t)
109-
112+
110113
noodfv_tests = [t for t in items if "noodfv" in t.keywords]
111114
if should_run_without_odfv:
112115
items.clear()

‎sdk/python/tests/integration/feature_repos/repo_configuration.py‎

Lines changed: 3 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -103,7 +103,7 @@
103103
),
104104
]
105105
)
106-
106+
107107
full_repo_configs_module = os.environ.get(FULL_REPO_CONFIGS_MODULE_ENV_NAME)
108108
if full_repo_configs_module is not None:
109109
try:
@@ -115,10 +115,7 @@
115115
) from e
116116
elif os.getenv("FEAST_IS_GO_SERVER_TEST", "True") == "True":
117117
FULL_REPO_CONFIGS = [
118-
IntegrationTestRepoConfig(
119-
online_store=REDIS_CONFIG,
120-
go_feature_server=True,
121-
),
118+
IntegrationTestRepoConfig(online_store=REDIS_CONFIG, go_feature_server=True,),
122119
IntegrationTestRepoConfig(
123120
provider="gcp",
124121
offline_store_creator=BigQueryDataSourceCreator,
@@ -307,6 +304,7 @@ def construct_universal_feature_views(
307304
field_mapping=create_field_mapping_feature_view(data_sources.field_mapping),
308305
)
309306

307+
310308
def construct_universal_feature_views_without_odfv(
311309
data_sources: Dict[str, DataSource],
312310
) -> Dict[str, FeatureView]:

‎sdk/python/tests/integration/feature_repos/universal/feature_views.py‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,7 @@ def global_feature_view(
4343
# name="conv_rate_input", schema={"val_to_add": ValueType.INT32}
4444
# )
4545

46+
4647
def conv_rate_plus_100(features_df: pd.DataFrame) -> pd.DataFrame:
4748
df = pd.DataFrame()
4849
df["conv_rate_plus_100"] = features_df["conv_rate"] + 100
@@ -72,6 +73,7 @@ def conv_rate_plus_100_feature_view(
7273
udf=conv_rate_plus_100,
7374
)
7475

76+
7577
def conv_rate_plus_100_feature_view_without_odfv(
7678
inputs: Dict[str, Union[RequestDataSource, FeatureView]],
7779
infer_features: bool = False,

‎sdk/python/tests/integration/online_store/test_online_retrieval.py‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -403,5 +403,3 @@ def test_online_to_df():
403403
]
404404
expected_df = pd.DataFrame({k: reversed(v) for (k, v) in df_dict.items()})
405405
assert_frame_equal(result_df[ordered_column], expected_df)
406-
407-

0 commit comments

Comments
 (0)