Source code for ewokscore.graph.schema

from typing import Callable, Optional
import networkx
from packaging.version import parse as parse_version, Version
import logging
import importlib

# Major version: increment when changing the existing schema
# Minor version: increment when adding features or deprecating the existing schema
LATEST_VERSION = parse_version("1.1")

# The default version may be set to something else if we don't want the latest version to be the default
DEFAULT_VERSION = LATEST_VERSION

# Map graph versions to ewokscore version bounds. Whenever we change the schema
# which the current ewokscore version needs and updating is not possible:
#   - increment the ewokscore version
#   - use that version as upper bound of the last item of _VERSION_BOUNDS
#   - use that version as lower bound of a new item of _VERSION_BOUNDS
_VERSION_BOUNDS = None




def get_version_bounds() -> dict:
    global _VERSION_BOUNDS
    if _VERSION_BOUNDS:
        return _VERSION_BOUNDS

    _VERSION_BOUNDS = dict()
    _VERSION_BOUNDS[parse_version("0.0")] = parse_version("0.0"), parse_version("0.0.1")
    _VERSION_BOUNDS[parse_version("0.1")] = parse_version("0.1.0-rc"), None
    _VERSION_BOUNDS[parse_version("0.2")] = parse_version("0.1.0-rc"), None
    _VERSION_BOUNDS[parse_version("1.0")] = parse_version("0.1.0-rc"), None
    _VERSION_BOUNDS[parse_version("1.1")] = parse_version("0.1.0-rc"), None
    return _VERSION_BOUNDS



logger = logging.getLogger(__name__)




def normalize_schema_version(graph: dict):
    schema_version = graph["graph"].get("schema_version", None)
    if not schema_version:
        schema_version = DEFAULT_VERSION
        graph["graph"]["schema_version"] = str(schema_version)
        logger.info(
            'Graph has no "schema_version": assume version "%s"', schema_version
        )
        return
    pversion = parse_version(schema_version)
    if pversion != LATEST_VERSION:
        # This warning is given because an exception may occur before `update_graph_schema`
        # is called due to the different schema version.
        logger.warning(
            'Graph schema version "%s" is not equal to the latest version "%s"',
            pversion,
            LATEST_VERSION,
        )
    graph["graph"]["schema_version"] = str(pversion)





def update_graph_schema(graph: networkx.DiGraph) -> bool:
    """Updates the graph description to a higher schema version (returns `True`) or raises an
    exception. If the schema version is known it will provide library version bounds
    in the exception message. Returns `False` when the graph does not need
    any update.
    """
    schema_version = graph.graph.get("schema_version", None)
    if schema_version is None:
        schema_version = DEFAULT_VERSION
        graph.graph["schema_version"] = str(schema_version)
        logger.info(
            'Graph has no "schema_version": assume version "%s"', schema_version
        )
    else:
        schema_version = parse_version(schema_version)
    if schema_version == LATEST_VERSION:
        return False

    update_method = _get_update_method(schema_version)
    if not update_method:
        raise GraphSchemaError(schema_version)

    before = graph.graph.get("schema_version", None)
    try:
        update_method(graph)
    except Exception:
        raise GraphSchemaError(schema_version)
    else:
        after = graph.graph.get("schema_version", None)
        assert before != after, "graph conversion did not update the schema version"
        return True



def _get_update_method(
    schema_version: Version,
) -> Optional[Callable[[networkx.DiGraph], None]]:
    try:
        mod = importlib.import_module(
            __name__ + ".v" + str(schema_version).replace(".", "_")
        )
    except ImportError:
        return None
    return mod.update_graph_schema




class GraphSchemaError(ValueError):
    def __init__(self, schema_version: Version) -> None:
        lbound, ubound = get_version_bounds().get(schema_version, (None, None))
        if lbound and ubound:
            return super().__init__(
                f'Graph schema version "{schema_version}" requires another library version: python3 -m pip install "ewokscore>={lbound},<{ubound}"`'
            )
        elif lbound:
            return super().__init__(
                f'Graph schema version "{schema_version}" requires another library version: python3 -m pip install "ewokscore>={lbound}"'
            )
        elif ubound:
            return super().__init__(
                f'Graph schema version "{schema_version}" requires another library version: python3 -m pip install "ewokscore<{ubound}"'
            )
        else:
            return super().__init__(
                f'Graph schema version "{schema_version}" is either invalid or requires a newer library version: python3 -m pip install --upgrade ewokscore'
            )