# Copyright 2019 The Feast Authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from datetime import datetime, timezone
from typing import Dict, Optional
from google.protobuf.json_format import MessageToJson
from typeguard import typechecked
from feast.protos.feast.core.Project_pb2 import Project as ProjectProto
from feast.protos.feast.core.Project_pb2 import ProjectMeta as ProjectMetaProto
from feast.protos.feast.core.Project_pb2 import ProjectSpec as ProjectSpecProto
from feast.utils import _utc_now
@typechecked
class Project:
"""
Project is a collection of Feast Objects. Projects provide complete isolation of
feature stores at the infrastructure level.
Attributes:
name: The unique name of the project.
description: A human-readable description.
tags: A dictionary of key-value pairs to store arbitrary metadata.
owner: The owner of the project, typically the email of the primary maintainer.
created_timestamp: The time when the entity was created.
last_updated_timestamp: The time when the entity was last updated.
"""
name: str
description: str
tags: Dict[str, str]
owner: str
created_timestamp: datetime
last_updated_timestamp: datetime
def __init__(
self,
*,
name: str,
description: str = "",
tags: Optional[Dict[str, str]] = None,
owner: str = "",
created_timestamp: Optional[datetime] = None,
last_updated_timestamp: Optional[datetime] = None,
):
"""
Creates Project object.
Args:
name: The unique name of the project.
description (optional): A human-readable description.
tags (optional): A dictionary of key-value pairs to store arbitrary metadata.
owner (optional): The owner of the project, typically the email of the primary maintainer.
created_timestamp (optional): The time when the project was created. Defaults to
last_updated_timestamp (optional): The time when the project was last updated.
Raises:
ValueError: Parameters are specified incorrectly.
"""
self.name = name
self.description = description
self.tags = tags if tags is not None else {}
self.owner = owner
updated_time = _utc_now()
self.created_timestamp = created_timestamp or updated_time
self.last_updated_timestamp = last_updated_timestamp or updated_time
def __hash__(self) -> int:
return hash((self.name))
def __eq__(self, other):
if not isinstance(other, Project):
raise TypeError("Comparisons should only involve Project class objects.")
if (
self.name != other.name
or self.description != other.description
or self.tags != other.tags
or self.owner != other.owner
or self.created_timestamp != other.created_timestamp
or self.last_updated_timestamp != other.last_updated_timestamp
):
return False
return True
def __str__(self):
return str(MessageToJson(self.to_proto()))
def __lt__(self, other):
return self.name < other.name
def is_valid(self):
"""
Validates the state of this project locally.
Raises:
ValueError: The project does not have a name or does not have a type.
"""
if not self.name:
raise ValueError("The project does not have a name.")
from feast.repo_operations import is_valid_name
if not is_valid_name(self.name):
raise ValueError(
f"Project name, {self.name}, should only have "
f"alphanumerical values, underscores, and hyphens but not start with an underscore or hyphen."
)
@classmethod
def from_proto(cls, project_proto: ProjectProto):
"""
Creates a project from a protobuf representation of an project.
Args:
entity_proto: A protobuf representation of an project.
Returns:
An Entity object based on the entity protobuf.
"""
project = cls(
name=project_proto.spec.name,
description=project_proto.spec.description,
tags=dict(project_proto.spec.tags),
owner=project_proto.spec.owner,
)
if project_proto.meta.HasField("created_timestamp"):
project.created_timestamp = (
project_proto.meta.created_timestamp.ToDatetime().replace(
tzinfo=timezone.utc
)
)
if project_proto.meta.HasField("last_updated_timestamp"):
project.last_updated_timestamp = (
project_proto.meta.last_updated_timestamp.ToDatetime().replace(
tzinfo=timezone.utc
)
)
return project
def to_proto(self) -> ProjectProto:
"""
Converts an project object to its protobuf representation.
Returns:
An ProjectProto protobuf.
"""
meta = ProjectMetaProto()
if self.created_timestamp:
meta.created_timestamp.FromDatetime(self.created_timestamp)
if self.last_updated_timestamp:
meta.last_updated_timestamp.FromDatetime(self.last_updated_timestamp)
spec = ProjectSpecProto(
name=self.name,
description=self.description,
tags=self.tags,
owner=self.owner,
)
return ProjectProto(spec=spec, meta=meta)