import importlib

import json

import logging

from typing import TYPE_CHECKING, Any, List, Optional, Set

from colorama import Fore, Style

from fastapi import status as HttpStatusCode

if TYPE_CHECKING:

from grpc import StatusCode as GrpcStatusCode

from feast.field import Field

logger = logging.getLogger(__name__)

class FeastError(Exception):

pass

def grpc_status_code(self) -> "GrpcStatusCode":

from grpc import StatusCode as GrpcStatusCode

return GrpcStatusCode.INTERNAL

def http_status_code(self) -> int:

return HttpStatusCode.HTTP_500_INTERNAL_SERVER_ERROR

def __str__(self) -> str:

if hasattr(self, "__overridden_message__"):

return str(getattr(self, "__overridden_message__"))

return super().__str__()

def __repr__(self) -> str:

if hasattr(self, "__overridden_message__"):

return f"{type(self).__name__}('{getattr(self, '__overridden_message__')}')"

return super().__repr__()

def to_error_detail(self) -> str:

"""

Returns a JSON representation of the error for serialization purposes.

Returns:

str: a string representation of a JSON document including `module`, `class` and `message` fields.

"""

m = {

"module": f"{type(self).__module__}",

"class": f"{type(self).__name__}",

"message": f"{str(self)}",

}

return json.dumps(m)

@staticmethod

def from_error_detail(detail: str) -> Optional["FeastError"]:

try:

m = json.loads(detail)

if all(f in m for f in ["module", "class", "message"]):

module_name = m["module"]

class_name = m["class"]

message = m["message"]

module = importlib.import_module(module_name)

class_reference = getattr(module, class_name)

instance = class_reference.__new__(class_reference)

setattr(instance, "__overridden_message__", message)

return instance

except Exception as e:

logger.warning(f"Invalid error detail: {detail}: {e}")

return None

class DataSourceNotFoundException(FeastError):

def __init__(self, path):

super().__init__(

f"Unable to find table at '{path}'. Please check that table exists."

)

class DataSourceNoNameException(FeastError):

def __init__(self):

super().__init__(

"Unable to infer a name for this data source. Either table or name must be specified."

)

class DataSourceRepeatNamesException(FeastError):

def __init__(self, ds_name: str):

super().__init__(

f"Multiple data sources share the same case-insensitive name {ds_name}."

)

class FeastObjectNotFoundException(FeastError):

pass

def grpc_status_code(self) -> "GrpcStatusCode":

from grpc import StatusCode as GrpcStatusCode

return GrpcStatusCode.NOT_FOUND

def http_status_code(self) -> int:

return HttpStatusCode.HTTP_404_NOT_FOUND

class EntityNotFoundException(FeastObjectNotFoundException):

def __init__(self, name, project=None):

if project:

super().__init__(f"Entity {name} does not exist in project {project}")

else:

super().__init__(f"Entity {name} does not exist")

class FeatureServiceNotFoundException(FeastObjectNotFoundException):

def __init__(self, name, project=None):

if project:

super().__init__(

f"Feature service {name} does not exist in project {project}"

)

else:

super().__init__(f"Feature service {name} does not exist")

class FeatureViewNotFoundException(FeastObjectNotFoundException):

def __init__(self, name, project=None):

if project:

super().__init__(f"Feature view {name} does not exist in project {project}")

else:

super().__init__(f"Feature view {name} does not exist")

class FeatureViewVersionNotFound(FeastObjectNotFoundException):

def __init__(self, name, version, project=None):

if project:

super().__init__(

f"Version {version} of feature view {name} does not exist in project {project}"

)

else:

super().__init__(f"Version {version} of feature view {name} does not exist")

class VersionedOnlineReadNotSupported(FeastError):

def __init__(self, store_name: str, version: int):

super().__init__(

f"Versioned feature reads (@v{version}) are not yet supported by {store_name}. "

f"Currently only SQLite, PostgreSQL, MySQL, FAISS, Redis, and DynamoDB support version-qualified feature references. "

)

class FeatureViewPinConflict(FeastError):

def __init__(self, name, version):

super().__init__(

f"Cannot pin feature view '{name}' to {version} because the definition has also been modified. "

f"To pin to an older version, only change the 'version' parameter — do not modify other fields. "

f"To apply a new definition, use version='latest' or omit the version parameter."

)

