API Reference¶

Complete reference for all public classes, methods, and module-level helpers in yieldgraph.

Graph¶

The main entry point. Build a pipeline with add_chain, then execute it with run.

from yieldgraph import Graph

g = Graph()
g.add_chain(source, transform, load)
g.run()
print(g.output)

`yieldgraph.graph.Graph()` ¶

Bases: LoggingBehavior

Directed graph of :class:~yieldgraph.node.Node objects forming an ETL pipeline.

Nodes are added via :meth:add_chain and executed in insertion order by :meth:run. The graph manages all :class:~yieldgraph.edge.Edge queues between nodes and exposes the final outputs through :attr:output.

ATTRIBUTE	DESCRIPTION
`nodes`	Ordered mapping of node name → :class:`~yieldgraph.node.Node`. TYPE: `dict[str, Node]`
`edges`	All edge queues keyed by the name of the source node (or :data:`~yieldgraph.config.START_NODE_NAME` for the pipeline entry). TYPE: `defaultdict[str, list[Edge]]`
`terminal_nodes`	Names of the last node in each chain; their edges feed :attr:`output`. TYPE: `list[str]`
`cancelled`	Set to `True` to request cancellation of a running pipeline. Also set automatically on :exc:`KeyboardInterrupt`. TYPE: `bool`
`finished`	`True` once :meth:`run` has completed (successfully or not). TYPE: `bool`
`error`	Non-empty string describing the first error that halted the run. TYPE: `str`
`labels`	Optional display labels keyed by node name, used by :attr:`step`. TYPE: `dict[str, str]`
`observer`	Optional observer that receives callbacks at the start and end of each node and at the start and end of the whole run. Assign any :class:`GraphObserver` subclass instance before calling :meth:`run`. TYPE: `GraphObserver or None`

Source code in src/yieldgraph/graph.py

def __init__(self) -> None:
    self.nodes = {}
    self.edges = defaultdict(list)
    self.terminal_nodes = []
    self._output = []
    self._current_node_name = ''
    self._node_index = 0
    self._threaded = False
    self.cancelled = False
    self.finished = False
    self.error = ''
    self.labels = {}
    self.observer: GraphObserver | None = None

`nodes = {}` `instance-attribute` ¶

Ordered mapping of node name → :class:~yieldgraph.node.Node.

Nodes are inserted in the order :meth:add_chain is called and executed in that same order by :meth:run.

`edges = defaultdict(list)` `instance-attribute` ¶

All edge queues, keyed by the name of the source node.

The special key :data:~yieldgraph.config.START_NODE_NAME holds the seed edges for chains that do not attach to an existing node.

`terminal_nodes = []` `instance-attribute` ¶

Names of the last node in each chain.

Their outgoing edges are collected into :attr:output after the run.

`cancelled = False` `instance-attribute` ¶

Set to True to request early termination of a running pipeline.

Checked by each node before and after every yielded value. Also set automatically when a :exc:KeyboardInterrupt is raised inside a node.

`finished = False` `instance-attribute` ¶

True once :meth:run has returned, regardless of outcome.

`error = ''` `instance-attribute` ¶

Non-empty description of the first error that halted the run.

Empty string when the run succeeded. Check :attr:succeeded as a convenient boolean alternative.

`labels = {}` `instance-attribute` ¶

Optional display labels keyed by node name.

Used by :attr:step and :meth:_node_label to produce human-readable step names for progress displays. If a node name is absent the label is derived automatically by uppercasing each _-separated token.

`observer = None` `instance-attribute` ¶

Optional :class:GraphObserver instance.

Assign before calling :meth:run to receive push callbacks at the start/end of each node and at the start/end of the whole run::

g.observer = MyObserver()
g.run()

None by default (no callbacks fired).

`succeeded` `property` ¶

True if the run completed without errors (read-only).

Both :attr:finished must be True and :attr:error must be empty for this to return True.

`has_output` `property` ¶

True if the run finished and produced at least one output item (read-only).

`current_node` `property` ¶

The :class:~yieldgraph.node.Node currently being executed (read-only).

`output` `property` ¶

Flat list of all output tuples from the terminal nodes (read-only).

