Skip to content

hyperion.ports.schema_registry

hyperion.ports.schema_registry

Port: schema store abstraction.

Abstract :class:SchemaStore base. Concrete adapters (LocalSchemaStore, S3SchemaStore) live in hyperion.adapters.schema_registry.*; _create_new delegates backend selection to :mod:hyperion.composition (the single composition root). AssetProtocol / AssetType are referenced only in annotations, so this port stays free of the pandera/polars data stack.

SchemaStore 

SchemaStore(path)

Bases: ABC

Abstract base class for schema stores.

Initialize the schema store with the given path.

Parameters:

Name Type Description Default
path str

The path to the schema store.

 required
Source code in hyperion/ports/schema_registry.py 
27
28
29
30
31
32
33
def __init__(self, path: str) -> None:
    """Initialize the schema store with the given path.

    Args:
        path (str): The path to the schema store.
    """
    self.path = path

get_asset_schema 

get_asset_schema(asset)

Get the schema for the given asset.

Parameters:

Name Type Description Default
asset AssetProtocol

The asset to get the schema for.

 required

Returns:

Type Description
dict[str, Any]

dict[str, Any]: The schema for the asset.
Source code in hyperion/ports/schema_registry.py 
35
36
37
38
39
40
41
42
43
44
def get_asset_schema(self, asset: AssetProtocol) -> dict[str, Any]:
    """Get the schema for the given asset.

    Args:
        asset (AssetProtocol): The asset to get the schema for.

    Returns:
        dict[str, Any]: The schema for the asset.
    """
    return self.get_schema(asset.name, asset.schema_version, asset_type=asset.asset_type)

get_schema_from_path abstractmethod 

get_schema_from_path(schema_path)

Get the schema given by its path.

Parameters:

Name Type Description Default
schema_path str

The path to the schema.

 required

Returns:

Type Description
dict[str, Any]

dict[str, Any]: The schema.
Source code in hyperion/ports/schema_registry.py 
46
47
48
49
50
51
52
53
54
55
@abc.abstractmethod
def get_schema_from_path(self, schema_path: str) -> dict[str, Any]:
    """Get the schema given by its path.

    Args:
        schema_path (str): The path to the schema.

    Returns:
        dict[str, Any]: The schema.
    """

get_schema abstractmethod 

get_schema(asset_name, schema_version, asset_type)

Get the schema for the asset with the given name and version.

Source code in hyperion/ports/schema_registry.py 
57
58
59
@abc.abstractmethod
def get_schema(self, asset_name: str, schema_version: int, asset_type: AssetType) -> dict[str, Any]:
    """Get the schema for the asset with the given name and version."""

from_path staticmethod 

from_path(path)

Get a schema store from the given path.

Parameters:

Name Type Description Default
path str

The path to the schema store.

 required

Returns:

Name Type Description
SchemaStore SchemaStore

The schema store.
Source code in hyperion/ports/schema_registry.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
@staticmethod
def from_path(path: str) -> SchemaStore:
    """Get a schema store from the given path.

    Args:
        path (str): The path to the schema store.

    Returns:
        SchemaStore: The schema store.
    """
    if path not in SchemaStore._instances:
        SchemaStore._instances[path] = SchemaStore._create_new(path)
    return SchemaStore._instances[path]

from_config staticmethod 

from_config()

Get a schema store from the configuration.

Returns:

Name Type Description
SchemaStore SchemaStore

The schema store.
Source code in hyperion/ports/schema_registry.py 
81
82
83
84
85
86
87
88
@staticmethod
def from_config() -> SchemaStore:
    """Get a schema store from the configuration.

    Returns:
        SchemaStore: The schema store.
    """
    return SchemaStore.from_path(storage_config.schema_path)