class ConcurrentVersionConflict(FeastError):

def __init__(self, msg: str):

super().__init__(msg)

class OnDemandFeatureViewNotFoundException(FeastObjectNotFoundException):

def __init__(self, name, project=None):

if project:

super().__init__(

f"On demand feature view {name} does not exist in project {project}"

)

else:

super().__init__(f"On demand feature view {name} does not exist")

class RequestDataNotFoundInEntityDfException(FeastObjectNotFoundException):

def __init__(self, feature_name, feature_view_name):

super().__init__(

f"Feature {feature_name} not found in the entity dataframe, but required by feature view {feature_view_name}"

)

class RequestDataNotFoundInEntityRowsException(FeastObjectNotFoundException):

def __init__(self, feature_names):

super().__init__(

f"Required request data source features {feature_names} not found in the entity rows, but required by feature views"

)

class DataSourceObjectNotFoundException(FeastObjectNotFoundException):

def __init__(self, name, project=None):

if project:

super().__init__(f"Data source {name} does not exist in project {project}")

else:

super().__init__(f"Data source {name} does not exist")

class S3RegistryBucketNotExist(FeastObjectNotFoundException):

def __init__(self, bucket):

super().__init__(f"S3 bucket {bucket} for the Feast registry does not exist")

class S3RegistryBucketForbiddenAccess(FeastObjectNotFoundException):

def __init__(self, bucket):

super().__init__(f"S3 bucket {bucket} for the Feast registry can't be accessed")

class SavedDatasetNotFound(FeastObjectNotFoundException):

def __init__(self, name: str, project: str):

super().__init__(f"Saved dataset {name} does not exist in project {project}")

class ValidationReferenceNotFound(FeastObjectNotFoundException):

def __init__(self, name: str, project: str):

super().__init__(

f"Validation reference {name} does not exist in project {project}"

)

class FeastProviderLoginError(FeastError):

"""Error class that indicates a user has not authenticated with their provider."""

class FeastProviderNotImplementedError(FeastError):

def __init__(self, provider_name):

super().__init__(f"Provider '{provider_name}' is not implemented")

class FeastRegistryNotSetError(FeastError):

def __init__(self):

super().__init__("Registry is not set, but is required")

class FeastFeatureServerTypeInvalidError(FeastError):

def __init__(self, feature_server_type: str):

super().__init__(

f"Feature server type was set to {feature_server_type}, but this type is invalid"

)

class FeastRegistryTypeInvalidError(FeastError):

def __init__(self, registry_type: str):

super().__init__(

f"Feature server type was set to {registry_type}, but this type is invalid"

)

class FeastModuleImportError(FeastError):

def __init__(self, module_name: str, class_name: str):

super().__init__(

f"Could not import module '{module_name}' while attempting to load class '{class_name}'"

)

class FeastClassImportError(FeastError):

def __init__(self, module_name: str, class_name: str):

super().__init__(

f"Could not import class '{class_name}' from module '{module_name}'"

)

class FeastExtrasDependencyImportError(FeastError):

def __init__(self, extras_type: str, nested_error: str):

message = (

nested_error

+ "\n"

+ f"You may need run {Style.BRIGHT + Fore.GREEN}pip install 'feast[{extras_type}]'{Style.RESET_ALL}"

)

super().__init__(message)

class FeastOfflineStoreUnsupportedDataSource(FeastError):

def __init__(self, offline_store_name: str, data_source_name: str):

super().__init__(

f"Offline Store '{offline_store_name}' does not support data source '{data_source_name}'"

)

class FeatureNameCollisionError(FeastError):

def __init__(self, feature_refs_collisions: List[str], full_feature_names: bool):

if full_feature_names:

collisions = [ref.replace(":", "__") for ref in feature_refs_collisions]

error_message = (

"To resolve this collision, please ensure that the feature views or their own features "

"have different names. If you're intentionally joining the same feature view twice on "

"different sets of entities, please rename one of the feature views with '.with_name'."

)

else:

collisions = [ref.split(":")[1] for ref in feature_refs_collisions]

error_message = (

"To resolve this collision, either use the full feature name by setting "

"'full_feature_names=True', or ensure that the features in question have different names."

)

