### Install ipyflow via pip

Source: https://ipyflow.readthedocs.io/en/latest/_sources/getting_started/installation.rst.txt

Standard installation command for the ipyflow package.

```bash
pip install ipyflow
```

--------------------------------

### Retrieve path information

Source: https://ipyflow.readthedocs.io/en/latest/_modules/ipyflow/data_model/symbol.html

Properties to get the full path or namespace-qualified path of the symbol.

```python
@property
def full_path(self) -> Tuple[str, ...]:
    return self.containing_scope.full_path + (str(self.name),)

@property
def full_namespace_path(self) -> str:
    return self.containing_scope.make_namespace_qualified_name(self)
```

--------------------------------

### Manually install the kernelspec

Source: https://ipyflow.readthedocs.io/en/latest/_sources/getting_started/installation.rst.txt

Use this command to register the ipyflow kernel in a specific environment, such as a container or moved virtualenv.

```bash
python -m ipyflow.install --sys-prefix
```

--------------------------------

### Multiset operation relations

Source: https://ipyflow.readthedocs.io/en/latest/_modules/collections.html

Examples showing how Counter multiset operations correspond to regular set operations when multiplicities are zero or one.

```python
    cp = Counter(a=1, b=0, c=1)
    cq = Counter(c=1, d=0, e=1)
The corresponding regular sets would be:
    sp = {'a', 'c'}
    sq = {'c', 'e'}
All of the following relations would hold:
    set(cp + cq) == sp | sq
    set(cp - cq) == sp - sq
    set(cp | cq) == sp | sq
    set(cp & cq) == sp & sq
    (cp == cq) == (sp == sq)
    (cp != cq) == (sp != sq)
    (cp <= cq) == (sp <= sq)
    (cp < cq) == (sp < sq)
    (cp >= cq) == (sp >= sq)
    (cp > cq) == (sp > sq)
```

--------------------------------

### namedtuple Factory Implementation

Source: https://ipyflow.readthedocs.io/en/latest/_modules/collections.html

Implementation of the namedtuple factory, including field validation and docstring examples.

```python
try:
    from _collections import _tuplegetter
except ImportError:
    _tuplegetter = lambda index, doc: property(_itemgetter(index), doc=doc)

def namedtuple(typename, field_names, *, rename=False, defaults=None, module=None):
    """Returns a new subclass of tuple with named fields.

    >>> Point = namedtuple('Point', ['x', 'y'])
    >>> Point.__doc__                   # docstring for the new class
    'Point(x, y)'
    >>> p = Point(11, y=22)             # instantiate with positional args or keywords
    >>> p[0] + p[1]                     # indexable like a plain tuple
    33
    >>> x, y = p                        # unpack like a regular tuple
    >>> x, y
    (11, 22)
    >>> p.x + p.y                       # fields also accessible by name
    33
    >>> d = p._asdict()                 # convert to a dictionary
    >>> d['x']
    11
    >>> Point(**d)                      # convert from a dictionary
    Point(x=11, y=22)
    >>> p._replace(x=100)               # _replace() is like str.replace() but targets named fields
    Point(x=100, y=22)

    """

    # Validate the field names.  At the user's option, either generate an error
    # message or automatically replace the field name with a valid name.
    if isinstance(field_names, str):
        field_names = field_names.replace(',', ' ').split()
    field_names = list(map(str, field_names))
    typename = _sys.intern(str(typename))

    if rename:
        seen = set()
        for index, name in enumerate(field_names):
            if (not name.isidentifier()
                or _iskeyword(name)
                or name.startswith('_')
                or name in seen):
                field_names[index] = f'_{index}'
            seen.add(name)

    for name in [typename] + field_names:
        if type(name) is not str:
            raise TypeError('Type names and field names must be strings')
        if not name.isidentifier():
            raise ValueError('Type names and field names must be valid '
                             f'identifiers: {name!r}')
        if _iskeyword(name):
            raise ValueError('Type names and field names cannot be a '
                             f'keyword: {name!r}')

    seen = set()
    for name in field_names:
        if name.startswith('_') and not rename:
            raise ValueError('Field names cannot start with an underscore: '
                             f'{name!r}')
        if name in seen:
            raise ValueError(f'Encountered duplicate field name: {name!r}')
        seen.add(name)

    field_defaults = {}
    if defaults is not None:
        defaults = tuple(defaults)
        if len(defaults) > len(field_names):
            raise TypeError('Got more default values than field names')
        field_defaults = dict(reversed(list(zip(reversed(field_names),
                                                reversed(defaults)))))
```

--------------------------------

### Retrieve or create namespace

Source: https://ipyflow.readthedocs.io/en/latest/_modules/ipyflow/data_model/symbol.html

Returns the existing namespace or initializes a new one if it does not exist.

```python
    def namespaced(self) -> "Namespace":
        ns = self.namespace
        if ns is not None:
            return ns
        return namespaces()(
            self.obj,
            self.name,
            parent_scope=self.containing_scope,
            original_symbol=self,
        )
```

--------------------------------

### Get sanitized content

Source: https://ipyflow.readthedocs.io/en/latest/_modules/ipyflow/data_model/cell.html

Returns the sanitized content of the cell.

```python
    def sanitized_content(self) -> str:
        return self._rewriter_and_sanitized_content()[1]
```

--------------------------------

### Get memoized counter

Source: https://ipyflow.readthedocs.io/en/latest/_modules/ipyflow/data_model/cell.html

Retrieves the memoized counter for the current cell if applicable.

```python
def get_memoized_counter(self) -> Optional[int]:
    prev_cell = self.prev_cell
    if not self.is_memoized or prev_cell is None:
        return None
```

--------------------------------

### %flow waiting [global|all]

Source: https://ipyflow.readthedocs.io/en/latest/reference/flow_magic.html

List symbols that are waiting on newer dependencies.

```APIDOC
## %flow waiting [global|all]

### Description
List symbols that are waiting on newer dependencies (stale). `global` (the default) considers top-level names; `all` considers every alias. Alias: `show_waiting`.
```