Collected lazily on first access after :meth:run completes. Each item is a tuple that was yielded by the last node of a chain.

g.run()
for row in g.output:
    print(row)

`at_first_node` `property` ¶

True when :attr:current_node is the very first node added to the graph (read-only).

`step` `property` ¶

Human-readable label of the currently executing node (read-only).

Returns an empty string when cancelled, 'ETL Prozess Ende' when finished, or the node label during execution. Intended for progress displays.

`progress` `property` ¶

Progress as a float between 0.0 and 1.0 (read-only).

Returns 0.0 when cancelled, 1.0 when finished, or the average progress of all nodes currently in progress. Each node's progress is weighted equally.

`add_chain(*job_functions, labels=(), initial_input=(), attach_to='')` ¶

Add an ordered sequence of callables to the graph as a chain.

Each function's outputs are automatically routed as inputs to the next function in the sequence. If attach_to is given the chain is spliced in as a parallel branch starting from the named node's output edge.

PARAMETER	DESCRIPTION
`*job_functions`	The callables that form the chain, in execution order. TYPE: `Callable` DEFAULT: `()`
`labels`	Human-readable display labels for each node, in the same order as job_functions. Defaults to auto-derived labels. TYPE: `tuple[str, ...]` DEFAULT: `()`
`initial_input`	Seed arguments passed to the first node of a new chain (i.e. when attach_to is not given). The graph instance is prepended automatically if not already present. TYPE: `tuple` DEFAULT: `()`
`attach_to`	Name of an existing node whose output edge this chain should tap into, creating a parallel branch. TYPE: `str` DEFAULT: `''`

Notes

The first function of a new (non-attached) chain receives the graph instance as its first argument so it can inspect :attr:output, react to :attr:cancelled, etc.

Examples:

g = Graph()
g.add_chain(source, transform, load)

# parallel branch from transform onwards
g.add_chain(alternate_load, attach_to='transform')

Source code in src/yieldgraph/graph.py

def add_chain(
        self,
        *job_functions: Callable,
        labels: Tuple[str, ...] = (),
        initial_input: Tuple[Any, ...] = (),
        attach_to: str = '') -> None:
    """Add an ordered sequence of callables to the graph as a chain.

    Each function's outputs are automatically routed as inputs to
    the next function in the sequence. If *attach_to* is given the
    chain is spliced in as a parallel branch starting from the named
    node's output edge.

    Parameters
    ----------
    *job_functions : Callable
        The callables that form the chain, in execution order.
    labels : tuple[str, ...], optional
        Human-readable display labels for each node, in the same 
        order as *job_functions*.  Defaults to auto-derived labels.
    initial_input : tuple, optional
        Seed arguments passed to the first node of a *new* chain 
        (i.e. when *attach_to* is not given). The graph instance is 
        prepended automatically if not already present.
    attach_to : str, optional
        Name of an existing node whose output edge this chain should
        tap into, creating a parallel branch.

    Notes
    -----
    The first function of a new (non-attached) chain receives the
    graph instance as its first argument so it can inspect
    :attr:`output`, react to :attr:`cancelled`, etc.

    Examples
    --------
    ```python
    g = Graph()
    g.add_chain(source, transform, load)

    # parallel branch from transform onwards
    g.add_chain(alternate_load, attach_to='transform')
    ```
    """
    n_nodes = len(job_functions)
    if not n_nodes:
        return

    if not attach_to:
        if self not in initial_input:
            initial_input = (self,) + initial_input
        self.edges[START_NODE_NAME].append(Edge([initial_input]))
        self.log_trace(f'Initial input = {initial_input}')

    edge_in = attach_to if attach_to else START_NODE_NAME
    if not labels:
        labels = tuple('' for _ in job_functions)

    for i, (fn, label) in enumerate(zip(job_functions, labels)):
        node = Node(
            graph=self,
            job_function=fn,
            inputs_from=edge_in,
            label=label,
            first=(i == 0),
            last=(i == n_nodes - 1),
        )
        self.nodes[node.name] = node
        if edge_in == attach_to:
            self.edges[attach_to].append(Edge())
        if node.last:
            self.terminal_nodes.append(node.name)
        self.edges[node.name].append(Edge())
        self.log_trace(f'Added node {node}')
        edge_in = node.name