feature_names = ", ".join(set(collisions))

super().__init__(

f"Duplicate features named {feature_names} found.\n{error_message}"

)

class SpecifiedFeaturesNotPresentError(FeastError):

def __init__(

self,

specified_features: List[Field],

inferred_features: List[Field],

feature_view_name: str,

):

super().__init__(

f"Explicitly specified features {specified_features} not found in inferred list of features "

f"{inferred_features} for '{feature_view_name}'"

)

class SavedDatasetLocationAlreadyExists(FeastError):

def __init__(self, location: str):

super().__init__(f"Saved dataset location {location} already exists.")

class FeastOfflineStoreInvalidName(FeastError):

def __init__(self, offline_store_class_name: str):

super().__init__(

f"Offline Store Class '{offline_store_class_name}' should end with the string `OfflineStore`.'"

)

class FeastOnlineStoreInvalidName(FeastError):

def __init__(self, online_store_class_name: str):

super().__init__(

f"Online Store Class '{online_store_class_name}' should end with the string `OnlineStore`.'"

)

class FeastInvalidAuthConfigClass(FeastError):

def __init__(self, auth_config_class_name: str):

super().__init__(

f"Auth Config Class '{auth_config_class_name}' should end with the string `AuthConfig`.'"

)

class FeastInvalidBaseClass(FeastError):

def __init__(self, class_name: str, class_type: str):

super().__init__(

f"Class '{class_name}' should have `{class_type}` as a base class."

)

class FeastOnlineStoreUnsupportedDataSource(FeastError):

def __init__(self, online_store_name: str, data_source_name: str):

super().__init__(

f"Online Store '{online_store_name}' does not support data source '{data_source_name}'"

)

class FeastEntityDFMissingColumnsError(FeastError):

def __init__(self, expected, missing):

super().__init__(

f"The entity dataframe you have provided must contain columns {expected}, "

f"but {missing} were missing."

)

class FeastJoinKeysDuringMaterialization(FeastError):

def __init__(

self, source: str, join_key_columns: Set[str], source_columns: Set[str]

):

super().__init__(

f"The DataFrame from {source} being materialized must have at least {join_key_columns} columns present, "

f"but these were missing: {join_key_columns - source_columns} "

)

class DockerDaemonNotRunning(FeastError):

def __init__(self):

super().__init__(

"The Docker Python sdk cannot connect to the Docker daemon. Please make sure you have"

"the docker daemon installed, and that it is running."

)

class RegistryInferenceFailure(FeastError):

def __init__(self, repo_obj_type: str, specific_issue: str):

super().__init__(

f"Inference to fill in missing information for {repo_obj_type} failed. {specific_issue}. "

"Try filling the information explicitly."

)

class BigQueryJobStillRunning(FeastError):

def __init__(self, job_id):

super().__init__(f"The BigQuery job with ID '{job_id}' is still running.")

class BigQueryJobCancelled(FeastError):

def __init__(self, job_id):

super().__init__(f"The BigQuery job with ID '{job_id}' was cancelled")

class RedshiftCredentialsError(FeastError):

def __init__(self):

super().__init__("Redshift API failed due to incorrect credentials")

class RedshiftQueryError(FeastError):

def __init__(self, details):

super().__init__(f"Redshift SQL Query failed to finish. Details: {details}")

class RedshiftTableNameTooLong(FeastError):

def __init__(self, table_name: str):

super().__init__(

f"Redshift table names have a maximum length of 127 characters, but the table name {table_name} has length {len(table_name)} characters."

)

class SnowflakeCredentialsError(FeastError):

def __init__(self):

super().__init__("Snowflake Connector failed due to incorrect credentials")

class SnowflakeQueryError(FeastError):

def __init__(self, details):

super().__init__(f"Snowflake SQL Query failed to finish. Details: {details}")

class EntityTimestampInferenceException(FeastError):

def __init__(self, expected_column_name: str):

super().__init__(

f"Please provide an entity_df with a column named {expected_column_name} representing the time of events."

)

class FeatureViewMissingDuringFeatureServiceInference(FeastError):

def __init__(self, feature_view_name: str, feature_service_name: str):

super().__init__(

f"Missing {feature_view_name} feature view during inference for {feature_service_name} feature service."

)