--------------------------------

### Define notebook cells

Source: https://ipyflow.readthedocs.io/en/latest/_sources/getting_started/first_notebook.rst.txt

Initializes variables to establish a basic dependency chain.

```python
run_cell("raw = 10")
run_cell("scaled = raw * 2")
run_cell("summary = scaled + 1")
```

--------------------------------

### Get type annotations

Source: https://ipyflow.readthedocs.io/en/latest/_modules/ipyflow/data_model/symbol.html

Retrieves type annotation information for the symbol's object.

```python
def get_type_annotation(self):
    return get_type_annotation(self.obj)

def get_type_annotation_string(self) -> str:
    return make_annotation_string(self.get_type_annotation())
```

--------------------------------

### Initializing Counter objects

Source: https://ipyflow.readthedocs.io/en/latest/_modules/collections.html

Various ways to instantiate a new Counter object.

```python
>>> c = Counter()                           # a new, empty counter
    >>> c = Counter('gallahad')                 # a new counter from an iterable
    >>> c = Counter({'a': 4, 'b': 2})           # a new counter from a mapping
    >>> c = Counter(a=4, b=2)                   # a new counter from keyword args
```

--------------------------------

### Get top-level symbol

Source: https://ipyflow.readthedocs.io/en/latest/_modules/ipyflow/data_model/symbol.html

Traverses the scope hierarchy to find the top-level symbol definition.

```python
    def get_top_level(self) -> Optional["Symbol"]:
        if not self.containing_scope.is_namespace_scope:
            return self
        else:
            containing_scope = cast("Namespace", self.containing_scope)
            for alias in flow().aliases.get(containing_scope.obj_id, []):
                if alias.is_globally_accessible:
                    return alias.get_top_level()
            return None
```

--------------------------------

### Use Reactive-Variable Syntax

Source: https://ipyflow.readthedocs.io/en/latest/_sources/guides/reactive_execution.rst.txt

Prefix a symbol with $ to opt an individual variable into reactivity, allowing updates to propagate even in lazy mode.

```text
# Cell 1
x = 1

# Cell 2 -- `$x` marks this usage as reactive
y = $x + 1
```

--------------------------------

### Resetting flow and capturing output

Source: https://ipyflow.readthedocs.io/en/latest/_sources/guides/introspection_api.rst.txt

Initializes a clean notebook state and verifies captured output from a cell execution.

```python
run_cell = reset_flow()  # start this example from a clean notebook
run_cell("print('captured output')")
run_cell("assert stdout(1) is not None and 'captured output' in stdout(1)")
```

--------------------------------

### Get statement counts

Source: https://ipyflow.readthedocs.io/en/latest/_modules/ipyflow/data_model/cell.html

Properties to retrieve the number of original and total statements in the cell.

```python
    @property
    def num_original_stmts(self) -> int:
        return len(self.to_ast().body)

    @property
    def num_stmts(self) -> int:
        return self.num_original_stmts + (self._extra_stmt is not None)
```

--------------------------------

### Switch Execution Mode at Runtime

Source: https://ipyflow.readthedocs.io/en/latest/_sources/guides/reactive_execution.rst.txt

Use the %flow magic command to toggle between lazy and reactive execution modes during a session.

```python
run_cell("from ipyflow import flow")
run_cell("get_ipython().run_line_magic('flow', 'mode reactive')")
run_cell("assert flow().mut_settings.exec_mode.value == 'reactive'")

run_cell("get_ipython().run_line_magic('flow', 'mode lazy')")
run_cell("assert flow().mut_settings.exec_mode.value == 'lazy'")
```

--------------------------------

### Get namespace symbols

Source: https://ipyflow.readthedocs.io/en/latest/_modules/ipyflow/data_model/symbol.html

Yields symbols within the namespace, optionally recursing through nested namespaces.

```python
    def get_namespace_symbols(
        self, recurse: bool = False, seen: Optional[Set["Symbol"]] = None
    ) -> Generator["Symbol", None, None]:
        ns = self.namespace
        if ns is None:
            return
        if seen is None:
            seen = set()
        if self in seen:
            return
        seen.add(self)
        for sym in ns.all_symbols_this_indentation():
            yield sym
            if recurse:
                yield from sym.get_namespace_symbols(recurse=recurse, seen=seen)
```

--------------------------------

### Match call arguments with definition

Source: https://ipyflow.readthedocs.io/en/latest/_modules/ipyflow/data_model/symbol.html

Maps call-site arguments to definition-site arguments for tracking data flow.

```python
def _match_call_args_with_definition_args(
    self,
) -> Generator[Tuple[ast.arg, List["Symbol"]], None, None]:
    # TODO: handle posonlyargs, kwonlyargs
    assert self.func_def_stmt is not None and isinstance(
        self.func_def_stmt, (ast.FunctionDef, ast.AsyncFunctionDef, ast.Lambda)
    )
    caller_node = self._get_calling_ast_node()
    if caller_node is None or not isinstance(caller_node, ast.Call):
        return
    kwarg_by_name = {
        arg_key.arg: arg_key
        for arg_key in self.func_def_stmt.args.args[
            -len(self.func_def_stmt.args.defaults) :
        ]
    }
    if not all(keyword.arg in kwarg_by_name for keyword in caller_node.keywords):
        logger.warning("detected mismatched kwargs from caller node to definition")
        return
    def_args = self.func_def_stmt.args.args
    if len(self.func_def_stmt.args.defaults) > 0:
        def_args = def_args[: -len(self.func_def_stmt.args.defaults)]
    if len(def_args) > 0 and def_args[0].arg == "self":
        # FIXME: this is bad and I should feel bad
        def_args = def_args[1:]
    for def_arg, call_arg in zip(def_args, caller_node.args):
        if isinstance(call_arg, ast.Starred):
            # give up
            # TODO: handle this case
            break
        yield def_arg, tracer().resolve_loaded_symbols(call_arg)
    seen_keys = set()
    for keyword in caller_node.keywords:
        keyword_key, keyword_value = keyword.arg, keyword.value
        if keyword_key is None or keyword_value is None:
            continue
        seen_keys.add(keyword_key)
        yield kwarg_by_name[keyword_key], tracer().resolve_loaded_symbols(
            keyword_value
        )
    for arg_key, arg_value in zip(
        self.func_def_stmt.args.args[-len(self.func_def_stmt.args.defaults) :],
        self.func_def_stmt.args.defaults,
    ):
        if arg_key.arg in seen_keys:
            continue
        yield arg_key, tracer().resolve_loaded_symbols(arg_value)
```