`run()` ¶

Execute all nodes — sequentially or in parallel threads.

Reads :attr:ENV.THREADED to decide the execution mode, then delegates to :meth:_run_sequential or :meth:_run_threaded. Sets :attr:finished to True when done regardless of outcome.

RAISES	DESCRIPTION
`Nothing — node-level exceptions are stored in`
:attr:`~yieldgraph.node.Node.errors`. Graph-level errors are
recorded in :attr:`error`.

Source code in src/yieldgraph/graph.py

def run(self) -> None:
    """Execute all nodes — sequentially or in parallel threads.

    Reads :attr:`ENV.THREADED` to decide the execution mode, then
    delegates to :meth:`_run_sequential` or :meth:`_run_threaded`.
    Sets :attr:`finished` to ``True`` when done regardless of
    outcome.

    Raises
    ------
    Nothing — node-level exceptions are stored in
    :attr:`~yieldgraph.node.Node.errors`.  Graph-level errors are
    recorded in :attr:`error`.
    """
    self._reset()
    if self._threaded:
        self._run_threaded()
    else:
        self._run_sequential()
    self.finished = True
    if self.observer is not None:
        self.observer.on_run_end(self.succeeded, self.error)
    self.log_info(
        'ETL process done:\n' + '\n'.join(repr(n)
        for n in self.nodes.values()))

Node¶

A single processing step managed internally by Graph. You rarely instantiate Node directly — use Graph.add_chain instead. The attributes and properties are useful for inspecting pipeline state during or after a run.

# After g.run(), inspect each node:
for name, node in g.nodes.items():
    print(f"{name}: consumed={node.n_consumed}, produced={node.n_produced}, errors={node.n_errors}")

`yieldgraph.node.Node(graph, job_function, inputs_from, label='', first=True, last=False)` ¶

Bases: LoggingBehavior

A single processing step in an ETL pipeline graph.

Each :class:Node wraps one :class:~yieldgraph.job.Job and is responsible for:

pulling items from its incoming :class:~yieldgraph.edge.Edge queues,
feeding each item as *args into the job,
fanning every yielded result to all outgoing edges,
tracking counts (n_consumed, n_produced) and caught errors,
cooperating with the owning :class:~yieldgraph.graph.Graph on cancellation.

PARAMETER	DESCRIPTION
`graph`	The owning graph. The node reads `graph.cancel` to decide whether to skip or abort jobs. TYPE: `Graph`
`job_function`	The callable that does the actual work. May be a plain function or a generator function. TYPE: `Callable`
`inputs_from`	Name of the upstream node (or :data:`~yieldgraph.config.START_NODE_NAME` for the first node) whose edges feed this node. TYPE: `str`
`label`	Human-readable display name passed through to the underlying :class:`~yieldgraph.job.Job`. TYPE: `str` DEFAULT: `''`
`first`	Mark this node as the first in its chain, by default `True`. TYPE: `bool` DEFAULT: `True`
`last`	Mark this node as the last in its chain, by default `False`. TYPE: `bool` DEFAULT: `False`

ATTRIBUTE	DESCRIPTION
`inputs_from`	Name of the upstream node that feeds this node. TYPE: `str`
`first`	`True` for the first node of a chain. TYPE: `bool`
`last`	`True` for the last node of a chain. TYPE: `bool`
`n_consumed`	Number of input items consumed so far in the current run. TYPE: `int`
`n_produced`	Number of output items produced so far in the current run. TYPE: `int`
`errors`	Exceptions caught during :meth:`_run_one` in the current run. TYPE: `list[Exception]`
`info`	Free-form string available for progress displays or annotations. TYPE: `str`

Source code in src/yieldgraph/node.py

def __init__(
        self,
        graph,
        job_function: Callable,
        inputs_from: str,
        label: str = '',
        first: bool = True,
        last: bool = False,
        ) -> None:
    self._graph = graph
    self._job = Job(job_function, label)
    self.inputs_from = inputs_from
    self.first = first
    self.last = last
    self.reset()

