Skip to content

dizzy.generators.policies

dizzy.generators.policies

Policies generator — gen_int/python/policy/ context and protocol.

render_policy_context(policy)

Render gen_int/python/policy/_context.py.

Source code in dizzy/src/dizzy/generators/policies.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_policy_context(policy: PolicyDef) -> str:
    """Render gen_int/python/policy/<policy.name>_context.py."""
    emitters_class = f"{policy.name}_emitters"
    queries_class = f"{policy.name}_queries"
    context_class = f"{policy.name}_context"

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

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

    if policy.emits or policy.queries or extras.imports:
        lines.append("")
        for cmd_name in policy.emits or []:
            lines.append(f"from gen_def.pydantic.commands import {camelcase(cmd_name)}")
        for query_name in policy.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 policy.emits:
        for cmd_name in policy.emits:
            lines.append(f"    {cmd_name}: Callable[[{camelcase(cmd_name)}], None]")
    else:
        lines.append("    pass")

    # queries dataclass (only if there are queries)
    if policy.queries:
        lines.append("")
        lines.append("")
        lines.append("@dataclass")
        lines.append(f"class {queries_class}:")
        for query_name in policy.queries or []:
            # Host-bound query closure: the host injects the read adapter, so the
            # policy 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 policy.queries:
        lines.append(f"    query: {queries_class}")
    lines.extend(extras.fields)

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

write_policy_context(policy, output_dir)

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

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

render_policy_protocol(policy)

Render gen_int/python/policy/_protocol.py.

Source code in dizzy/src/dizzy/generators/policies.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_policy_protocol(policy: PolicyDef) -> str:
    """Render gen_int/python/policy/<policy.name>_protocol.py."""
    context_class = f"{policy.name}_context"
    protocol_class = f"{policy.name}_protocol"
    event_class = camelcase(policy.event)
    description = (policy.description or "").strip()

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

write_policy_protocol(policy, output_dir)

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

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

render_src_policy_stub(policy)

Render a src/policy/.py implementation stub.

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