--------------------------------

### Get reference count

Source: https://ipyflow.readthedocs.io/en/latest/_modules/ipyflow/data_model/symbol.html

Calculates the reference count for the symbol, adjusting for aliases and namespace ownership.

```python
def get_ref_count(self) -> int:
    if self.obj is None or self.obj is Symbol.NULL:
        return -1
    total = sys.getrefcount(self.obj) - 1
    total -= len(flow().aliases.get(self.obj_id, []))
    ns = self.namespace
    if ns is not None and ns.obj is self.obj:
        total -= 1
    ns = self.containing_namespace
    if ns is not None and not ns.is_garbage:
        total += 1
    return total
```

--------------------------------

### Symbol class initialization and internal methods

Source: https://ipyflow.readthedocs.io/en/latest/_modules/ipyflow/data_model/symbol.html

Internal implementation details for the Symbol class, including state initialization, implicit handling, and property accessors for aliases and liveness.

```python
        self._tombstone = False
        self._cached_out_of_sync = True
        self.cached_obj_id: Optional[int] = None
        self.cached_obj_type: Optional[Type[object]] = None
        self.cached_obj_len: Optional[int] = None
        if refresh_cached_obj:
            self._refresh_cached_obj()
        self.containing_scope = containing_scope or flow().global_scope
        self.call_scope: Optional[Scope] = None
        self.func_def_stmt: Optional[Union[ast.stmt, ast.Lambda]] = None
        self.stmt_node = self.update_stmt_node(stmt_node)
        self.symbol_node = symbol_node
        self._funcall_live_symbols = None
        self.parents: Dict["Symbol", List[Timestamp]] = {}
        self.children: Dict["Symbol", List[Timestamp]] = {}

        # initialize at -1 for implicit since the corresponding piece of data could already be around,
        # and we don't want liveness checker to think this was newly created unless we
        # explicitly trace an update somewhere
        self._timestamp: Timestamp = (
            Timestamp.uninitialized() if implicit else Timestamp.current()
        )
        self._snapshot_timestamps: List[Timestamp] = []
        self._snapshot_timestamp_ubounds: List[Timestamp] = []
        self._defined_cell_num = cells().exec_counter()
        self._is_dangling_on_edges = False
        self._override_ready_liveness_cell_num = -1
        self._override_timestamp: Optional[Timestamp] = None
        self.watchpoints = Watchpoints()

        # The necessary last-updated timestamp / cell counter for this symbol to not be waiting
        self.required_timestamp: Timestamp = self.timestamp

        # for each usage of this sym, the version that was used, if different from the timestamp of usage
        self.timestamp_by_used_time: Dict[Timestamp, Timestamp] = {}
        self.used_node_by_used_time: Dict[Timestamp, ast.AST] = {}
        # History of definitions at time of liveness
        self.timestamp_by_liveness_time: Dict[Timestamp, Timestamp] = {}
        # All timestamps associated with updates to this symbol
        self._updated_timestamps: Set[Timestamp] = set()
        # The most recent timestamp associated with a particular object id
        self.last_updated_timestamp_by_obj_id: Dict[int, Timestamp] = {}

        self.fresher_ancestors: Set["Symbol"] = set()
        self.fresher_ancestor_timestamps: Set[Timestamp] = set()

        # cells where this symbol was live
        self.cells_where_deep_live: Set[Cell] = set()
        self.cells_where_shallow_live: Set[Cell] = set()

        self._last_computed_ready_or_waiting_cache_ts: int = -1
        self._is_ready_or_waiting_at_position_cache: Dict[Tuple[int, bool], bool] = {}

        # if implicitly created when tracing non-store-context ast nodes
        self._implicit = implicit

        # Will never be stale if no_warning is True
        self.disable_warnings = False
        self._temp_disable_warnings = False

        self._num_ipywidget_observers = 0
        self._num_mercury_widget_observers = 0

        flow().aliases.setdefault(id(obj), set()).add(self)
        if (
            isinstance(self.name, str)
            and not self.is_anonymous
            and not self.containing_scope.is_namespace_scope
        ):
            ns = self.namespace
            if ns is not None and ns.scope_name == "self":
                # hack to get a better name than `self.whatever` for fields of this object
                # not ideal because it relies on the `self` convention but is probably
                # acceptable for the use case of improving readable names
                ns.scope_name = self.name
        self._maybe_fix_implicitness()
```

```python
    def _maybe_fix_implicitness(self) -> None:
        if (
            self.is_implicit
            and not self._timestamp.is_initialized
            and self.containing_namespace is not None
        ):
            self._timestamp = max(
                (
                    sym.timestamp
                    for sym in flow().aliases.get(self.containing_namespace.obj_id, [])
                ),
                default=self._timestamp,
            )
        if self._timestamp.is_initialized and self._implicit:
            self._implicit = False
```

```python
    @property
    def updated_timestamps(self) -> Set[Timestamp]:
        init_ts = self._initialized_timestamp
        if init_ts.is_initialized:
            return self._updated_timestamps | {init_ts}
        else:
            return self._updated_timestamps
```

```python
    @property
    def aliases(self) -> List["Symbol"]:
        return list(flow().aliases.get(self.obj_id, []))
```

```python
    @property
    def cells_where_live(self) -> Set[Cell]:
        return self.cells_where_deep_live | self.cells_where_shallow_live
```

