Class RequestSource

sdk/python/feast/data_source.py:592–706

RequestSource that can be used to provide input features for on demand transforms Attributes: name: Name of the request data source schema: Schema mapping from the input feature name to a ValueType description: A human-readable description. tags: A dictionar

@typechecked
class RequestSource(DataSource):
    """
    RequestSource that can be used to provide input features for on demand transforms

    Attributes:
        name: Name of the request data source
        schema: Schema mapping from the input feature name to a ValueType
        description: A human-readable description.
        tags: A dictionary of key-value pairs to store arbitrary metadata.
        owner: The owner of the request data source, typically the email of the primary
            maintainer.
    """

    name: str
    schema: List[Field]
    description: str
    tags: Dict[str, str]
    owner: str

    def __init__(
        self,
        *,
        name: str,
        schema: List[Field],
        timestamp_field: Optional[str] = None,
        description: Optional[str] = "",
        tags: Optional[Dict[str, str]] = None,
        owner: Optional[str] = "",
    ):
        """Creates a RequestSource object."""
        super().__init__(
            name=name,
            timestamp_field=timestamp_field,
            description=description,
            tags=tags,
            owner=owner,
        )
        self.schema = schema

    def validate(self, config: RepoConfig):
        raise NotImplementedError

    def get_table_column_names_and_types(
        self, config: RepoConfig
    ) -> Iterable[Tuple[str, str]]:
        raise NotImplementedError

    def __eq__(self, other):
        if not isinstance(other, RequestSource):
            raise TypeError(
                "Comparisons should only involve RequestSource class objects."
            )

        if not super().__eq__(other):
            return False

        if isinstance(self.schema, List) and isinstance(other.schema, List):
            for field1, field2 in zip(self.schema, other.schema):