Skip to content

dizzy.generators.procedures

dizzy.generators.procedures

Procedures generator — gen_int/python/procedure/ context and protocol.

render_procedure_context(proc)

Render gen_int/python/procedure/_context.py.

Source code in dizzy/src/dizzy/generators/procedures.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
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
def render_procedure_context(proc: ProcedureDef) -> str:
    """Render gen_int/python/procedure/<proc.name>_context.py."""
    emitters_class = f"{proc.name}_emitters"
    queries_class = f"{proc.name}_queries"
    context_class = f"{proc.name}_context"

    extras = render_context_extras(proc.name, proc.environment, proc.telemetry)

    lines = ["# AUTO-GENERATED — do not edit"]
    lines.append("from dataclasses import dataclass")
    if proc.emits or proc.queries or extras.needs_callable:
        lines.append("from typing import Callable")

    if proc.emits or proc.queries or extras.imports:
        lines.append("")
        for event_name in proc.emits or []:
            lines.append(f"from gen_def.pydantic.events import {camelcase(event_name)}")
        for query_name in proc.queries or []:
            lines.append(
                f"from gen_def.pydantic.query.{query_name} import "
                f"{camelcase(query_name)}Input, {camelcase(query_name)}Output"
            )
        lines.extend(extras.imports)

    # emitters dataclass
    lines.append("")
    lines.append("")
    lines.append("@dataclass")
    lines.append(f"class {emitters_class}:")
    if proc.emits:
        for event_name in proc.emits:
            lines.append(f"    {event_name}: Callable[[{camelcase(event_name)}], None]")
    else:
        lines.append("    pass")

    # queries dataclass (only if there are queries)
    if proc.queries:
        lines.append("")
        lines.append("")
        lines.append("@dataclass")
        lines.append(f"class {queries_class}:")
        for query_name in proc.queries:
            # Host-bound query closure: the host injects the read adapter, so the
            # handler calls it with just the query input (symmetric with emit).
            lines.append(
                f"    {query_name}: Callable"
                f"[[{camelcase(query_name)}Input], {camelcase(query_name)}Output]"
            )

    # env / telemetry dataclasses
    lines.extend(extras.classes)

    # context dataclass
    lines.append("")
    lines.append("")
    lines.append("@dataclass")
    lines.append(f"class {context_class}:")
    lines.append(f"    emit: {emitters_class}")
    if proc.queries:
        lines.append(f"    query: {queries_class}")
    lines.extend(extras.fields)

    lines.append("")
    return "\n".join(lines)

write_procedure_context(proc, output_dir)

Write gen_int/python/procedure/_context.py (always overwritten).

Source code in dizzy/src/dizzy/generators/procedures.py 
79
80
81
82
83
84
def write_procedure_context(proc: ProcedureDef, output_dir: Path) -> None:
    """Write gen_int/python/procedure/<proc.name>_context.py (always overwritten)."""
    dest = gen_int_root(output_dir) / "python" / "procedure" / f"{proc.name}_context.py"
    dest.parent.mkdir(parents=True, exist_ok=True)
    dest.write_text(render_procedure_context(proc))
    logger.debug("wrote file", extra={"path": str(dest)})

render_procedure_protocol(proc)

Render gen_int/python/procedure/_protocol.py.

Source code in dizzy/src/dizzy/generators/procedures.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def render_procedure_protocol(proc: ProcedureDef) -> str:
    """Render gen_int/python/procedure/<proc.name>_protocol.py."""
    context_class = f"{proc.name}_context"
    protocol_class = f"{proc.name}_protocol"
    command_class = camelcase(proc.command)
    description = (proc.description or "").strip()

    lines = [
        "# AUTO-GENERATED — do not edit",
        "from typing import Protocol",
        "",
        f"from gen_def.pydantic.commands import {command_class}",
        f"from gen_int.python.procedure.{proc.name}_context import (",
        f"    {context_class},",
        ")",
        "",
        "",
        f"class {protocol_class}(Protocol):",
        f'    """{description}"""',
        "",
        "    def __call__(",
        "        self,",
        f"        context: {context_class},",
        f"        command: {command_class},",
        "    ) -> None:",
        "        ...",
        "",
    ]
    return "\n".join(lines)

write_procedure_protocol(proc, output_dir)

Write gen_int/python/procedure/_protocol.py (always overwritten).

Source code in dizzy/src/dizzy/generators/procedures.py 
118
119
120
121
122
123
def write_procedure_protocol(proc: ProcedureDef, output_dir: Path) -> None:
    """Write gen_int/python/procedure/<proc.name>_protocol.py (always overwritten)."""
    dest = gen_int_root(output_dir) / "python" / "procedure" / f"{proc.name}_protocol.py"
    dest.parent.mkdir(parents=True, exist_ok=True)
    dest.write_text(render_procedure_protocol(proc))
    logger.debug("wrote file", extra={"path": str(dest)})

render_src_procedure_stub(proc)

Render a src/procedure/.py implementation stub.

Source code in dizzy/src/dizzy/generators/procedures.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def render_src_procedure_stub(proc: ProcedureDef) -> str:
    """Render a src/procedure/<proc.name>.py implementation stub."""
    context_class = f"{proc.name}_context"
    protocol_class = f"{proc.name}_protocol"
    command_class = camelcase(proc.command)
    lines = [
        "# Implementation stub — fill in your logic here",
        f"from gen_int.python.procedure.{proc.name}_protocol import {protocol_class}",
        f"from gen_int.python.procedure.{proc.name}_context import {context_class}",
        f"from gen_def.pydantic.commands import {command_class}",
        "",
        "",
        f"def {proc.name}(",
        f"    context: {context_class},",
        f"    command: {command_class},",
        ") -> None:",
        "    raise NotImplementedError",
        "",
    ]
    return "\n".join(lines)