Class DirectedGraph

marimo/_runtime/dataflow/graph.py:32–440

Main entry point that coordinates all graph operations. Responsibilities: - Coordinate topology, definitions, cycles - Execute register_cell/delete_cell operations - Maintain thread safety - Delegate to specialists

@dataclass(frozen=True)
class DirectedGraph(GraphTopology):
    """Main entry point that coordinates all graph operations.

    Responsibilities:
    - Coordinate topology, definitions, cycles
    - Execute register_cell/delete_cell operations
    - Maintain thread safety
    - Delegate to specialists
    """

    topology: MutableGraphTopology = field(
        default_factory=MutableGraphTopology
    )
    definition_registry: DefinitionRegistry = field(
        default_factory=DefinitionRegistry
    )
    cycle_tracker: CycleTracker = field(default_factory=CycleTracker)

    # Cell ids whose last post-execution broadcast advertised a top-level
    # reusability hint. Tracked here (not on the per-run CellImpl, which is
    # replaced when a cell's code changes) so the hint can be cleared exactly
    # on the top-level -> non-top-level transition. Not topology; broadcast
    # bookkeeping that needs to survive cell re-registration.
    cells_serving_serialization_hint: set[CellId_t] = field(
        default_factory=set
    )

    # This lock must be acquired during methods that mutate the graph; it's
    # only needed because a graph is shared between the kernel and the code
    # completion service. It should almost always be uncontended.
    lock: threading.Lock = field(default_factory=threading.Lock)

    def is_cell_cached(self, cell_id: CellId_t, code: str) -> bool:
        """Whether a cell with id `cell_id` and code `code` is in the graph."""
        return (
            cell_id in self.topology.cells
            and code_key(code) == self.topology.cells[cell_id].key
        )

    def get_defining_cells(self, name: Name) -> set[CellId_t]:
        """Get all cells that define name.

        This is a singleton for well-formed graphs.
        """
        return self.definition_registry.get_defining_cells(name)

    def get_referring_cells(
        self, name: Name, language: Literal["python", "sql"]
    ) -> set[CellId_t]:
        """Get all cells that have a ref to `name`.

        The variable can be either a Python variable or a SQL variable (table).
        SQL variables don't leak to Python cells, but Python variables do leak
        to SQL.

        Only does a local analysis of refs, without taking into consideration
        whether refs are defined by other cells.
        """