```python
    def __repr__(self) -> str:
        return f"<{self.readable_name}>"

    def __str__(self) -> str:
        return self.readable_name

    def __hash__(self) -> int:
        return hash(id(self))
```

--------------------------------

### Get cells with placeholder IDs

Source: https://ipyflow.readthedocs.io/en/latest/_modules/ipyflow/data_model/cell.html

Returns a sorted list of cells that currently have a placeholder ID.

```python
@classmethod
    def with_placeholder_ids(cls):
        return sorted(
            (
                cell
                for cell in cls._current_cell_by_cell_id.values()
                if cell._placeholder_id
            ),
            key=lambda cell: cell.cell_ctr,
        )
```

--------------------------------

### Verify execution timestamps and staleness

Source: https://ipyflow.readthedocs.io/en/latest/_sources/concepts/dataflow_model.rst.txt

Demonstrates how timestamps track execution order and how updating a dependency marks dependent symbols as stale.

```python
run_cell("a = 1")            # a updated at cell 1
run_cell("b = a + 1")        # b updated at cell 2
run_cell("assert timestamp(a).cell_num == 1")
run_cell("assert timestamp(b).cell_num == 2")
# Re-running a gives it a newer timestamp than b, so b is now stale.
run_cell("a = 2")
run_cell("assert timestamp(a).cell_num > timestamp(b).cell_num")
run_cell("assert lift(b).is_waiting")
```

--------------------------------

### Initialize Cell class

Source: https://ipyflow.readthedocs.io/en/latest/_modules/ipyflow/data_model/cell.html

Constructor for the Cell class, initializing state tracking, symbol dependencies, and execution metadata.

```python
    def __init__(
        self,
        cell_id: IdType,
        cell_ctr: int,
        content: str,
        tags: Tuple[str, ...],
        prev_cell: Optional["Cell"] = None,
        placeholder_id: bool = False,
        memoized_output_level: Optional[MemoizedOutputLevel] = None,
    ) -> None:
        self.cell_id: IdType = cell_id
        self.cell_ctr: int = cell_ctr
        self.last_check_content: Optional[str] = None
        self.last_check_cell_ctr: Optional[int] = None
        self.last_check_result: Optional[CheckerResult] = None
        self.error_in_exec: Optional[BaseException] = None
        self.history: List[int] = [cell_ctr] if cell_ctr > -1 else []
        self.executed_content: Optional[str] = None
        self.current_content: str = content
        self.last_ast_content: Optional[str] = None
        self.captured_output: Optional[IPyflowCapturedIO] = None
        self.tags: Tuple[str, ...] = tags
        self.prev_cell = prev_cell
        self.override_live_refs: Optional[List[str]] = None
        self.override_dead_refs: Optional[List[str]] = None
        self.reactive_tags: Set[str] = set()
        self.raw_dynamic_parents: Dict[IdType, Set["Symbol"]] = {}
        self.raw_dynamic_children: Dict[IdType, Set["Symbol"]] = {}
        self.raw_static_parents: Dict[IdType, Set["Symbol"]] = {}
        self.raw_static_children: Dict[IdType, Set["Symbol"]] = {}
        self.used_symbols: Set["Symbol"] = set()
        self.static_removed_symbols: Set["Symbol"] = set()
        self.static_writes: Set["Symbol"] = set()
        self.dynamic_writes: Set["Symbol"] = set()
        # pending dynamic writes are not finalized until the stmt is done executing
        self._pending_dynamic_writes: Set["Symbol"] = set()
        self._used_cell_counters_by_live_symbol: Dict["Symbol", Set[int]] = defaultdict(
            set
        )
        self._cached_ast: Optional[ast.Module] = None
        self._cached_typecheck_result: Optional[bool] = (
            None if flow().settings.mark_typecheck_failures_unsafe else True
        )
        self._ready: bool = False
        self._extra_stmt: Optional[ast.stmt] = None
        self._placeholder_id = placeholder_id
        self.memoized_output_level = memoized_output_level
        self.skipped_due_to_memoization_ctr = -1
```

--------------------------------

### Get raw and sanitized content

Source: https://ipyflow.readthedocs.io/en/latest/_modules/ipyflow/data_model/cell.html

Returns a tuple containing the raw cell content and the sanitized content.

```python
    def raw_and_sanitized_content(self, path: Optional[str] = None) -> Tuple[str, str]:
        raw_cell = self.raw_cell
        return raw_cell, self._rewriter_and_sanitized_content(raw_cell, path=path)[1]
```

--------------------------------

### Initialize Symbol container

Source: https://ipyflow.readthedocs.io/en/latest/_modules/ipyflow/data_model/symbol.html

Initializes the global symbol container if it is empty.

```python
if len(_SymbolContainer) == 0:
    _SymbolContainer.append(Symbol)
else:
    _SymbolContainer[0] = Symbol
```

--------------------------------

### Get transformed memoized content

Source: https://ipyflow.readthedocs.io/en/latest/_modules/ipyflow/data_model/cell.html

Returns the memoized content string or a pass statement based on the output level.

```python
    def get_transformed_memoized_content(
        self, ctr: Optional[int] = None
    ) -> Optional[str]:
        if ctr is None:
            ctr = self.get_memoized_counter()
        if ctr is None:
            return None
        if self.memoized_output_level == MemoizedOutputLevel.QUIET:
            return "pass"
        else:
            return f"Out.get({ctr})"
```

--------------------------------

### Switch Flow Direction at Runtime

Source: https://ipyflow.readthedocs.io/en/latest/_sources/guides/reactive_execution.rst.txt

Toggle between in-order and any-order flow directions using the %flow magic command.

```python
run_cell("from ipyflow import flow")
run_cell("get_ipython().run_line_magic('flow', 'direction any_order')")
run_cell("assert flow().mut_settings.flow_order.value == 'any_order'")

run_cell("get_ipython().run_line_magic('flow', 'direction in_order')")
run_cell("assert flow().mut_settings.flow_order.value == 'in_order'")
```

