Skip to content

dizzy.generators.projections

dizzy.generators.projections

Projections generator — gen_int/python/projection/ combined context+protocol.

render_projection(proj)

Render gen_int/python/projection/_projection.py.

Source code in dizzy/src/dizzy/generators/projections.py 
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
def render_projection(proj: ProjectionDef) -> str:
    """Render gen_int/python/projection/<proj.name>_projection.py."""
    context_class = f"{proj.name}_context"
    protocol_class = f"{proj.name}_projection"
    event_class = camelcase(proj.event)
    description = proj.description.strip()

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

    context_body: list[str] = []
    adapter_import = ""
    if proj.adapter is not None:
        adapter_cls = _adapter_class_name(proj.adapter)
        adapter_import = f"from gen_int.python.adapters.{proj.adapter} import {adapter_cls}"
        context_body.append(f"    adapter: {adapter_cls}")
    context_body.extend(extras.fields)
    if not context_body:
        context_body.append("    pass")

    imports = [
        "# AUTO-GENERATED — do not edit",
        "from dataclasses import dataclass",
        "from typing import Protocol",
    ]
    if extras.needs_callable:
        imports.append("from typing import Callable")
    imports += [
        "",
        f"from gen_def.pydantic.events import {event_class}",
    ]
    if adapter_import:
        imports.append(adapter_import)
    imports.extend(extras.imports)

    lines = imports + [
        *extras.classes,
        "",
        "",
        "@dataclass",
        f"class {context_class}:",
        *context_body,
    ]

    lines += [
        "",
        "",
        f"class {protocol_class}(Protocol):",
        f'    """{description}"""',
        "",
        f"    def __call__(self, event: {event_class}, context: {context_class}) -> None:",
        '        """Apply the projection — mutate model state in response to the event."""',
        "        ...",
        "",
    ]
    return "\n".join(lines)

write_projection(proj, output_dir)

Write gen_int/python/projection/_projection.py (always overwritten).

Source code in dizzy/src/dizzy/generators/projections.py 
75
76
77
78
79
80
def write_projection(proj: ProjectionDef, output_dir: Path) -> None:
    """Write gen_int/python/projection/<proj.name>_projection.py (always overwritten)."""
    dest = gen_int_root(output_dir) / "python" / "projection" / f"{proj.name}_projection.py"
    dest.parent.mkdir(parents=True, exist_ok=True)
    dest.write_text(render_projection(proj))
    logger.debug("wrote file", extra={"path": str(dest)})

render_src_projection_stub(proj)

Render a src/projection/.py implementation stub.

Source code in dizzy/src/dizzy/generators/projections.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def render_src_projection_stub(proj: ProjectionDef) -> str:
    """Render a src/projection/<proj.name>.py implementation stub."""
    context_class = f"{proj.name}_context"
    protocol_class = f"{proj.name}_projection"
    event_class = camelcase(proj.event)
    lines = [
        "# Implementation stub — fill in your logic here",
        f"from gen_int.python.projection.{proj.name}_projection import {protocol_class}",
        f"from gen_int.python.projection.{proj.name}_projection import {context_class}",
        f"from gen_def.pydantic.events import {event_class}",
        "",
        "",
        f"def {proj.name}(",
        f"    event: {event_class},",
        f"    context: {context_class},",
        ") -> None:",
        "    raise NotImplementedError",
        "",
    ]
    return "\n".join(lines)