Source code for edri.api.grpc.export

"""Render edri.api.grpc.schema's generated FileDescriptorProto to .proto3
source text, for consumers who want a static `.proto` file to run through
their own language's codegen instead of pointing it at the reflection
service. Pure text generation over the FileDescriptorProto's own fields -
no DescriptorPool/message_factory/live message classes involved (unlike
edri/api/grpc/runtime.py), so this direction doesn't touch anything beyond
the plain (proto2-defined) descriptor_pb2 shape schema.py already builds.

Usage: ``python -m edri.api.grpc.export [output_path]`` - writes to
`output_path` if given, otherwise to stdout.
"""
from __future__ import annotations

import sys

from google.protobuf import descriptor_pb2

from edri.api.grpc.schema import build_schema

__all__ = ["render_proto"]

_FD = descriptor_pb2.FieldDescriptorProto

_SCALAR_KEYWORDS = {
    _FD.TYPE_STRING: "string",
    _FD.TYPE_INT32: "int32",
    _FD.TYPE_INT64: "int64",
    _FD.TYPE_UINT32: "uint32",
    _FD.TYPE_UINT64: "uint64",
    _FD.TYPE_DOUBLE: "double",
    _FD.TYPE_FLOAT: "float",
    _FD.TYPE_BOOL: "bool",
    _FD.TYPE_BYTES: "bytes",
}


def _unqualified(type_name: str) -> str:
    """".package.Message" (FileDescriptorProto's fully-qualified form) ->
    "Message" - every message schema.py generates lives in one flat file
    (no message is physically nested inside another), so an unqualified
    reference is always enough within that one file."""
    return type_name.rsplit(".", 1)[-1]


def _field_type_name(field: descriptor_pb2.FieldDescriptorProto) -> str:
    if field.type in (_FD.TYPE_MESSAGE, _FD.TYPE_ENUM):
        return _unqualified(field.type_name)
    return _SCALAR_KEYWORDS[field.type]


def _render_message(message: descriptor_pb2.DescriptorProto) -> list[str]:
    lines = [f"message {message.name} {{"]

    # schema.py's _add_field() gives every proto3-`optional` field its own
    # synthetic one-field oneof named "_<field>" (the standard proto3
    # encoding of `optional`) - rendered as a plain `optional` field, the
    # same way protoc itself prints that construct back, not as its own
    # explicit oneof block. Only a "real", multi-field oneof (currently:
    # ClientMessage/ServerMessage's "event" oneof) gets a rendered block.
    synthetic_oneofs = {i for i, o in enumerate(message.oneof_decl) if o.name.startswith("_")}

    for field in message.field:
        if field.HasField("oneof_index") and field.oneof_index not in synthetic_oneofs:
            continue  # rendered below, inside its oneof block
        label = ""
        if field.label == _FD.LABEL_REPEATED:
            label = "repeated "
        elif field.proto3_optional:
            label = "optional "
        lines.append(f"    {label}{_field_type_name(field)} {field.name} = {field.number};")

    for index, oneof in enumerate(message.oneof_decl):
        if index in synthetic_oneofs:
            continue
        lines.append(f"    oneof {oneof.name} {{")
        for field in message.field:
            if field.HasField("oneof_index") and field.oneof_index == index:
                lines.append(f"        {_field_type_name(field)} {field.name} = {field.number};")
        lines.append("    }")

    lines.append("}")
    return lines


def _render_service(service: descriptor_pb2.ServiceDescriptorProto) -> list[str]:
    lines = [f"service {service.name} {{"]
    for method in service.method:
        client_stream = "stream " if method.client_streaming else ""
        server_stream = "stream " if method.server_streaming else ""
        input_type = _unqualified(method.input_type)
        output_type = _unqualified(method.output_type)
        lines.append(f"    rpc {method.name} ({client_stream}{input_type}) returns ({server_stream}{output_type});")
    lines.append("}")
    return lines




def render_proto(file_proto: descriptor_pb2.FileDescriptorProto) -> str:
    """Render a FileDescriptorProto (as produced by
    edri.api.grpc.schema.build_schema()) to .proto3 source text."""
    lines = ['syntax = "proto3";', "", f"package {file_proto.package};", ""]

    for dependency in file_proto.dependency:
        lines.append(f'import "{dependency}";')
    if file_proto.dependency:
        lines.append("")

    for message in file_proto.message_type:
        lines.extend(_render_message(message))
        lines.append("")

    for service in file_proto.service:
        lines.extend(_render_service(service))
        lines.append("")

    return "\n".join(lines).rstrip() + "\n"



def main(argv: list[str] | None = None) -> None:
    argv = sys.argv[1:] if argv is None else argv
    text = render_proto(build_schema().file_descriptor_proto)
    if argv:
        with open(argv[0], "w") as out:
            out.write(text)
    else:
        sys.stdout.write(text)


if __name__ == "__main__":
    main()