`n_queued` `instance-attribute` ¶

Total number of input items queued at the start of :meth:process.

`n_consumed` `instance-attribute` ¶

Number of input items pulled from the active edge so far.

`n_produced` `instance-attribute` ¶

Number of output tuples pushed to outgoing edges so far.

`errors` `instance-attribute` ¶

List of exceptions caught during :meth:_run_one in the current run.

`outputs` `instance-attribute` ¶

List of outgoing edges.

`info` `instance-attribute` ¶

Free-form string available for progress displays or annotations.

`inputs_from = inputs_from` `instance-attribute` ¶

Name of the upstream node that feeds this node.

`first = first` `instance-attribute` ¶

True for the first node of a chain.

`last = last` `instance-attribute` ¶

True for the last node of a chain.

`name` `property` ¶

Function name of the wrapped job (read-only).

Derived from job_function.__name__ at construction time.

`n_errors` `property` ¶

Number of exceptions caught so far in the current run (read-only).

`inputs` `property` `writable` ¶

Active input edge (read-only getter).

Returns the first non-empty edge from :attr:_inputs, as tracked by :attr:_active_edge_index. Returns an empty :class:~yieldgraph.edge.Edge when no inputs have been set.

The setter accepts a list of edges and automatically selects the first non-empty one.

`input_count` `property` ¶

Number of items available on the active input edge (read-only).

`output_count` `property` ¶

Number of items sitting on the first output edge (read-only).

`outputs_empty` `property` ¶

True if every outgoing edge is empty (read-only).

`progress` `property` ¶

Fraction of queued items processed, clamped to [0.01, 1.0] (read-only).

Returns 0.5 when :attr:n_queued is zero (e.g. a source node that generates its own data rather than receiving inputs).

`reset()` ¶

Clear all counters, errors, and edge references for a new run.

Called automatically from :meth:__init__ and by :class:~yieldgraph.graph.Graph before each run.

Source code in src/yieldgraph/node.py

def reset(self) -> None:
    """Clear all counters, errors, and edge references for a new 
    run.

    Called automatically from :meth:`__init__` and by
    :class:`~yieldgraph.graph.Graph` before each run.
    """
    self.n_queued = 0
    self.n_consumed = 0
    self.n_produced = 0
    self.errors = []
    self._inputs = []
    self._active_edge_index = 0
    self.outputs = []
    self.info = ''
    self._processing_first = False
    self._processing_last = False

`process(edges_in, edges_out)` ¶

Consume all items from edges_in and push results to edges_out.

Iterates over every queued input item and calls :meth:_run_one for each one. Sets :attr:_processing_first and :attr:_processing_last flags so the job function can branch on position if needed.

PARAMETER	DESCRIPTION
`edges_in`	Incoming edge queues to drain. TYPE: `list[Edge]`
`edges_out`	Outgoing edge queues to fill with results. TYPE: `list[Edge]`

Source code in src/yieldgraph/node.py

def process(self, edges_in: List[Edge], edges_out: List[Edge]) -> None:
    """Consume all items from *edges_in* and push results to 
    *edges_out*.

    Iterates over every queued input item and calls :meth:`_run_one`
    for each one.  Sets :attr:`_processing_first` and
    :attr:`_processing_last` flags so the job function can branch on
    position if needed.

    Parameters
    ----------
    edges_in : list[Edge]
        Incoming edge queues to drain.
    edges_out : list[Edge]
        Outgoing edge queues to fill with results.
    """
    self.log_info(f'{self.input_count} jobs for node "{self._job.label}"')
    self.inputs = edges_in
    self.outputs = edges_out
    self.n_queued = deepcopy(self.input_count)
    self._job.cancelled = False
    for job_count in range(1, self.n_queued + 1):
        self._processing_first = job_count == 1
        self._processing_last = job_count == self.n_queued
        self._run_one(self.inputs.popleft())

`process_streaming(edges_in, edges_out)` ¶

Consume items from edges_in using blocking :meth:~yieldgraph.edge.Edge.get calls.

