hyperion.domain.assets

hyperion.domain.assets

Pure asset identities for the catalog.

Split out of :mod:hyperion.entities.catalog (F6 / DDD refactor Step 2). These types describe asset identity (name, schema version, partition keys, path / metadata helpers) and depend only on the stdlib and lite-core :mod:hyperion.dateutils -- no pydantic, pandera or polars. The pandera/polars feature models live in :mod:hyperion.data.asset_schemas.

AssetProtocol

Bases: Protocol

Protocol for assets in the catalog.

This protocol defines the interface for assets in the catalog.

Attributes:

Name	Type	Description
asset_type	ClassVar[AssetType]	The type of the asset.
name	str	The name of the asset.
schema_version	int	The schema version of the asset.

schema_version property

schema_version

The schema version of the asset.

get_path

get_path(prefix='')

Get the path for the asset with the given prefix.

Parameters:

Name	Type	Description	Default
prefix	str	The prefix for the path.	''

Returns:

Name	Type	Description
str	str	The path for the asset.

Source code in hyperion/domain/assets.py

def get_path(self, prefix: str = "") -> str:
    """Get the path for the asset with the given prefix.

    Args:
        prefix (str): The prefix for the path.

    Returns:
        str: The path for the asset.
    """

to_metadata

to_metadata()

Get the metadata for the asset.

Returns:

Type	Description
dict[str, str]	dict[str, str]: The metadata for the asset.

Source code in hyperion/domain/assets.py

def to_metadata(self) -> dict[str, str]:
    """Get the metadata for the asset.

    Returns:
        dict[str, str]: The metadata for the asset.
    """

DataLakeAsset dataclass

DataLakeAsset(name, date, schema_version=1)

Data lake asset.

Attributes:

Name	Type	Description
asset_type	ClassVar[AssetType]	The type of the asset.
name	str	The name of the asset.
date	datetime	The date of the asset.
schema_version	int	The schema version of the asset.

get_path

get_path(prefix='')

Get the path for the asset with the given prefix.

Source code in hyperion/domain/assets.py

def get_path(self, prefix: str = "") -> str:
    """Get the path for the asset with the given prefix."""
    date = assure_timezone(self.date).isoformat()
    return get_prefixed_path(f"{self.name}/date={date}/v{self.schema_version}.avro", prefix)

to_metadata

to_metadata()

Get the metadata for the asset.

Source code in hyperion/domain/assets.py

def to_metadata(self) -> dict[str, str]:
    """Get the metadata for the asset."""
    return {"name": self.name, "date": self.date.isoformat(), "schema_version": str(self.schema_version)}

PersistentStoreAsset dataclass

PersistentStoreAsset(name, schema_version=1)

Persistent store asset.

Attributes:

Name	Type	Description
asset_type	ClassVar[AssetType]	The type of the asset.
name	str	The name of the asset.
schema_version	int	The schema version of the asset.

get_path

get_path(prefix='')

Get the path for the asset with the given prefix.

Source code in hyperion/domain/assets.py

def get_path(self, prefix: str = "") -> str:
    """Get the path for the asset with the given prefix."""
    return get_prefixed_path(f"{self.name}/v{self.schema_version}.avro", prefix)

to_metadata

to_metadata()

Get the metadata for the asset.

Source code in hyperion/domain/assets.py

def to_metadata(self) -> dict[str, str]:
    """Get the metadata for the asset."""
    return {"name": self.name, "schema_version": str(self.schema_version)}

FeatureAsset dataclass

FeatureAsset(name, partition_date, resolution, schema_version=1, partition_keys=dict())

Feature asset.

Attributes:

Name	Type	Description
asset_type	ClassVar[AssetType]	The type of the asset.
name	str	The name of the asset.
partition_date	datetime	The partition timestamp of the asset.
resolution	TimeResolution | str	The resolution of the asset.
schema_version	int	The schema version of the asset.
partition_keys	dict[str, str]	The partition keys of the asset.

time_resolution property

time_resolution

The time resolution of the feature.

feature_name property

feature_name

The name of the feature including the time resolution.

get_path

get_path(prefix='')

Get the path for the asset with the given prefix.

Source code in hyperion/domain/assets.py

def get_path(self, prefix: str = "") -> str:
    """Get the path for the asset with the given prefix."""
    partition_date = assure_timezone(self.partition_date).isoformat()
    keys_prefix = self._get_partition_keys_prefix()
    if keys_prefix:
        keys_prefix = keys_prefix + "/"
    return get_prefixed_path(
        f"{self.feature_name}/{keys_prefix}partition_date={partition_date}/v{self.schema_version}.avro", prefix
    )

to_metadata

to_metadata()

Get the metadata for the asset.

Source code in hyperion/domain/assets.py

def to_metadata(self) -> dict[str, str]:
    """Get the metadata for the asset."""
    return {
        "name": self.name,
        "partition_date": self.partition_date.isoformat(),
        "schema_version": str(self.schema_version),
        "partition_keys": json.dumps(self.partition_keys),
    }

get_prefixed_path

get_prefixed_path(path, prefix='')

Get the path with the given prefix.

Parameters:

Name	Type	Description	Default
path	str	The path.	required
prefix	str	The prefix.	''

Returns:

Name	Type	Description
str	str	The path with the prefix.

Source code in hyperion/domain/assets.py

def get_prefixed_path(path: str, prefix: str = "") -> str:
    """Get the path with the given prefix.

    Args:
        path (str): The path.
        prefix (str): The prefix.

    Returns:
        str: The path with the prefix.
    """
    prefix = prefix.strip("/")
    if prefix:
        prefix = f"{prefix}/"
    return prefix + path