type_bridge.generator.render.functions

functions

Render function definitions from parsed schema.

FunctionParamContext dataclass

FunctionParamContext(name, py_name, type_hint, typedb_type)

Context for a single function parameter.

FunctionContext dataclass

FunctionContext(name, py_name, params, return_types, is_stream, docstring, param_signature='', return_hint='')

Context for rendering a single function.

render_functions

render_functions(schema)

Render the complete functions module.

Generates Python function wrappers that return FunctionQuery objects for each TypeDB function defined in the schema.

Parameters:

Name	Type	Description	Default
schema	ParsedSchema	Parsed schema containing function definitions	required

Returns:

Type	Description
str	Complete Python source code for functions.py, or empty string if no functions

Source code in type_bridge/generator/render/functions.py

def render_functions(schema: ParsedSchema) -> str:
    """Render the complete functions module.

    Generates Python function wrappers that return FunctionQuery objects
    for each TypeDB function defined in the schema.

    Args:
        schema: Parsed schema containing function definitions

    Returns:
        Complete Python source code for functions.py, or empty string if no functions
    """
    if not schema.functions:
        return ""

    logger.debug(f"Rendering {len(schema.functions)} function wrappers")

    contexts = []
    all_names = []
    for name, spec in sorted(schema.functions.items()):
        py_name = to_python_name(name)
        all_names.append(py_name)
        contexts.append(_build_function_context(name, spec))

    imports = _get_required_imports(contexts)
    datetime_imports = sorted(imports & {"datetime", "date"})
    has_decimal = "Decimal" in imports
    has_duration = "Duration" in imports

    # Generate the source code directly (template is simple enough)
    lines = [
        '"""Function wrappers generated from a TypeDB schema.',
        "",
        "These functions return FunctionQuery objects that can generate TypeQL queries",
        "for calling the corresponding TypeDB schema functions.",
        '"""',
        "",
        "from __future__ import annotations",
        "",
    ]

    # Add imports
    if datetime_imports:
        lines.append(f"from datetime import {', '.join(datetime_imports)}")
    if has_decimal:
        lines.append("from decimal import Decimal")
    if has_duration:
        lines.append("from isodate import Duration")
    lines.append("from typing import Iterator")
    lines.append("")
    lines.append("from type_bridge.expressions import FunctionQuery, ReturnType")
    lines.append("")
    lines.append("")

    # Generate each function
    for ctx in contexts:
        # Function signature
        if ctx.param_signature:
            lines.append(f"def {ctx.py_name}({ctx.param_signature}) -> {ctx.return_hint}:")
        else:
            lines.append(f"def {ctx.py_name}() -> {ctx.return_hint}:")

        # Docstring
        if ctx.docstring:
            lines.append(f'    """{ctx.docstring}')
        else:
            lines.append(f'    """Call TypeDB function `{ctx.name}`.')

        # Add return type info to docstring
        stream_info = "stream of " if ctx.is_stream else ""
        type_info = ", ".join(ctx.return_types)
        lines.append("")
        lines.append(f"    Returns: {stream_info}{type_info}")
        lines.append('    """')

        # Build args list
        if ctx.params:
            args_items = [f'("{p.name}", {p.py_name})' for p in ctx.params]
            args_str = f"[{', '.join(args_items)}]"
        else:
            args_str = "[]"

        # Build return type
        return_types_str = ", ".join(f'"{t}"' for t in ctx.return_types)
        is_stream_str = "True" if ctx.is_stream else "False"

        lines.append("    return FunctionQuery(")
        lines.append(f'        name="{ctx.name}",')
        lines.append(f"        args={args_str},")
        lines.append(
            f"        return_type=ReturnType([{return_types_str}], is_stream={is_stream_str}),"
        )
        lines.append("    )")
        lines.append("")
        lines.append("")

    # Generate __all__
    lines.append("__all__ = [")
    for name in sorted(all_names):
        lines.append(f'    "{name}",')
    lines.append("]")
    lines.append("")

    result = "\n".join(lines)
    logger.info(f"Rendered {len(all_names)} function wrappers")
    return result