Used in threaded execution mode where items arrive incrementally from a concurrently-running producer node. Loops until every input edge is :attr:~yieldgraph.edge.Edge.closed and empty, or until the owning graph signals cancellation.

Each input edge is drained fully before the next one is started (same ordering as the sequential :meth:process). :attr:_processing_first is set for the very first item; :attr:_processing_last is always False because the end of the stream is not known in advance.

PARAMETER	DESCRIPTION
`edges_in`	Incoming edge queues to read from (blocking). TYPE: `list[Edge]`
`edges_out`	Outgoing edge queues to push results into. TYPE: `list[Edge]`

Notes

For fan-out pipelines with multiple chains attached to the same node each input edge is consumed by exactly one downstream node in sequential mode (first-non-empty assignment). In threaded mode the caller (:meth:~yieldgraph.graph.Graph._run_threaded) is responsible for passing the correct subset of edges_in to each node.

Source code in src/yieldgraph/node.py

def process_streaming(self, edges_in: List[Edge], edges_out: List[Edge]) -> None:
    """Consume items from *edges_in* using blocking 
    :meth:`~yieldgraph.edge.Edge.get` calls.

    Used in threaded execution mode where items arrive incrementally
    from a concurrently-running producer node.  Loops until every 
    input edge is :attr:`~yieldgraph.edge.Edge.closed` and empty, or 
    until the owning graph signals cancellation.

    Each input edge is drained fully before the next one is started
    (same ordering as the sequential :meth:`process`).
    :attr:`_processing_first` is set for the very first item;
    :attr:`_processing_last` is always ``False`` because the end of
    the stream is not known in advance.

    Parameters
    ----------
    edges_in : list[Edge]
        Incoming edge queues to read from (blocking).
    edges_out : list[Edge]
        Outgoing edge queues to push results into.

    Notes
    -----
    For fan-out pipelines with multiple chains attached to the same
    node each input edge is consumed by exactly one downstream node
    in sequential mode (first-non-empty assignment).  In threaded
    mode the caller (:meth:`~yieldgraph.graph.Graph._run_threaded`)
    is responsible for passing the correct subset of *edges_in* to
    each node.
    """
    self.inputs = edges_in
    self.outputs = edges_out
    self._job.cancelled = False
    first_item = True

    for edge in edges_in:
        while True:
            if self._graph.cancelled:
                self._job.cancelled = True
                return

            item = edge.get(timeout=0.05)

            if item is None:
                if edge.closed:
                    break        # this edge exhausted — move to next
                continue         # timeout — loop back to check cancel

            self._processing_first = first_item
            self._processing_last = False
            first_item = False
            self._run_one(item)

Job¶

The execution wrapper around a callable. Normalises plain functions and generators into a uniform, cancellable generator protocol.

from yieldgraph import Job

def process(x):
    return x * 2

job = Job(process, label="Double It")
results = list(job(21))   # → [42]
print(job.label)          # → 'Double It'

`yieldgraph.job.Job(function, label='')` ¶

Wraps a callable for interruptible, generator-driven execution.

:class:Job is the execution unit used by :class:~yieldgraph.node.Node. It normalises any callable into a generator, adds a per-yield cancellation gate, and exposes a simple cancelled flag that external code (e.g. the owning :class:~yieldgraph.graph.Graph) can set to stop a long-running job gracefully.

PARAMETER	DESCRIPTION
`function`	The callable to wrap. May be a plain function or a generator function. TYPE: `Callable`
`label`	Human-readable display name. When omitted the label is derived automatically from function.name by splitting on `_` and uppercasing each part (e.g. `"load_data"` → `"LOAD DATA"`). TYPE: `str` DEFAULT: `''`

ATTRIBUTE	DESCRIPTION
`name`	Raw function name (`function.__name__`). TYPE: `str`
`cancelled`	Set to `True` to request early termination. The runner checks this flag after every yielded value and exits on the next iteration. Reset to `False` automatically on each new :meth:`__call__`. TYPE: `bool`

Examples:

Wrapping a plain function:

job = Job(lambda x: x ** 2)
assert list(job(4)) == [16]

Wrapping a generator:

def letters():
    yield from 'abc'