--------------------------------

### deps(sym: Any) -> List[Symbol]

Source: https://ipyflow.readthedocs.io/en/latest/_modules/ipyflow/api/lift.html

Looks up the corresponding dependencies for a given symbol.

```APIDOC
## deps(sym)

### Description
Given the programmatic usage of some symbol, look up the corresponding dependencies for that symbol.

### Parameters
- **sym** (Any) - Required - The symbol to look up.
```

--------------------------------

### reproduce_cell(ctr, show_input=True, show_output=True, lookback=0)

Source: https://ipyflow.readthedocs.io/en/latest/_modules/ipyflow/api/cells.html

Re-renders the input and/or output of a previous cell execution.

```APIDOC
## reproduce_cell(ctr, show_input=True, show_output=True, lookback=0)

### Description
Re-renders the input and/or output of a previous cell execution. This can recover results that have been overwritten within the current kernel session.

### Parameters
- **ctr** (int) - Required - The execution counter of the cell to reproduce.
- **show_input** (bool) - Optional - Whether to render the cell's input source. Defaults to True.
- **show_output** (bool) - Optional - Whether to render the cell's captured output. Defaults to True.
- **lookback** (int) - Optional - How many executions to step back for this cell. 0 is the latest execution, 1 is the one before it, etc. Defaults to 0.
```

--------------------------------

### Initialize Symbol Class

Source: https://ipyflow.readthedocs.io/en/latest/_modules/ipyflow/data_model/symbol.html

Constructor for the Symbol class, which tracks variable bindings and their associated metadata in the dataflow graph.

```python
def __init__(
    self,
    name: SupportedIndexType,
    symbol_type: SymbolType,
    obj: Any,
    containing_scope: "Scope",
    stmt_node: Optional[Union[ast.stmt, ast.Lambda]] = None,
    symbol_node: Optional[ast.AST] = None,
    refresh_cached_obj: bool = False,
    implicit: bool = False,
) -> None:
    if refresh_cached_obj:
        # TODO: clean up redundancies
        assert implicit
        assert stmt_node is None
    self.name = name
    self.symbol_type = symbol_type
    self.obj = obj

    # additional user-specific metadata
    self._tags: Set[str] = set()
    self.extra_metadata: Dict[str, Any] = {}
```

--------------------------------

### Cell execution and readiness properties

Source: https://ipyflow.readthedocs.io/en/latest/_modules/ipyflow/data_model/cell.html

Provides properties to check if a cell has been executed or is ready.

```python
@property
    def is_executed(self) -> bool:
        return self.cell_ctr > -1

    @property
    def is_ready(self) -> bool:
        return self._ready

    def set_ready(self, new_ready: bool) -> bool:
        old_ready = self._ready
        self._ready = new_ready
        return old_ready
```

--------------------------------

### OrderedDict View Methods

Source: https://ipyflow.readthedocs.io/en/latest/_modules/collections.html

Methods to return set-like views of keys, items, and values.

```python
def keys(self):
    "D.keys() -> a set-like object providing a view on D's keys"
    return _OrderedDictKeysView(self)
```

```python
def items(self):
    "D.items() -> a set-like object providing a view on D's items"
    return _OrderedDictItemsView(self)
```

```python
def values(self):
    "D.values() -> an object providing a view on D's values"
    return _OrderedDictValuesView(self)
```

--------------------------------

### %flow mode [lazy|reactive]

Source: https://ipyflow.readthedocs.io/en/latest/_sources/reference/flow_magic.rst.txt

Set the execution mode for the session.

```APIDOC
## %flow mode [lazy|reactive]

### Description
Set the ExecutionMode for the session.
```

--------------------------------

### Configure Default Execution Mode

Source: https://ipyflow.readthedocs.io/en/latest/_sources/guides/reactive_execution.rst.txt

Set the default execution mode globally in the IPython configuration file.

```python
c = get_config()
c.ipyflow.exec_mode = "reactive"  # defaults to "lazy"
```

--------------------------------

### Initialize cell container

Source: https://ipyflow.readthedocs.io/en/latest/_modules/ipyflow/data_model/cell.html

Registers the Cell class into the global container.

```python
if len(_CodeCellContainer) == 0:
    _CodeCellContainer.append(Cell)
else:
    _CodeCellContainer[0] = Cell
```

--------------------------------

### rdeps(sym: Any) -> List[Symbol]

Source: https://ipyflow.readthedocs.io/en/latest/_modules/ipyflow/api/lift.html

Looks up the corresponding recursive dependencies for a given symbol.

```APIDOC
## rdeps(sym)

### Description
Given the programmatic usage of some symbol, look up the corresponding recursive dependencies for that symbol.

### Parameters
- **sym** (Any) - Required - The symbol to look up.
```

--------------------------------

### Format multi-slice for a symbol

Source: https://ipyflow.readthedocs.io/en/latest/_modules/ipyflow/data_model/symbol.html

Returns a formatted multi-slice representation of the symbol for a specific version.

```python
        :param format_type: optional output format (e.g. ``str``); defaults to a
            rich :class:`~ipywidgets.HTML` when available, otherwise plain text.
        :param version: which historical version of the symbol to slice for.
            ``-1`` (the default) is the current version.
        """
        return statements().format_multi_slice(
            self._get_timestamps_for_version(version=version),
            blacken=True,
            format_type=format_type,
        )
```

--------------------------------

### Reproduce cell execution

Source: https://ipyflow.readthedocs.io/en/latest/_modules/ipyflow/data_model/cell.html

Prints the executed content and captured output of a cell, with optional lookback functionality.

