Skip to content

compiled

compiled #

@cg.procedure — compile a body function into a procedure graph.

The decorated function records its structure by constructing nodes against the schema root: navigate on root, cache with |= / &=, add side-effect steps. There is no global state — a RootContext collects the body (run inside a procedure_sequence) and reduces it to one root ProcedureSequence. GetRoot is built inside that sequence with the target schema; if the body never navigates it, it is a dropped no-op rather than a recorded step.

CompiledProcedure #

A @cg.procedure-decorated function; build it to the node graph + schema.

Source code in capturegraph-lib/capturegraph/procedures/procedure/authoring/compiled.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
class CompiledProcedure:
    """A ``@cg.procedure``-decorated function; build it to the node graph + schema."""

    def __init__(
        self,
        schema: type,
        fn: Callable[..., None],
        name: str,
        label: str,
    ) -> None:
        """Wrap a recorded body ``fn`` with the ``schema`` it navigates, its name, and label."""
        self.schema = schema
        self.fn = fn
        self.name = name
        self.label = label
        self.__name__ = name
        self.__doc__ = fn.__doc__

    def build(self) -> Procedure[CGType]:
        """Run the body once and return the root ``ProcedureSequence``."""
        from capturegraph.procedures.nodes.destination.access import (
            GetRoot,
        )
        from capturegraph.procedures.procedure.authoring.contexts.root import (
            RootContext,
        )
        from capturegraph.procedures.procedure.authoring.contexts.sequence import (
            procedure_sequence,
        )

        context = RootContext()
        with context.collect():
            with procedure_sequence():
                self.fn(GetRoot(schema=self.schema))

        return context.root().set_label(self.label)

    def to_dict(self) -> dict:
        """The full wire form ``{schema, root, nodes}``."""
        from capturegraph.procedures.procedure.exporting.serializer import (
            procedure_to_dict,
        )

        return procedure_to_dict(self.build(), target_schema=self.schema)

    def to_json(
        self,
        indent: int = 4,
        **json_kwargs: Unpack[_JSONDumpsKwargs],
    ) -> str:
        """The full wire form as a JSON string."""
        from capturegraph.procedures.procedure.exporting.serializer import (
            procedure_to_json,
        )

        return procedure_to_json(
            self.build(),
            indent=indent,
            target_schema=self.schema,
            **json_kwargs,
        )

__init__(schema, fn, name, label) #

Wrap a recorded body fn with the schema it navigates, its name, and label.

Source code in capturegraph-lib/capturegraph/procedures/procedure/authoring/compiled.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def __init__(
    self,
    schema: type,
    fn: Callable[..., None],
    name: str,
    label: str,
) -> None:
    """Wrap a recorded body ``fn`` with the ``schema`` it navigates, its name, and label."""
    self.schema = schema
    self.fn = fn
    self.name = name
    self.label = label
    self.__name__ = name
    self.__doc__ = fn.__doc__

build() #

Run the body once and return the root ProcedureSequence.

Source code in capturegraph-lib/capturegraph/procedures/procedure/authoring/compiled.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def build(self) -> Procedure[CGType]:
    """Run the body once and return the root ``ProcedureSequence``."""
    from capturegraph.procedures.nodes.destination.access import (
        GetRoot,
    )
    from capturegraph.procedures.procedure.authoring.contexts.root import (
        RootContext,
    )
    from capturegraph.procedures.procedure.authoring.contexts.sequence import (
        procedure_sequence,
    )

    context = RootContext()
    with context.collect():
        with procedure_sequence():
            self.fn(GetRoot(schema=self.schema))

    return context.root().set_label(self.label)

to_dict() #

The full wire form {schema, root, nodes}.

Source code in capturegraph-lib/capturegraph/procedures/procedure/authoring/compiled.py 
67
68
69
70
71
72
73
def to_dict(self) -> dict:
    """The full wire form ``{schema, root, nodes}``."""
    from capturegraph.procedures.procedure.exporting.serializer import (
        procedure_to_dict,
    )

    return procedure_to_dict(self.build(), target_schema=self.schema)

to_json(indent=4, **json_kwargs) #

The full wire form as a JSON string.

Source code in capturegraph-lib/capturegraph/procedures/procedure/authoring/compiled.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def to_json(
    self,
    indent: int = 4,
    **json_kwargs: Unpack[_JSONDumpsKwargs],
) -> str:
    """The full wire form as a JSON string."""
    from capturegraph.procedures.procedure.exporting.serializer import (
        procedure_to_json,
    )

    return procedure_to_json(
        self.build(),
        indent=indent,
        target_schema=self.schema,
        **json_kwargs,
    )

procedure(schema, *, label=None) #

Decorate def f(root: cg.Procedure[cg.Path[Schema]]) as a procedure.

The decorated function becomes a [CompiledProcedure][]; call .to_json() / .to_dict() / .build() to emit the graph. Inside the body, root is the GetRoot node rooted at Schema; cache with |= / &= and add side-effect steps with cg.do(step).

label is the human-readable name shown in the app; it defaults to the function name, so pass it when that reads poorly (label="Clementine Rating" beats clementine_rating).

Source code in capturegraph-lib/capturegraph/procedures/procedure/authoring/compiled.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def procedure(
    schema: type,
    *,
    label: str | None = None,
) -> Callable[[Callable[..., None]], CompiledProcedure]:
    """Decorate ``def f(root: cg.Procedure[cg.Path[Schema]])`` as a procedure.

    The decorated function becomes a [CompiledProcedure][]; call ``.to_json()``
    / ``.to_dict()`` / ``.build()`` to emit the graph. Inside the body, ``root`` is
    the ``GetRoot`` node rooted at ``Schema``; cache with ``|=`` / ``&=`` and add
    side-effect steps with ``cg.do(step)``.

    ``label`` is the human-readable name shown in the app; it defaults to the
    function name, so pass it when that reads poorly (``label="Clementine Rating"``
    beats ``clementine_rating``).
    """
    from capturegraph.types.containers.struct import (
        Struct,
    )

    if not (isinstance(schema, type) and issubclass(schema, Struct)):
        raise AuthoringError(f"@cg.procedure needs a cg.Struct schema type, got {schema!r}")

    def decorate(fn: Callable[..., None]) -> CompiledProcedure:
        return CompiledProcedure(
            schema=schema,
            fn=fn,
            name=fn.__name__,
            label=label if label is not None else fn.__name__,
        )

    return decorate