job = Job(letters, label='Letters')
assert list(job()) == ['a', 'b', 'c']
assert job.label == 'Letters'

Auto-derived label:

job = Job(lambda: None)
job.name = 'extract_raw_data'    # simulate a named function
assert job.label == 'EXTRACT RAW DATA'  # after label is cleared

Source code in src/yieldgraph/job.py

def __init__(self, function: Callable, label: str = '') -> None:
    self.name = function.__name__
    self._label = label
    self._wrapped = _wrap(function, self)

`cancelled = False` `class-attribute` `instance-attribute` ¶

Cancellation request flag.

Set to True from outside the generator loop (e.g. from the owning graph) to stop iteration after the current yield. Cleared automatically at the start of every new call so that a :class:Job instance can be reused across pipeline runs.

`running` `property` ¶

True while the generator is actively yielding (read-only).

Managed internally by :func:_wrap: set to True when the runner starts and to False when it finishes or is cancelled. Safe to poll from another thread to check job status.

`label` `property` ¶

Human-readable display name for this job (read-only).

Returns the value passed as label to :meth:__init__. If none was given, the label is derived lazily from :attr:name by splitting on _ and uppercasing each token:

job = Job(load_raw_data)
job.label   # → 'LOAD RAW DATA'

`name = function.name` `instance-attribute` ¶

Raw name of the wrapped function (function.__name__).

Edge¶

A directed queue that carries data tuples between nodes. Subclasses collections.deque for sequential use; adds thread-safe put / get / close for threaded pipelines.

from yieldgraph import Edge

# Sequential mode
e = Edge()
e.append((1, 2))
print(e.popleft())   # → (1, 2)

# Threaded mode
e2 = Edge()
e2.put(("hello",))
print(e2.get())      # → ('hello',)
e2.close()
print(e2.get())      # → None  (closed + empty)

`yieldgraph.edge.Edge(iterable=())` ¶

Bases: deque

A directed queue connecting two :class:~yieldgraph.node.Node objects.

Subclasses :class:~collections.deque so it can be used as a plain list-like queue in sequential mode. Thread-safe :meth:put, :meth:get, and :meth:close are provided for pipelined (threaded) execution, where producer and consumer nodes run concurrently.

ATTRIBUTE	DESCRIPTION
`closed`	`True` once the producer has called :meth:`close`. No further items will be added. A consumer blocked in :meth:`get` will be woken up and will drain any remaining items before returning `None`. TYPE: `bool`

Examples:

Sequential mode (existing interface unchanged):

e = Edge()
e.append((1, 2))
item = e.popleft()   # → (1, 2)

Threaded mode:

import threading
e = Edge()

def producer():
    for i in range(3):
        e.put((i,))
    e.close()

def consumer():
    while True:
        item = e.get()
        if item is None:   # closed + empty
            break
        print(item)

t1 = threading.Thread(target=producer)
t2 = threading.Thread(target=consumer)
t1.start(); t2.start()
t1.join();  t2.join()

Source code in src/yieldgraph/edge.py

def __init__(self, iterable=()) -> None:
    super().__init__(iterable)
    self._cond: threading.Condition = threading.Condition(threading.Lock())
    self._closed: bool = False

`closed` `property` ¶

True once the producer has signalled end-of-stream via :meth:close.

`put(item)` ¶

Append item to the right end of the queue and notify waiters.

Thread-safe alternative to :meth:~collections.deque.append. Any consumer blocked in :meth:get will be woken up immediately.

PARAMETER	DESCRIPTION
`item`	The data tuple to enqueue. TYPE: `Any`

Source code in src/yieldgraph/edge.py

def put(self, item: Any) -> None:
    """Append *item* to the right end of the queue and notify 
    waiters.

    Thread-safe alternative to :meth:`~collections.deque.append`.
    Any consumer blocked in :meth:`get` will be woken up 
    immediately.

    Parameters
    ----------
    item : Any
        The data tuple to enqueue.
    """
    with self._cond:
        self.append(item)
        self._cond.notify()

`get(timeout=None)` ¶

Remove and return the leftmost item, blocking until one is available.

