Schema Registry

Each schema table is registered in SchemaRegistry dynamically at the runtime.

build_schema(name, filepath)

Build schema dataclass from json file path.

Parameters:

Name	Type	Description	Default
name	str | SchemaName	Name of schema table.	required
filepath	str	Path to json file.	required

Returns:

Type	Description
list[SchemaTable]	List of schema dataclasses.

Source code in t4_devkit/schema/builder.py

def build_schema(name: str | SchemaName, filepath: str) -> list[SchemaTable]:
    """Build schema dataclass from json file path.

    Args:
        name (str | SchemaName): Name of schema table.
        filepath (str): Path to json file.

    Returns:
        List of schema dataclasses.
    """
    return SCHEMAS.build_from_json(name, filepath)

SCHEMAS = SchemaRegistry() module-attribute

SchemaRegistry

A manager class to register schema tables.

Source code in t4_devkit/schema/tables/registry.py

class SchemaRegistry:
    """A manager class to register schema tables."""

    def __init__(self) -> None:
        self.__schemas: dict[str, SchemaTable] = {}

    @property
    def schemas(self) -> dict[str, SchemaTable]:
        return self.__schemas

    def __len__(self) -> int:
        return len(self.__schemas)

    def __contains__(self, key: str | SchemaName) -> bool:
        if isinstance(key, SchemaName):
            key = key.value

        return key in self.keys()

    def get(self, key: str | SchemaName) -> SchemaTable:
        if isinstance(key, SchemaName):
            key = key.value

        if key not in self.__schemas:
            raise KeyError(f"{key} has not registered yet.")
        return self.__schemas[key]

    def keys(self) -> KeysView[str]:
        return self.__schemas.keys()

    def register(self, name: SchemaName, *, force: bool = False) -> Callable:
        if not isinstance(name, SchemaName):
            raise TypeError(f"name must be SchemaName, but got {type(name)}.")

        def _register_decorator(obj: object) -> object:
            self._add_module(obj, name=name, force=force)
            return obj

        return _register_decorator

    def _add_module(self, module: object, name: SchemaName, *, force: bool = False) -> None:
        if not inspect.isclass(module):
            raise TypeError(f"module must be a class, but got {type(module)}.")

        if not force and name in self:
            raise KeyError(f"{name.value} has already been registered.")

        self.__schemas[name.value] = module

    def build_from_json(self, key: str | SchemaName, filepath: str) -> list[SchemaTable]:
        """Build schema dataclass from json.

        Args:
            key (str): Name of schema field.
            filepath (str): Path to schema json file.

        Returns:
            Instantiated dataclass.
        """
        if isinstance(key, SchemaName):
            key = key.value

        schema: SchemaTable = self.get(key)
        return schema.from_json(filepath)

build_from_json(key, filepath)

Build schema dataclass from json.

Parameters:

Name	Type	Description	Default
key	str	Name of schema field.	required
filepath	str	Path to schema json file.	required

Returns:

Type	Description
list[SchemaTable]	Instantiated dataclass.

Source code in t4_devkit/schema/tables/registry.py

def build_from_json(self, key: str | SchemaName, filepath: str) -> list[SchemaTable]:
    """Build schema dataclass from json.

    Args:
        key (str): Name of schema field.
        filepath (str): Path to schema json file.

    Returns:
        Instantiated dataclass.
    """
    if isinstance(key, SchemaName):
        key = key.value

    schema: SchemaTable = self.get(key)
    return schema.from_json(filepath)