```python
    def reproduce(
        self, show_input: bool = True, show_output: bool = True, lookback: int = 0
    ) -> Any:
        cell_to_repro = self
        for _ in range(lookback):
            assert cell_to_repro.prev_cell is not None
            cell_to_repro = cell_to_repro.prev_cell
        if show_input:
            print_ = print
            print_(cell_to_repro.executed_content or "")
            max_len = max(
                len(line)
                for line in (cell_to_repro.executed_content or "").splitlines()
            )
            print_("-" * max_len)
        if show_output and cell_to_repro.captured_output is not None:
            cell_to_repro.captured_output.show(render_out_expr=False)
        return shell().user_ns["Out"].get(cell_to_repro.cell_ctr)
```

--------------------------------

### ipyflow.models.namespaces()

Source: https://ipyflow.readthedocs.io/en/latest/reference/models.html

Return the Namespace class.

```APIDOC
## ipyflow.models.namespaces()

### Description
Return the Namespace class.
```

--------------------------------

### Inspect symbol dependencies

Source: https://ipyflow.readthedocs.io/en/latest/_sources/getting_started/first_notebook.rst.txt

Uses deps and users to query immediate upstream and downstream relationships in the dataflow graph.

```python
run_cell("assert deps(summary) == [lift(scaled)]")
run_cell("assert deps(scaled) == [lift(raw)]")
run_cell("assert users(raw) == [lift(scaled)]")
```

--------------------------------

### Register watchpoints

Source: https://ipyflow.readthedocs.io/en/latest/_sources/guides/introspection_api.rst.txt

Use watchpoints to register callbacks that fire whenever a symbol is written.

```python
run_cell("counter = 0")
run_cell("watchpoints(counter).add(lambda obj, position, name: None)")
run_cell("assert len(watchpoints(counter)) == 1")
```

--------------------------------

### code(sym: Any, **kwargs: Any) -> Union[HTML, str]

Source: https://ipyflow.readthedocs.io/en/latest/_modules/ipyflow/api/lift.html

Looks up the corresponding code for a given symbol.

```APIDOC
## code(sym, **kwargs)

### Description
Given the programmatic usage of some symbol, look up the corresponding code for that symbol.

### Parameters
- **sym** (Any) - Required - The symbol to look up.
- **kwargs** (Any) - Optional - Additional arguments for code retrieval.
```

--------------------------------

### UserString Class Definition and Methods

Source: https://ipyflow.readthedocs.io/en/latest/_modules/collections.html

Implementation of the UserString wrapper class, providing string-like behavior and method delegation.

```python
class UserString(_collections_abc.Sequence):

    def __init__(self, seq):
        if isinstance(seq, str):
            self.data = seq
        elif isinstance(seq, UserString):
            self.data = seq.data[:]
        else:
            self.data = str(seq)

    def __str__(self):
        return str(self.data)

    def __repr__(self):
        return repr(self.data)

    def __int__(self):
        return int(self.data)

    def __float__(self):
        return float(self.data)

    def __complex__(self):
        return complex(self.data)

    def __hash__(self):
        return hash(self.data)

    def __getnewargs__(self):
        return (self.data[:],)

    def __eq__(self, string):
        if isinstance(string, UserString):
            return self.data == string.data
        return self.data == string

    def __lt__(self, string):
        if isinstance(string, UserString):
            return self.data < string.data
        return self.data < string

    def __le__(self, string):
        if isinstance(string, UserString):
            return self.data <= string.data
        return self.data <= string

    def __gt__(self, string):
        if isinstance(string, UserString):
            return self.data > string.data
        return self.data > string

    def __ge__(self, string):
        if isinstance(string, UserString):
            return self.data >= string.data
        return self.data >= string

    def __contains__(self, char):
        if isinstance(char, UserString):
            char = char.data
        return char in self.data

    def __len__(self):
        return len(self.data)

    def __getitem__(self, index):
        return self.__class__(self.data[index])

    def __add__(self, other):
        if isinstance(other, UserString):
            return self.__class__(self.data + other.data)
        elif isinstance(other, str):
            return self.__class__(self.data + other)
        return self.__class__(self.data + str(other))

    def __radd__(self, other):
        if isinstance(other, str):
            return self.__class__(other + self.data)
        return self.__class__(str(other) + self.data)

    def __mul__(self, n):
        return self.__class__(self.data * n)

    __rmul__ = __mul__

    def __mod__(self, args):
        return self.__class__(self.data % args)

    def __rmod__(self, template):
        return self.__class__(str(template) % self)

    # the following methods are defined in alphabetical order:
    def capitalize(self):
        return self.__class__(self.data.capitalize())

    def casefold(self):
        return self.__class__(self.data.casefold())

    def center(self, width, *args):
        return self.__class__(self.data.center(width, *args))

    def count(self, sub, start=0, end=_sys.maxsize):
        if isinstance(sub, UserString):
            sub = sub.data
        return self.data.count(sub, start, end)
```

--------------------------------

### %flow register_tracer <module.path.to.TracerClass>

Source: https://ipyflow.readthedocs.io/en/latest/_sources/reference/flow_magic.rst.txt

Register an additional pyccolo tracer for the session.

```APIDOC
## %flow register_tracer <module.path.to.TracerClass>

### Description
Register an additional pyccolo tracer for the session.

### Parameters
- **module.path.to.TracerClass** (string) - Required - The import path to the tracer class.
```

--------------------------------

### users(sym: Any) -> List[Symbol]

Source: https://ipyflow.readthedocs.io/en/latest/_modules/ipyflow/api/lift.html

Looks up the corresponding users of a given symbol.

```APIDOC
## users(sym)

### Description
Given the programmatic usage of some symbol, look up the corresponding users of that symbol.

### Parameters
- **sym** (Any) - Required - The symbol to look up.
```

--------------------------------

### %flow deps <symbol>

Source: https://ipyflow.readthedocs.io/en/latest/_sources/reference/flow_magic.rst.txt

Print a symbol's immediate dependencies and dependents.

```APIDOC
## %flow deps <symbol>

### Description
Print a symbol's immediate dependencies and dependents.

### Parameters
- **symbol** (string) - Required - The symbol name to inspect.
```

--------------------------------

### Query transitive dependencies

Source: https://ipyflow.readthedocs.io/en/latest/_sources/getting_started/first_notebook.rst.txt

