API

Cython module

__file__ = '/home/docs/checkouts/readthedocs.org/user_builds/finance-dag/checkouts/latest/fdag/fast_dag.cpython-312-x86_64-linux-gnu.so' module-attribute

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

__name__ = 'fdag.fast_dag' module-attribute

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

__package__ = 'fdag' module-attribute

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

__test__ = {} module-attribute

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

DAGEngine

Manages the lifecycle and topology of the graph.

__doc__ = 'Manages the lifecycle and topology of the graph.' class-attribute

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

__pyx_vtable__ = <capsule object NULL at 0x7d49a345b420> class-attribute

Capsule objects let you wrap a C "void *" pointer in a Python object. They're a way of passing data through the Python interpreter without creating your own custom type.

Capsules are used for communication between extension modules. They provide a way for an extension module to export a C interface to other extension modules, so that extension modules can use the Python import mechanism to link to one another.

__init__(*args, **kwargs) method descriptor

Initialize self. See help(type(self)) for accurate signature.

__new__(*args, **kwargs) builtin

Create and return a new object. See help(type) for accurate signature.

Node

A C-level extension type representing a single node in the DAG.

__doc__ = '\n A C-level extension type representing a single node in the DAG.\n ' class-attribute

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

__pyx_vtable__ = <capsule object NULL at 0x7d49a2eb3720> class-attribute

Capsule objects let you wrap a C "void *" pointer in a Python object. They're a way of passing data through the Python interpreter without creating your own custom type.

Capsules are used for communication between extension modules. They provide a way for an extension module to export a C interface to other extension modules, so that extension modules can use the Python import mechanism to link to one another.

__init__(*args, **kwargs) method descriptor

Initialize self. See help(type(self)) for accurate signature.

__new__(*args, **kwargs) builtin

Create and return a new object. See help(type) for accurate signature.

add_dependency(dep) method descriptor

Creates a bidirectional edge. Prevents duplicates during dynamic access.

invalidate() method descriptor

Fast C-level recursive invalidation. Uses bint (boolean integer) for 0-overhead checks.

Python module

ReactiveMixin

Source code in fdag/dag.py

class ReactiveMixin:
    def __init__(self):
        super().__setattr__("_dag_dependencies", {})
        super().__setattr__("_call_stack", [])
        super().__setattr__("_cache", {})
        super().__setattr__("_invalidated", set())

    def __setattr__(self, name: str, value: Any) -> None:
        if name.startswith("_"):
            super().__setattr__(name, value)
            return

        cls_attr = getattr(self.__class__, name, None)
        if callable(cls_attr) and getattr(cls_attr, "__is_reactive_node__", False):
            if not (getattr(cls_attr, "__node_flags__", 0) & CAN_OVERRIDE):
                msg = f"Cannot override node '{name}' because it was not marked with @node(CAN_OVERRIDE)."
                raise AttributeError(msg)

            def override_wrapper(*_args, **_kwargs):
                return value

            override_wrapper.__is_reactive_node__ = True  # type: ignore[attr-defined]
            override_wrapper.__node_flags__ = getattr(cls_attr, "__node_flags__", 0)  # type: ignore[attr-defined]

            super().__setattr__(name, override_wrapper)
            self.invalidate(name)
            return

        super().__setattr__(name, value)

    def invalidate(self, name: str):
        """
        Recursively marks a node and all of its dependents as invalidated,
        clearing them from the cache so they recalculate on the next access.
        """
        invalidated_set = super().__getattribute__("_invalidated")
        cache = super().__getattribute__("_cache")
        deps = super().__getattribute__("_dag_dependencies")

        def cascade(node):
            invalidated_set.add(node)
            if node in cache:
                del cache[node]

            for dependent, dependencies in deps.items():
                if node in dependencies and dependent not in invalidated_set:
                    cascade(dependent)

        cascade(name)

    def __getattribute__(self, name):
        if name.startswith("__") or name in {
            "_dag_dependencies",
            "_call_stack",
            "_cache",
            "_invalidated",
            "invalidate",
        }:
            return super().__getattribute__(name)

        stack = super().__getattribute__("_call_stack")
        deps = super().__getattribute__("_dag_dependencies")
        cache = super().__getattribute__("_cache")
        invalidated_set = super().__getattribute__("_invalidated")

        # map edge from caller to this node
        if stack:
            caller = stack[-1]
            if caller != name:
                deps.setdefault(caller, set()).add(name)

        attr = super().__getattribute__(name)

        if callable(attr):
            if getattr(attr, "__is_reactive_node__", False):
                return apply_cache(self, name, attr)
            return attr

        if name in cache and name not in invalidated_set:
            return cache[name]

        return attr

