Skip to content

dizzy.generators.queries

dizzy.generators.queries

Queries generator — scaffold def/queries/ stubs and gen_int/python/query/ protocols.

render_scaffold_query(query)

Render a combined LinkML stub for def/queries/.yaml.

Source code in dizzy/src/dizzy/generators/queries.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def render_scaffold_query(query: QueryDef) -> str:
    """Render a combined LinkML stub for def/queries/<query.name>.yaml."""
    camel = _to_camel(query.name)
    lines = [
        f"id: https://example.org/queries/{query.name}",
        f"name: {query.name}",
        *description_lines(query.description, ""),
        "prefixes:",
        "  linkml: https://w3id.org/linkml/",
        "default_range: string",
        "imports:",
        "  - linkml:types",
        "classes:",
        f"  {camel}Input:",
        f"    description: Input for {query.name}",
        "    attributes: {}",
        f"  {camel}Output:",
        f"    description: Output for {query.name}",
        "    attributes: {}",
        "",
    ]
    return "\n".join(lines)

write_scaffold_query(query, output_dir)

Write def/queries/.yaml; skip if file already exists.

Source code in dizzy/src/dizzy/generators/queries.py 
41
42
43
44
45
46
47
48
49
def write_scaffold_query(query: QueryDef, output_dir: Path) -> None:
    """Write def/queries/<query.name>.yaml; skip if file already exists."""
    dest = output_dir / "def" / "queries" / f"{query.name}.yaml"
    if dest.exists():
        logger.debug("skipped existing file", extra={"path": str(dest)})
        return
    dest.parent.mkdir(parents=True, exist_ok=True)
    dest.write_text(render_scaffold_query(query))
    logger.debug("wrote file", extra={"path": str(dest)})

render_gen_query_protocol(query)

Render gen_int/python/query/.py — Protocol + context dataclass.

Source code in dizzy/src/dizzy/generators/queries.py 
 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
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def render_gen_query_protocol(query: QueryDef) -> str:
    """Render gen_int/python/query/<query.name>.py — Protocol + context dataclass."""
    camel = _to_camel(query.name)
    input_class = f"{camel}Input"
    output_class = f"{camel}Output"
    context_class = f"{query.name}_context"
    protocol_class = f"{query.name}_query"

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

    context_body: list[str] = []
    adapter_import = ""
    if query.adapter is not None:
        adapter_cls = _adapter_class_name(query.adapter)
        adapter_import = f"from gen_int.python.adapters.{query.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.query.{query.name} import {input_class}, {output_class}",
    ]
    if adapter_import:
        imports.append(adapter_import)
    imports.extend(extras.imports)

    lines = imports + [
        *extras.classes,
        "",
        "",
        "@dataclass",
        f"class {context_class}:",
        *context_body,
        "",
        "",
        f"class {protocol_class}(Protocol):",
        f'    """{query.description}"""',
        "",
        "    def __call__(",
        f"        self, input: {input_class}, context: {context_class}",
        f"    ) -> {output_class}:",
        "        ...",
        "",
    ]
    return "\n".join(lines)

write_gen_query_protocol(query, output_dir)

Write gen_int/python/query/.py (always overwritten).

Source code in dizzy/src/dizzy/generators/queries.py 
113
114
115
116
117
118
def write_gen_query_protocol(query: QueryDef, output_dir: Path) -> None:
    """Write gen_int/python/query/<query.name>.py (always overwritten)."""
    dest = gen_int_root(output_dir) / "python" / "query" / f"{query.name}.py"
    dest.parent.mkdir(parents=True, exist_ok=True)
    dest.write_text(render_gen_query_protocol(query))
    logger.debug("wrote file", extra={"path": str(dest)})

render_src_query_stub(query_name)

Render a src/query/.py implementation stub.

Source code in dizzy/src/dizzy/generators/queries.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def render_src_query_stub(query_name: str) -> str:
    """Render a src/query/<query_name>.py implementation stub."""
    camel = _to_camel(query_name)
    input_class = f"{camel}Input"
    output_class = f"{camel}Output"
    protocol_class = f"{query_name}_query"
    context_class = f"{query_name}_context"
    lines = [
        "# Implementation stub — fill in your logic here",
        f"from gen_int.python.query.{query_name} import {protocol_class}, {context_class}",
        f"from gen_def.pydantic.query.{query_name} import {input_class}, {output_class}",
        "",
        "",
        f"def {query_name}(input: {input_class}, context: {context_class}) -> {output_class}:",
        "    raise NotImplementedError",
        "",
    ]
    return "\n".join(lines)