Client Bugfixes (#722)

.github/workflows/e2e.yml

@@ -466,7 +466,7 @@ jobs:
       - name: Run CLI for Get and List Functions
         run: |
             kubectl get secret featureform-ca-secret -o=custom-columns=':.data.tls\.crt'| base64 -d > tls.crt
-            featureform apply client/examples/quickstart.py --host localhost:8000 --cert tls.crt
+            featureform apply --no-wait client/examples/quickstart.py --host localhost:8000 --cert tls.crt
             featureform get label fraudulent --host localhost:8000 --cert tls.crt
             featureform get label fraudulent quickstart --host localhost:8000 --cert tls.crt
             featureform list labels --host localhost:8000 --cert tls.crt

Makefile

@@ -369,7 +369,7 @@ test_e2e: update_python					## Runs End-to-End tests on minikube
 	while ! echo exit | nc localhost 7000; do sleep 10; done
 	while ! echo exit | nc localhost 2379; do sleep 10; done
 
-	featureform apply client/examples/quickstart.py --host localhost:8000 --cert tls.crt
+	featureform apply --no-wait client/examples/quickstart.py --host localhost:8000 --cert tls.crt
 	pytest client/tests/e2e.py
 	pytest -m 'hosted' client/tests/test_serving_model.py
 	pytest -m 'hosted' client/tests/test_getting_model.py

client/src/featureform/format.py

@@ -37,8 +37,7 @@ def format_pg(s=""):
 
 
 def display_statuses(stub, resources):
-
-    from featureform import Feature, TrainingSet, Label, Source, Provider
+    from featureform.resources import Feature, OnDemandFeature, TrainingSet, Label, Source, Provider
 
     @dataclass
     class Status:
@@ -50,7 +49,7 @@ class Status:
 
     def get_statuses() -> List[Status]:
         statuses = []
-        resources_to_check = {Feature, TrainingSet, Label, Source, Provider}
+        resources_to_check = {Feature, OnDemandFeature, TrainingSet, Label, Source, Provider}
         filtered_resources = filter(lambda r: type(r) in resources_to_check, resources)
         for r in filtered_resources:
             if r.name == "local-mode":
@@ -95,7 +94,7 @@ def is_finished(statuses):
                 header_style="bold",
                 box=None,
             )
-            table.add_column("Resource Type")
+            table.add_column("Resource Type", width=25)
             table.add_column("Name (Variant)", width=50, no_wrap=True)
             table.add_column("Status", width=10)
             table.add_column("Error", style="red")
@@ -118,8 +117,8 @@ def is_finished(statuses):
 
                 style = status_to_color[status_text]
                 table.add_row(
-                    Text(resource_type, style=style),
-                    Text(f"{name} ({status.variant})", style=style),
+                    Text(resource_type),
+                    Text(f"{name} ({status.variant})"),
                     Text(status_text, style=status_to_color[status_text]),
                     Text(error, style="red")
                 )

client/src/featureform/register.py

@@ -20,7 +20,7 @@
     AzureFileStoreConfig, OnlineBlobConfig, K8sConfig, S3StoreConfig, GCSFileStoreConfig, User, Location, Source, PrimaryData, SQLTable, \
     SQLTransformation, DFTransformation, Entity, Feature, Label, ResourceColumnMapping, TrainingSet, ProviderReference, \
     EntityReference, SourceReference, ExecutorCredentials, ResourceRedefinedError, ResourceStatus, Transformation, \
-    K8sArgs, AWSCredentials, GCPCredentials, HDFSConfig, K8sResourceSpecs, FilePrefix, OnDemandFeatureDecorator
+    K8sArgs, AWSCredentials, GCPCredentials, HDFSConfig, K8sResourceSpecs, FilePrefix, OnDemandFeature
 
 from .proto import metadata_pb2_grpc as ff_grpc
 from .search_local import search_local
@@ -626,6 +626,8 @@ def __getitem__(self, columns: List[str]):
         col_len = len(columns)
         if col_len < 2:
             raise Exception(f"Expected 2 columns, but found {col_len}. Missing entity and/or source columns")
+        elif col_len > 3:
+            raise Exception(f"Found unrecognized columns {', '.join(columns[3:])}. Expected 2 required columns and an optional 3rd timestamp column")
         return (self.registrar, self.name_variant(), columns)
 
     def name_variant(self):