invalidate(name)

Recursively marks a node and all of its dependents as invalidated, clearing them from the cache so they recalculate on the next access.

Source code in fdag/dag.py

def invalidate(self, name: str):
    """
    Recursively marks a node and all of its dependents as invalidated,
    clearing them from the cache so they recalculate on the next access.
    """
    invalidated_set = super().__getattribute__("_invalidated")
    cache = super().__getattribute__("_cache")
    deps = super().__getattribute__("_dag_dependencies")

    def cascade(node):
        invalidated_set.add(node)
        if node in cache:
            del cache[node]

        for dependent, dependencies in deps.items():
            if node in dependencies and dependent not in invalidated_set:
                cascade(dependent)

    cascade(name)

analyze_method_dependencies_1(cls, method_name)

Analyze and return the set of attribute or method dependencies used within a given method of a class.

Parameters:

Name	Type	Description	Default
cls	type	The class containing the method.	required
method_name	str	The name of the method to analyze.	required

Returns:

Name	Type	Description
set	set	A set of dependency names used in the specified method.

Source code in fdag/dag.py

def analyze_method_dependencies_1(cls: type, method_name: str) -> set:
    """
    Analyze and return the set of attribute or method dependencies used within a given method of a class.

    Args:
        cls: The class containing the method.
        method_name: The name of the method to analyze.

    Returns:
        set: A set of dependency names used in the specified method.
    """
    source = inspect.getsource(cls)
    tree = ast.parse(textwrap.dedent(source))

    for node in ast.walk(tree):
        if isinstance(node, ast.FunctionDef) and node.name == method_name:
            visitor = _FunctionDependencyVisitor()
            visitor.visit(node)
            return visitor.dependencies
    msg = f"Method '{method_name}' not found in class '{cls.__name__}'"
    raise ValueError(msg)

analyze_method_dependencies_2(cls, method_name)

Analyze and return the set of attribute or method dependencies used within a given method of a class.

Parameters:

Name	Type	Description	Default
cls	type	The class containing the method.	required
method_name	str	The name of the method to analyze.	required

Returns:

Name	Type	Description
set	set	A set of dependency names used in the specified method.

Source code in fdag/dag.py

def analyze_method_dependencies_2(cls: type, method_name: str) -> set:
    """
    Analyze and return the set of attribute or method dependencies used within a given method of a class.

    Args:
        cls: The class containing the method.
        method_name: The name of the method to analyze.

    Returns:
        set: A set of dependency names used in the specified method.
    """
    source = inspect.getsource(cls)
    tree = ast.parse(source)

    for node in ast.walk(tree):
        if isinstance(node, ast.FunctionDef) and node.name == method_name:
            visitor = _FunctionDependencyVisitor2()
            visitor.visit(node)
            return visitor.dependencies
    msg = f"Method '{method_name}' not found in class '{cls.__name__}'"
    raise ValueError(msg)

apply_cache(instance, name, func)

Wraps a method to add caching, dependency tracking, and eager evaluation tied to a specific ReactiveMixin instance.

Source code in fdag/dag.py

def apply_cache(instance, name: str, func: Callable) -> Callable:
    """
    Wraps a method to add caching, dependency tracking, and eager evaluation
    tied to a specific ReactiveMixin instance.
    """

    # noinspection PyProtectedMember
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        cache = instance._cache  # noqa: SLF001
        invalidated_set = instance._invalidated  # noqa: SLF001
        stack = instance._call_stack  # noqa: SLF001

        if name in cache and name not in invalidated_set:
            return cache[name]

        cache_key = (instance.__class__.__name__, name)
        if cache_key not in _STATIC_DEPS_CACHE:
            try:
                deps = analyze_method_dependencies_1(instance.__class__, name)
            except (ValueError, TypeError, OSError):
                deps = set()

            _STATIC_DEPS_CACHE[cache_key] = deps

        static_deps = _STATIC_DEPS_CACHE[cache_key]

        for dep in static_deps:
            if dep not in cache and hasattr(instance, dep):
                getattr(instance, dep)()

        stack.append(name)
        try:
            result = func(*args, **kwargs)
            cache[name] = result
            invalidated_set.discard(name)
            return result
        finally:
            stack.pop()

    return wrapper