Uses rdeps and rusers to traverse the dependency graph recursively.

```python
run_cell("assert set(rdeps(summary)) == {lift(raw), lift(scaled)}")
run_cell("assert set(rusers(raw)) == {lift(scaled), lift(summary)}")
```

--------------------------------

### Query dependencies and dependents

Source: https://ipyflow.readthedocs.io/en/latest/_sources/guides/introspection_api.rst.txt

Use deps, users, rdeps, and rusers to navigate the dataflow graph. These functions return lists of Symbol objects and skip anonymous internal symbols.

```python
run_cell("a = 0")
run_cell("b = a + 0")
run_cell("c = b + 0")

run_cell("assert deps(b) == [lift(a)]")
run_cell("assert users(a) == [lift(b)]")

run_cell("assert set(rdeps(c)) == {lift(a), lift(b)}")
run_cell("assert set(rusers(a)) == {lift(b), lift(c)}")
```

```python
run_cell('''
    def f():
        p = 0
        q = 1
        return p + q
''')
run_cell("z = f()")
run_cell("assert sorted(repr(d) for d in deps(z)) == ['<f>', '<p>', '<q>']")
```

--------------------------------

### UserList Operator and Method Implementations

Source: https://ipyflow.readthedocs.io/en/latest/_modules/collections.html

Implementation of list-like operations and operator overloading for the UserList class.

```python
    def __radd__(self, other):
        if isinstance(other, UserList):
            return self.__class__(other.data + self.data)
        elif isinstance(other, type(self.data)):
            return self.__class__(other + self.data)
        return self.__class__(list(other) + self.data)

    def __iadd__(self, other):
        if isinstance(other, UserList):
            self.data += other.data
        elif isinstance(other, type(self.data)):
            self.data += other
        else:
            self.data += list(other)
        return self

    def __mul__(self, n):
        return self.__class__(self.data * n)

    __rmul__ = __mul__

    def __imul__(self, n):
        self.data *= n
        return self

    def __copy__(self):
        inst = self.__class__.__new__(self.__class__)
        inst.__dict__.update(self.__dict__)
        # Create a copy and avoid triggering descriptors
        inst.__dict__["data"] = self.__dict__["data"][:]
        return inst

    def append(self, item):
        self.data.append(item)

    def insert(self, i, item):
        self.data.insert(i, item)

    def pop(self, i=-1):
        return self.data.pop(i)

    def remove(self, item):
        self.data.remove(item)

    def clear(self):
        self.data.clear()

    def copy(self):
        return self.__class__(self)

    def count(self, item):
        return self.data.count(item)

    def index(self, item, *args):
        return self.data.index(item, *args)

    def reverse(self):
        self.data.reverse()

    def sort(self, /, *args, **kwds):
        self.data.sort(*args, **kwds)

    def extend(self, other):
        if isinstance(other, UserList):
            self.data.extend(other.data)
        else:
            self.data.extend(other)
```

--------------------------------

### lift(sym: Any) -> Symbol

Source: https://ipyflow.readthedocs.io/en/latest/_modules/ipyflow/api/lift.html

Looks up the corresponding Symbol metadata for a given symbol.

```APIDOC
## lift(sym)

### Description
Given the programmatic usage of some symbol, look up the corresponding Symbol metadata.

### Parameters
- **sym** (Any) - Required - The symbol to look up.
```

--------------------------------

### Create Symbols for Call Arguments

Source: https://ipyflow.readthedocs.io/en/latest/_modules/ipyflow/data_model/symbol.html

Initializes symbols for function call arguments by matching them against definition arguments and updating the call scope.

```python
def create_symbols_for_call_args(self, call_frame: FrameType) -> None:
    assert self.func_def_stmt is not None
    seen_def_args = set()
    logger.info("create symbols for call to %s", self)
    for def_arg, deps in self._match_call_args_with_definition_args():
        seen_def_args.add(def_arg.arg)
        self.call_scope.upsert_symbol_for_name(  # type: ignore[union-attr]
            def_arg.arg,
            call_frame.f_locals.get(def_arg.arg),
            deps,
            self.func_def_stmt,
            propagate=False,
            symbol_node=def_arg,
        )
        logger.info("def arg %s matched with deps %s", def_arg, deps)
    for def_arg in self.get_definition_args():
        if def_arg.arg in seen_def_args:
            continue
        self.call_scope.upsert_symbol_for_name(  # type: ignore[union-attr]
            def_arg.arg,
            None,
            set(),
            self.func_def_stmt,
            propagate=False,
            symbol_node=def_arg,
        )
```

--------------------------------

### set_tag(sym: Any, tag_value: str) -> None

Source: https://ipyflow.readthedocs.io/en/latest/_modules/ipyflow/api/lift.html

Adds a tag to the symbol.

```APIDOC
## set_tag(sym, tag_value)

### Description
Add the tag `value` to the symbol.

### Parameters
- **sym** (Any) - Required - The symbol to tag.
- **tag_value** (str) - Required - The tag value to add.
```

--------------------------------

### Using reactive-variable syntax

Source: https://ipyflow.readthedocs.io/en/latest/guides/reactive_execution.html

The $ prefix marks a symbol as reactive, allowing updates to propagate even in lazy execution mode.

```python
# Cell 1
x = 1

# Cell 2 -- `$x` marks this usage as reactive
y = $x + 1

# Now re-running Cell 1 with a new value reactively refreshes Cell 2,
# even though the global execution mode is still lazy.
```

--------------------------------

### OrderedDict Import Handling

Source: https://ipyflow.readthedocs.io/en/latest/_modules/collections.html

Conditional import logic for OrderedDict, falling back to pure Python implementation if the C extension is unavailable.

```python
try:
    from _collections import OrderedDict
except ImportError:
    # Leave the pure Python version in place.
    pass
```

--------------------------------

### OrderedDict Implementation Methods

Source: https://ipyflow.readthedocs.io/en/latest/_modules/collections.html