Blocks when the queue is empty and the edge is not yet :attr:closed. Returns None when the edge is closed and empty (producer done, all items consumed).

PARAMETER	DESCRIPTION
`timeout`	Maximum seconds to wait. When the timeout expires without an item, returns `None` regardless of :attr:`closed`. Omit to wait indefinitely. TYPE: `float` DEFAULT: `None`

RETURNS	DESCRIPTION
`Any`	The next item from the left end, or `None` when the edge is closed and empty (or the timeout elapsed).

Examples:

edge = Edge()
edge.put((42,))
assert edge.get() == (42,)
edge.close()
assert edge.get() is None   # closed + empty → None

Source code in src/yieldgraph/edge.py

def get(self, timeout: Optional[float] = None) -> Any:
    """Remove and return the leftmost item, blocking until one is 
    available.

    Blocks when the queue is empty and the edge is not yet 
    :attr:`closed`. Returns ``None`` when the edge is closed *and* 
    empty (producer done, all items consumed).

    Parameters
    ----------
    timeout : float, optional
        Maximum seconds to wait.  When the timeout expires without 
        an item, returns ``None`` regardless of :attr:`closed`. 
        Omit to wait indefinitely.

    Returns
    -------
    Any
        The next item from the left end, or ``None`` when the edge 
        is closed and empty (or the timeout elapsed).

    Examples
    --------
    ```python
    edge = Edge()
    edge.put((42,))
    assert edge.get() == (42,)
    edge.close()
    assert edge.get() is None   # closed + empty → None
    ```
    """
    with self._cond:
        deadline = (time.monotonic() + timeout) if timeout is not None else None
        while not self and not self._closed:
            if deadline is not None:
                remaining = deadline - time.monotonic()
                if remaining <= 0:
                    return None
                self._cond.wait(remaining)
            else:
                self._cond.wait()
        if self:
            return self.popleft()
        return None  # closed + empty

`close()` ¶

Signal that no more items will be added to this edge.

Sets :attr:closed to True and wakes all threads blocked in :meth:get. They will drain any remaining items and then receive None. Calling :meth:close more than once is harmless.

Source code in src/yieldgraph/edge.py

def close(self) -> None:
    """Signal that no more items will be added to this edge.

    Sets :attr:`closed` to ``True`` and wakes all threads blocked in
    :meth:`get`.  They will drain any remaining items and then 
    receive ``None``.  Calling :meth:`close` more than once is 
    harmless.
    """
    with self._cond:
        self._closed = True
        self._cond.notify_all()

Config¶

Module-level constants and the LoggingBehavior mixin used throughout yieldgraph.

from yieldgraph import LOG, ENV, LoggingBehavior, START_NODE_NAME

# Log level constants
print(LOG.INFO)           # 'INFO'
print(LOG.TRACE_LEVEL) # 5

# Environment variable names
print(ENV.THREADED)       # 'YIELDGRAPH_THREADED'

# Sentinel for the implicit start node
print(START_NODE_NAME)    # '__START__'

`yieldgraph.config` ¶

Configuration module for the yieldgraph library.

This module defines constants and classes related to environment configuration and logging behavior in the yieldgraph library. It includes the :data:ENV instance of :class:_ENV_, which exposes the current values of environment variables as boolean properties; the :data:LOG instance of :class:_Log_, which contains log level constants; and the :class:LoggingBehavior mixin class that provides logging capabilities to classes that inherit from it.

The LoggingBehavior class defines a :attr:logger attribute that can be used by classes that inherit from it to log messages at various log levels. The log levels are defined in the LOG instance, and include TRACE, DEBUG, INFO, WARNING, ERROR, and CRITICAL. Whether traceback information is included in error log messages is controlled by the YIELDGRAPH_LOG_TRACEBACK environment variable via :attr:ENV.LOG_TRACEBACK. All logging can be suppressed entirely by setting the YIELDGRAPH_LOG_DISABLED environment variable to a truthy value, which causes every call to :meth:LoggingBehavior.log to return immediately without emitting anything.