@@ -722,10 +724,12 @@ def __init__(self, fn, registrar, provider, decorator_register_resources_method,
         self.name_variant = decorator_name_variant_method.__get__(self)
         pass
 
-    def __getitem__(self, columns):
+    def __getitem__(self, columns: List[str]):
         col_len = len(columns)
         if col_len < 2:
             raise Exception(f"Expected 2 columns, but found {col_len}. Missing entity and/or source columns")
+        elif col_len > 3:
+            raise Exception(f"Found unrecognized columns {', '.join(columns[3:])}. Expected 2 required columns and an optional 3rd timestamp column")
         return (self.registrar, self.name_variant(), columns)
 
     def __call__(self, *args, **kwds):
@@ -905,6 +909,8 @@ def __getitem__(self, columns: List[str]):
         col_len = len(columns)
         if col_len < 2:
             raise Exception(f"Expected 2 columns, but found {col_len}. Missing entity and/or source columns")
+        elif col_len > 3:
+            raise Exception(f"Found unrecognized columns {', '.join(columns[3:])}. Expected 2 required columns and an optional 3rd timestamp column")
         return (self.registrar(), self.id(), columns)
 
     def register_resources(
@@ -1273,7 +1279,33 @@ def get_snowflake(self, name):
         """
         get = ProviderReference(name=name, provider_type="snowflake", obj=None)
         self.__resources.append(get)
-        fakeConfig = SnowflakeConfig(account="", database="", organization="", username="", password="", schema="")
+        fakeConfig = SnowflakeConfig(account="ff_fake", database="ff_fake", organization="ff_fake", username="ff_fake", password="ff_fake", schema="ff_fake")
+        fakeProvider = Provider(name=name, function="OFFLINE", description="", team="", config=fakeConfig)
+        return OfflineSQLProvider(self, fakeProvider)
+
+    def get_snowflake_legacy(self, name: str):
+        """Get a Snowflake provider. The returned object can be used to register additional resources.
+
+        **Examples**:
+        ``` py
+        snowflake = ff.get_snowflake_legacy("snowflake-quickstart")
+        transactions = snowflake.register_table(
+            name="transactions",
+            variant="kaggle",
+            description="Fraud Dataset From Kaggle",
+            table="Transactions",  # This is the table's name in Postgres
+        )
+        ```
+        Args:
+            name (str): Name of Snowflake provider to be retrieved
+
+        Returns:
+            snowflake_legacy (OfflineSQLProvider): Provider
+        """
+        get = ProviderReference(name=name, provider_type="snowflake", obj=None)
+        self.__resources.append(get)
+
+        fakeConfig = SnowflakeConfig(account_locator="ff_fake", database="ff_fake", username="ff_fake", password="ff_fake", schema="ff_fake", warehouse="ff_fake", role="ff_fake")
         fakeProvider = Provider(name=name, function="OFFLINE", description="", team="", config=fakeConfig)
         return OfflineSQLProvider(self, fakeProvider)
 
@@ -2542,7 +2574,7 @@ def df_transformation(self,
     def ondemand_feature(self, 
                           fn=None, *,
                           tags: List[str] = None,
-                          properties: dict = {},
+                          properties: dict = None,
                           variant: str = "default",
                           name: str = "",
                           owner: Union[str, UserRegistrar] = "",
@@ -2559,7 +2591,7 @@ def ondemand_feature(self,
             properties (dict): Optional grouping mechanism for resources
 
         Returns:
-            decorator (OnDemandFeatureDecorator): decorator
+            decorator (OnDemandFeature): decorator
 
         **Examples**
         ```python
@@ -2574,13 +2606,13 @@ def avg_user_transactions():
         if owner == "":
             owner = self.must_get_default_owner()
 
-        decorator = OnDemandFeatureDecorator(
+        decorator = OnDemandFeature(
             name=name,
             variant=variant,
             owner=owner,
             description=description,
-            tags=tags,
-            properties=properties,
+            tags=tags or [],
+            properties=properties or {},
         )
         self.__resources.append(decorator)
 
@@ -2934,6 +2966,8 @@ def apply(self, asynchronous=True):
             resources = resource_state.sorted_list()
             display_statuses(self._stub, resources)
 
+        self.clear_state()
+
 
     def get_user(self, name, local=False):
         """Get a user. Prints out name of user, and all resources associated with the user.
@@ -4102,7 +4136,12 @@ def __init__(
         self.variant = variant
         self.owner = owner
         self.inference_store = inference_store
-        self.timestamp_column = timestamp_column
+        if not timestamp_column and len(columns) == 3:
+            self.timestamp_column = columns[2]
+        elif timestamp_column and len(columns) == 3:
+            raise Exception("Timestamp column specified twice.")
+        else:
+            self.timestamp_column = timestamp_column
         self.description = description
         self.schedule = schedule
         self.tags = tags
@@ -4289,6 +4328,7 @@ class User:
 get_postgres = global_registrar.get_postgres
 get_mongodb = global_registrar.get_mongodb
 get_snowflake = global_registrar.get_snowflake
+get_snowflake_legacy = global_registrar.get_snowflake_legacy
 get_redshift = global_registrar.get_redshift
 get_bigquery = global_registrar.get_bigquery
 get_spark = global_registrar.get_spark

client/src/featureform/resources.py

@@ -873,11 +873,7 @@ def type() -> str:
 
     def get(self, stub):
         name_variant = pb.NameVariant(name=self.name, variant=self.variant)
-        source = None
-        for x in stub.GetSourceVariants(iter([name_variant])):
-            source = x
-            break
-
+        source = next(stub.GetSourceVariants(iter([name_variant])))
         definition = self._get_source_definition(source)
 
         return Source(
@@ -1083,10 +1079,7 @@ def type() -> str:
 
     def get(self, stub) -> "Feature":
         name_variant = pb.NameVariant(name=self.name, variant=self.variant)
-        feature = None
-        for x in stub.GetFeatureVariants(iter([name_variant])):
-            feature = x
-            break
+        feature = next(stub.GetFeatureVariants(iter([name_variant])))
 
         return Feature(
             name=feature.name,
@@ -1177,22 +1170,17 @@ def __eq__(self, other):
                 return False
         return True
 
-
-class OnDemandFeatureDecorator:
-    def __init__(self,
-                 owner: str,
-                 tags: List[str] = None,
-                 properties: dict = {},
-                 variant: str = "default",
-                 name: str = "",
-                 description: str = ""):
-        self.name = name
-        self.variant = variant
-        self.owner = owner
-        self.description = description
-        self.tags = tags
-        self.properties = properties
-        self.status = "READY"
+@typechecked
+@dataclass
+class OnDemandFeature:
+    owner: str
+    tags: List[str] = field(default_factory=list)
+    properties: dict = field(default_factory=dict)
+    variant: str = "default"
+    name: str = ""
+    description: str = ""
+    status: str = "READY"
+    error: Optional[str] = None
 
     def __call__(self, fn):
         if self.description == "" and fn.__doc__ is not None:
@@ -1265,6 +1253,21 @@ def _write_feature_variant_and_mode(self, db) -> None:
             ComputationMode.CLIENT_COMPUTED.value,
             is_on_demand,
         )
+
+    def get(self, stub) -> "OnDemandFeature":
+        name_variant = pb.NameVariant(name=self.name, variant=self.variant)
+        ondemand_feature = next(stub.GetFeatureVariants(iter([name_variant])))
+
+        return OnDemandFeature(
+            name=ondemand_feature.name,
+            variant=ondemand_feature.variant,
+            owner=ondemand_feature.owner,
+            description=ondemand_feature.description,
+            tags=list(ondemand_feature.tags.tag),
+            properties={k: v for k, v in ondemand_feature.properties.property.items()},
+            status=ondemand_feature.status.Status._enum_type.values[ondemand_feature.status.status].name,
+            error=ondemand_feature.status.error_message,
+        )
 
     def get_status(self):
         return ResourceStatus(self.status)
@@ -1311,10 +1314,7 @@ def type() -> str:
 
     def get(self, stub) -> "Label":
         name_variant = pb.NameVariant(name=self.name, variant=self.variant)
-        label = None
-        for x in stub.GetLabelVariants(iter([name_variant])):
-            label = x
-            break
+        label = next(stub.GetLabelVariants(iter([name_variant])))
 
         return Label(
             name=label.name,
@@ -1524,10 +1524,7 @@ def type() -> str:
 
     def get(self, stub):
         name_variant = pb.NameVariant(name=self.name, variant=self.variant)
-        ts = None
-        for x in stub.GetTrainingSetVariants(iter([name_variant])):
-            ts = x
-            break
+        ts = next(stub.GetTrainingSetVariants(iter([name_variant])))
 
         return TrainingSet(
             name=ts.name,
@@ -1700,7 +1697,7 @@ def __eq__(self, other):
 
 
 Resource = Union[PrimaryData, Provider, Entity, User, Feature, Label,
-                 TrainingSet, Source, Schedule, ProviderReference, SourceReference, EntityReference, Model, OnDemandFeatureDecorator]
+                 TrainingSet, Source, Schedule, ProviderReference, SourceReference, EntityReference, Model, OnDemandFeature]
 
 
 class ResourceRedefinedError(Exception):

client/tests/register_test.py

@@ -107,6 +107,16 @@ def test_invalid_model_registration():
     with pytest.raises(TypeError, match="missing 1 required positional argument: 'name'"):
         model = ff.register_model()
 
+@pytest.mark.parametrize(
+    "provider_name,func",
+    [
+        ("snowflake", ff.get_snowflake),
+        ("snowflake_legacy", ff.get_snowflake_legacy)
+    ]
+)
+def test_get_snowflake_functions(provider_name, func):
+    offlineSQLProvider = func(provider_name)
+    assert offlineSQLProvider.name() == provider_name
 
 def del_rw(action, name, exc):
     os.chmod(name, stat.S_IWRITE)