Source code for esrf_pathlib._schemas.definitions.load

import importlib
from itertools import product
from typing import Dict
from typing import List
from typing import Optional

from ..errors import UnknownPathSchema
from ..schema.path import PathSchema

_ALL_VERSIONS = {"unknown": [1], "esrf": [1, 2, 3], "tomo": [1]}




def default_schema_version(name: str) -> int:
    """
    :raises UnknownPathSchema
    """
    if name not in _ALL_VERSIONS:
        raise UnknownPathSchema(f"Schema {name!r} is not known")
    return max(_ALL_VERSIONS[name])





def fallback_schema_versions(
    start_versions: Dict[str, int],
    fallback_depth: Optional[int] = None,
) -> List[Dict[str, int]]:
    """
    Return fallback schema versions starting from ``start_versions``.

    - Excludes ``start_versions``.
    - Versions are sorted from high to low.
    - If fallback_depth is None -> include all older versions.
    - If fallback_depth is set -> include up to that many older versions.

    :raises ValueError:
    :raises UnknownPathSchema:
    """
    fallback_versions = {}
    if fallback_depth and fallback_depth < 0:
        raise ValueError("'fallback_depth' must be >=0")

    for sname, start_version in start_versions.items():
        if sname not in _ALL_VERSIONS:
            raise UnknownPathSchema(f"Schema {sname!r} is not known")
        all_versions = sorted(_ALL_VERSIONS.get(sname), reverse=True)

        if start_version not in all_versions:
            raise UnknownPathSchema(
                f"Schema {sname!r} does not have version {start_version}"
            )

        # Exclude current version, take older ones
        idx = all_versions.index(start_version)
        fallbacks = all_versions[idx + 1 :]

        # Limit fallback depth
        if fallback_depth is not None:
            fallbacks = fallbacks[:fallback_depth]

        fallback_versions[sname] = fallbacks

    return [
        dict(zip(start_versions.keys(), combination))
        for combination in product(
            *(fallback_versions[sname] for sname in start_versions)
        )
    ]





def get_schema(name: str, version: int) -> PathSchema:
    """
    :raises UnknownPathSchema:
    """
    module_name = f"{name}_v{version}"
    try:
        mod = importlib.import_module(f".{module_name}", package=__package__)
    except ImportError as ex:
        raise UnknownPathSchema(f"{name!r} version {version} does not exist") from ex
    return mod.get_schema()