Skip to content

graph

django_spire.contrib.sync.core.graph

DependencyGraph

Source code in django_spire/contrib/sync/core/graph.py
def __init__(
    self,
    edges: dict[str, set[str]],
    deferred_edges: dict[str, set[str]] | None = None,
) -> None:
    for label in edges:
        if not label:
            message = 'edges must not contain empty labels'
            raise InvalidParameterError(message)

    self._edges = {
        label: set(dependencies)
        for label, dependencies in edges.items()
    }

    all_labels = set(self._edges)

    for label, dependencies in self._edges.items():
        unknown = dependencies - all_labels

        if unknown:
            message = (
                f'Model {label!r} declares dependencies on '
                f'unknown models: {unknown}'
            )

            raise UnknownDependencyError(message)

    self._deferred_edges: dict[str, set[str]] = {}

    if deferred_edges:
        for label, targets in deferred_edges.items():
            if not targets:
                continue

            if label not in all_labels:
                message = (
                    f'Deferred edge source {label!r} '
                    f'is not a known model'
                )

                raise InvalidParameterError(message)

            unknown = targets - all_labels

            if unknown:
                message = (
                    f'Deferred edges from {label!r} reference '
                    f'unknown models: {unknown}'
                )

                raise UnknownDependencyError(message)

            self._deferred_edges[label] = set(targets)

    self._dependents: dict[str, set[str]] = {
        label: set() for label in self._edges
    }

    for label, dependencies in self._edges.items():
        for dependency in dependencies:
            self._dependents[dependency].add(label)

    self._order = self._compute_order()

deferred_edges property

dependencies

Source code in django_spire/contrib/sync/core/graph.py
def dependencies(self, label: str) -> set[str]:
    return set(self._edges.get(label, set()))

has_deferred_edges

Source code in django_spire/contrib/sync/core/graph.py
def has_deferred_edges(self) -> bool:
    return bool(self._deferred_edges)

known_models

Source code in django_spire/contrib/sync/core/graph.py
def known_models(self) -> frozenset[str]:
    return frozenset(self._edges)

sync_order

Source code in django_spire/contrib/sync/core/graph.py
def sync_order(self) -> list[str]:
    return list(self._order)