The LoggingBehavior class provides a :attr:log_title property that can be overridden by classes that inherit from it to provide a custom log title for log messages. If not overridden, the log title defaults to the class name. The :attr:log method is used to log messages with a specified log level, and there are convenience methods for logging at specific levels (e.g., :attr:log_info, :attr:log_warning, etc.) that call the :attr:log method with the appropriate log level. This allows classes that inherit from LoggingBehavior to easily log messages with consistent formatting and log levels throughout the yieldgraph library.

`START_NODE_NAME = 'START'` `module-attribute` ¶

Sentinel name for the implicit start node that seeds pipeline chains.

`ENV = _ENV_()` `module-attribute` ¶

Instance of :class:_ENV_ that provides the current values of environment variables used to configure the yieldgraph library.

Properties are evaluated at access time, so changes to environment variables are immediately reflected. Use the _KEY class attributes (e.g. :attr:~_ENV_.THREADED_KEY, :attr:~_ENV_.LOG_TRACEBACK_KEY, :attr:~_ENV_.LOG_DISABLED_KEY) to access the raw key names when manipulating os.environ directly (e.g., in tests).

Quick reference:

+------------------------------+---------------------------+----------------------------------------+ | Property | Key constant | Effect when truthy | +==============================+===========================+========================================+ | :attr:ENV.THREADED | ENV.THREADED_KEY | Run graph nodes in threads | +------------------------------+---------------------------+----------------------------------------+ | :attr:ENV.LOG_TRACEBACK | ENV.LOG_TRACEBACK_KEY | Include tracebacks in error messages | +------------------------------+---------------------------+----------------------------------------+ | :attr:ENV.LOG_DISABLED | ENV.LOG_DISABLED_KEY | Suppress all logging output | +------------------------------+---------------------------+----------------------------------------+

`LOG = _Log_()` `module-attribute` ¶

Instance of the _Log_ class containing log level constants and configuration for logging behavior.

This instance can be used throughout the yieldgraph library to access log level constants (e.g., LOG.DEBUG, LOG.INFO, etc.). Use :attr:ENV.LOG_TRACEBACK to check whether traceback information should be included in log output.

The LoggingBehavior mixin class uses the LOG instance to determine log levels when logging messages. This allows for consistent log level references across the library and makes it easy to change log level values in one place if needed.

To access log level numbers by name, you can use dictionary-like access (e.g., LOG['DEBUG']), and to access log level names by number, you can call the instance (e.g., LOG(10) returns 'DEBUG').

`LoggingBehavior` ¶

Mixin class that provides logging capabilities to classes that inherit from it.

This class defines a :attr:logger attribute that can be used to log messages. It also provides a :attr:log_title property that can be used to customize the log title for the class. The :attr:log method is used to log messages with a specified log level, and there are convenience methods for logging at specific levels (e.g., :attr:log_info, :attr:log_warning, etc.).

To use this mixin, simply inherit from it in your class and use the logging methods as needed. You can also override the :attr:log_title property to provide a custom log title. If you do not override the :attr:log_title property, it will default to the class name, which will be included in the log messages.

Examples:

class MyClass(LoggingBehavior):
    def do_something(self):
        self.log_info('Doing something...')
my_instance = MyClass()
my_instance.do_something()
# Output: MyClass: Doing something... (logged at INFO level)

`logger = logger` `class-attribute` `instance-attribute` ¶

Logger instance to be used for logging. By default, it uses the logger from the loguru library if available, otherwise it uses the standard logging library's logger.

`log_title` `property` ¶

Get log title for class (read-only).

By default, it returns the class name. You can override this property in your class to provide a custom log title.

`log(message, level, exception=None)` ¶

Log a message with the given log level.

Does nothing if :attr:ENV.LOG_DISABLED is True (i.e. YIELDGRAPH_LOG_DISABLED is set to a truthy value).

This method uses the :attr:log_title property to include the class name in the log message. The log level can be specified as an integer or a string. If the log level is a string, it is converted to the corresponding integer value.

PARAMETER	DESCRIPTION
`message`	The message to log. TYPE: `str`
`level`	The log level, either as an integer or a string. TYPE: `int \| str`
`exception`	An optional exception to include in the log message. If provided, it will be included in the log output. TYPE: `Optional[Exception]` DEFAULT: `None`