class InvalidEntityType(FeastError):

def __init__(self, entity_type: type):

super().__init__(

f"The entity dataframe you have provided must be a Pandas DataFrame or a SQL query, "

f"but we found: {entity_type} "

)

class ConflictingFeatureViewNames(FeastError):

# TODO: print file location of conflicting feature views

def __init__(

self,

feature_view_name: str,

existing_type: Optional[str] = None,

new_type: Optional[str] = None,

):

if existing_type and new_type:

if existing_type == new_type:

# Same-type duplicate

super().__init__(

f"Multiple {existing_type}s with name '{feature_view_name}' found. "

f"Feature view names must be case-insensitively unique. "

f"It may be necessary to ignore certain files in your feature "

f"repository by using a .feastignore file."

)

else:

# Cross-type conflict

super().__init__(

f"Feature view name '{feature_view_name}' is already used by a {existing_type}. "

f"Cannot register a {new_type} with the same name. "

f"Feature view names must be unique across FeatureView, StreamFeatureView, and OnDemandFeatureView."

)

else:

super().__init__(

f"The feature view name: {feature_view_name} refers to feature views of different types."

)

class FeastInvalidInfraObjectType(FeastError):

def __init__(self):

super().__init__("Could not identify the type of the InfraObject.")

class SnowflakeIncompleteConfig(FeastError):

def __init__(self, e: KeyError):

super().__init__(f"{e} not defined in a config file or feature_store.yaml file")

class SnowflakeQueryUnknownError(FeastError):

def __init__(self, query: str):

super().__init__(f"Snowflake query failed: {query}")

class InvalidFeaturesParameterType(FeastError):

def __init__(self, features: Any):

super().__init__(

f"Invalid `features` parameter type {type(features)}. Expected one of List[str] and FeatureService."

)

class EntitySQLEmptyResults(FeastError):

def __init__(self, entity_sql: str):

super().__init__(

f"No entity values found from the specified SQL query to generate the entity dataframe: {entity_sql}."

)

class EntityDFNotDateTime(FeastError):

def __init__(self):

super().__init__(

"The entity dataframe specified does not have the timestamp field as a datetime."

)

class PushSourceNotFoundException(FeastError):

def __init__(self, push_source_name: str):

super().__init__(f"Unable to find push source '{push_source_name}'.")

def http_status_code(self) -> int:

return HttpStatusCode.HTTP_422_UNPROCESSABLE_ENTITY

class ReadOnlyRegistryException(FeastError):

def __init__(self):

super().__init__("Registry implementation is read-only.")

class DataFrameSerializationError(FeastError):

def __init__(self, input: Any):

if isinstance(input, dict):

super().__init__(

f"Failed to serialize the provided dictionary into a pandas DataFrame: {input.keys()}"

)

else:

super().__init__(

"Failed to serialize the provided input into a pandas DataFrame"

)

class PermissionNotFoundException(FeastError):

def __init__(self, name, project):

super().__init__(f"Permission {name} does not exist in project {project}")

class PermissionObjectNotFoundException(FeastObjectNotFoundException):

def __init__(self, name, project=None):

if project:

super().__init__(f"Permission {name} does not exist in project {project}")

else:

super().__init__(f"Permission {name} does not exist")

class ProjectNotFoundException(FeastError):

def __init__(self, project):

super().__init__(f"Project {project} does not exist in registry")

class ProjectObjectNotFoundException(FeastObjectNotFoundException):

def __init__(self, name, project=None):

super().__init__(f"Project {name} does not exist")

class ZeroRowsQueryResult(FeastError):

def __init__(self, query: str):

super().__init__(f"This query returned zero rows:\n{query}")

class ZeroColumnQueryResult(FeastError):

def __init__(self, query: str):

super().__init__(f"This query returned zero columns:\n{query}")

class FeastPermissionError(FeastError, PermissionError):

def __init__(self, details: str):

super().__init__(f"Permission error:\n{details}")

def grpc_status_code(self) -> "GrpcStatusCode":

from grpc import StatusCode as GrpcStatusCode

return GrpcStatusCode.PERMISSION_DENIED

def http_status_code(self) -> int:

return HttpStatusCode.HTTP_403_FORBIDDEN