Internal methods for OrderedDict handling pickling, copying, key-based creation, and dictionary-like operations.

```python
def __reduce__(self):
        'Return state information for pickling'
        state = self.__getstate__()
        if state:
            if isinstance(state, tuple):
                state, slots = state
            else:
                slots = {}
            state = state.copy()
            slots = slots.copy()
            for k in vars(OrderedDict()):
                state.pop(k, None)
                slots.pop(k, None)
            if slots:
                state = state, slots
            else:
                state = state or None
        return self.__class__, (), state, None, iter(self.items())

    def copy(self):
        'od.copy() -> a shallow copy of od'
        return self.__class__(self)

    @classmethod
    def fromkeys(cls, iterable, value=None):
        '''Create a new ordered dictionary with keys from iterable and values set to value.
        '''
        self = cls()
        for key in iterable:
            self[key] = value
        return self

    def __eq__(self, other):
        '''od.__eq__(y) <==> od==y.  Comparison to another OD is order-sensitive
        while comparison to a regular mapping is order-insensitive.

        '''
        if isinstance(other, OrderedDict):
            return dict.__eq__(self, other) and all(map(_eq, self, other))
        return dict.__eq__(self, other)

    def __ior__(self, other):
        self.update(other)
        return self

    def __or__(self, other):
        if not isinstance(other, dict):
            return NotImplemented
        new = self.__class__(self)
        new.update(other)
        return new

    def __ror__(self, other):
        if not isinstance(other, dict):
            return NotImplemented
        new = self.__class__(other)
        new.update(self)
        return new
```

--------------------------------

### %flow slice [<cell_num>]

Source: https://ipyflow.readthedocs.io/en/latest/_sources/reference/flow_magic.rst.txt

Print the code needed to reconstruct a cell using dynamic program slicing.

```APIDOC
## %flow slice [<cell_num>]

### Description
Print the code needed to reconstruct a cell using dynamic program slicing. With no argument, slices the most recent cell.

### Parameters
- **cell_num** (integer) - Optional - The cell number to slice.
- **--stmt** (flag) - Optional - Slice at statement granularity.
- **--blacken** (flag) - Optional - Reformat the result with black.
- **--tag** (string) - Optional - Slice the cells carrying a tag instead of a cell number.
- **--noheader** (flag) - Optional - Omit the # Cell N headers.
```

--------------------------------

### Verbose memoization output

Source: https://ipyflow.readthedocs.io/en/latest/_sources/guides/memoization.rst.txt

Use the --verbose flag to replay stdout and rich display outputs from the cache.

```python
%%memoize --verbose
print("this stdout is replayed from cache on a hit")
fig  # rich output replayed too
```

--------------------------------

### Retrieve import metadata

Source: https://ipyflow.readthedocs.io/en/latest/_modules/ipyflow/data_model/symbol.html

Properties to extract module information and original names for symbols identified as imports.

```python
    @property
    def imported_module(self) -> str:
        if not self.is_import:
            raise ValueError("only IMPORT symbols have `imported_module` property")
        if isinstance(self.stmt_node, ast.Import):
            for alias in self.stmt_node.names:
                name = alias.asname or alias.name
                if name == self.name:
                    return alias.name
            raise ValueError(
                "Unable to find module for symbol %s is stmt %s"
                % (self, ast.dump(self.stmt_node))
            )
        elif isinstance(self.stmt_node, ast.ImportFrom):
            assert (
                self.stmt_node.module is not None
            )  # in a repl, there shouldn't be relative imports
            return self.stmt_node.module
        else:
            raise TypeError(
                "Invalid stmt type for import symbol: %s" % None
                if self.stmt_node is None
                else ast.dump(self.stmt_node)
            )

    @property
    def imported_symbol_original_name(self) -> str:
        if not self.is_import:
            raise ValueError(
                "only IMPORT symbols have `imported_symbol_original_name` property"
            )
        if isinstance(self.stmt_node, ast.Import):
            return self.imported_module
        elif isinstance(self.stmt_node, ast.ImportFrom):
            for alias in self.stmt_node.names:
                name = alias.asname or alias.name
                if name == self.name:
                    return alias.name
            raise ValueError(
                "Unable to find module for symbol %s is stmt %s"
                % (self, ast.dump(self.stmt_node))
            )
        else:
            raise TypeError(
                "Invalid stmt type for import symbol: %s" % None
                if self.stmt_node is None
                else ast.dump(self.stmt_node)
            )
```

--------------------------------

### Interrogate Dataflow Graph with Programmatic API

Source: https://ipyflow.readthedocs.io/en/latest/_sources/index.rst.txt

Use run_cell to simulate notebook execution and inspect dependencies or reconstruct code slices using the ipyflow API.

```python
run_cell("x = 0")
run_cell("y = x + 1")

# `deps` / `users` report the immediate neighbors in the dataflow graph.
run_cell("assert deps(y) == [lift(x)]")
run_cell("assert users(x) == [lift(y)]")

# `code` reconstructs the minimal program slice that produces a symbol.
run_cell("assert str(code(y)) == '# Cell 1\nx = 0\n\n# Cell 2\ny = x + 1'")
```

--------------------------------

### Handle Widget Creation

Source: https://ipyflow.readthedocs.io/en/latest/_modules/ipyflow/data_model/symbol.html

Detects if an object is an ipywidget and sets up observation if applicable.

```python
    def _handle_possible_widget_creation(self) -> None:
        if self.obj is None:
            return
        Widget = getattr(sys.modules.get("ipywidgets"), "Widget", None)
        if (
            Widget is None
            or self.is_obj_lazy_module
            or not isinstance(self.obj, Widget)
            or not hasattr(self.obj, "observe")
            or not hasattr(self.obj, "value")
        ):
            return
        self.namespaced().upsert_symbol_for_name(
            "value", getattr(self.obj, "value", None), set(), self.stmt_node
        )
        self.obj.observe(self._observe_widget)
        self._num_ipywidget_observers += 1
```