add python generator

dcrpcgen/main.py

@@ -8,6 +8,7 @@
 
 from ._version import __version__
 from .java import java_cmd
+from .python import python_cmd
 
 
 def add(
@@ -64,7 +65,7 @@ def get_parser() -> argparse.ArgumentParser:
     parser.add_argument("-v", "--version", action="version", version=__version__)
     subparsers = parser.add_subparsers(title="subcommands")
 
-    for generator in [java_cmd]:
+    for generator in [java_cmd, python_cmd]:
         add(subparsers, base, generator)
 
     return parser

dcrpcgen/python/__init__.py

@@ -0,0 +1,106 @@
+"""Python code generation"""
+
+from argparse import Namespace
+from pathlib import Path
+from typing import Any
+
+from ..utils import camel2snake
+from .templates import get_template
+from .types import decode_type, generate_types
+from .utils import create_comment
+
+
+def python_cmd(args: Namespace) -> None:
+    """Generate JSON-RPC client for the Python programming language"""
+    root_folder = Path(args.folder)
+    root_folder.mkdir(parents=True, exist_ok=True)
+
+    path = root_folder / "types.py"
+    print(f"Generating {path}")
+    generate_types(path, args.openrpc_spec["components"]["schemas"])
+
+    path = root_folder / "rpc.py"
+    print(f"Generating {path}")
+    generate_methods(path, args.openrpc_spec["methods"])
+
+    path = root_folder / "transport.py"
+    print(f"Generating {path}")
+    generate_transport(path)
+
+    path = root_folder / "_utils.py"
+    print(f"Generating {path}")
+    generate_utils(path)
+
+
+def generate_methods(path: Path, methods: dict[str, Any]) -> str:
+    """Generate Rpc class"""
+    with path.open("w", encoding="utf-8") as output:
+        template = get_template("rpc.py.j2")
+        output.write(
+            template.render(
+                methods=methods,
+                generate_method=generate_method,
+            )
+        )
+
+
+def generate_method(method: dict[str, Any]) -> str:
+    """Generate a rpc module"""
+    assert method["paramStructure"] == "by-position"
+    params = method["params"]
+    params_types = {param["name"]: decode_type(param["schema"]) for param in params}
+    result_type = decode_type(method["result"]["schema"])
+    name = method["name"]
+    tab = "    "
+    text = ""
+    text += f"{tab}def {name}("
+    text += ", ".join(
+        f'{camel2snake(param["name"])}: {params_types[param["name"]]}'
+        for param in params
+    )
+    text += f") -> {result_type}:\n"
+    if "description" in method:
+        text += create_comment(method["description"], tab * 2, docstr=True)
+
+    args = [f'"{name}"']
+    for param in params:
+        param_name = camel2snake(param["name"])
+        if params_types[param["name"]] in (
+            "bool",
+            "int",
+            "float",
+            "str",
+            "Optional[bool]",
+            "Optional[int]",
+            "Optional[float]",
+            "Optional[str]",
+            "list[bool]",
+            "list[int]",
+            "list[float]",
+            "list[str]",
+            "dict[Any, Optional[str]]",
+            "dict[Any, str]",
+            "tuple[float, float]",
+        ):
+            args.append(param_name)
+        else:
+            args.append(f"_wrap({param_name})")
+
+    rtn = "" if result_type == "None" else "return "
+    stmt = f'transport.call({", ".join(args)})'
+    text += f"{tab*2}{rtn}{stmt}\n"
+    return text
+
+
+def generate_transport(path: Path) -> str:
+    """Generate transport module"""
+    with path.open("w", encoding="utf-8") as output:
+        template = get_template("transport.py.j2")
+        output.write(template.render())
+
+
+def generate_utils(path: Path) -> str:
+    """Generate utils module"""
+    with path.open("w", encoding="utf-8") as output:
+        template = get_template("utils.py.j2")
+        output.write(template.render())

dcrpcgen/python/templates/NormalClass.py.j2

@@ -0,0 +1,4 @@
+@dataclass(kw_only=True)
+class {{ name }}:
+{% if "description" in schema %}{{ create_comment(schema["description"], "    ", docstr=True) + "\n" }}{% endif -%}
+{{ generate_properties(schema["properties"], False) -}}

dcrpcgen/python/templates/StrEnumTemplate.py.j2

@@ -0,0 +1,10 @@
+class {{ name }}(StrEnum):
+{% for schema in schemas -%}
+  {% if "description" in schema %}
+    {{- create_comment(schema["description"], "    ").rstrip() }}
+  {%- endif %}
+  {%- for val in schema["enum"] %}
+    {{ camel2snake(val) | upper }} = "{{ val }}"
+  {%- endfor %}
+
+{% endfor -%}

dcrpcgen/python/templates/UnionType.py.j2

@@ -0,0 +1,5 @@
+{% for typ in schema["oneOf"] %}
+  {{- generate_subtype(typ, name) }}
+{% endfor -%}
+{% if "description" in schema %}{{ create_comment(schema["description"]) + "\n" }}{% endif -%}
+{{ name }}: TypeAlias = {{ get_union_type(schema["oneOf"], name) }}

dcrpcgen/python/templates/__init__.py

@@ -0,0 +1,12 @@
+"""Code templates"""
+
+from jinja2 import Environment, PackageLoader, Template
+
+env = Environment(
+    loader=PackageLoader(__name__.rsplit(".", maxsplit=1)[0], "templates")
+)
+
+
+def get_template(name: str) -> Template:
+    """Load the template with the given filename"""
+    return env.get_template(name)

dcrpcgen/python/templates/rpc.py.j2

@@ -0,0 +1,29 @@
+"""JSON-RPC API definition."""
+
+import dataclasses
+from typing import Any, Optional
+
+from .transport import RpcTransport
+from .types import *
+
+
+class Rpc:
+    """Access to the chatmail JSON-RPC API."""
+
+    def __init__(self, transport: RpcTransport) -> None:
+        self.transport = transport
+
+{% for method in methods %}
+  {{- generate_method(method)}}
+{% endfor -%}
+
+
+def _wrap(arg: Any) -> Any:
+    if isinstance(arg, dict):
+        return {key: _wrap(val) for key, val in arg.items()}
+    if isinstance(arg, list):
+        return [_wrap(elem) for elem in arg]
+
+    if dataclasses.is_dataclass(arg):
+        return snake2camel(dataclasses.asdict(arg))
+    return arg

dcrpcgen/python/templates/transport.py.j2

@@ -0,0 +1,153 @@
+"""JSON-RPC transports to communicate with chatmail core."""
+
+import itertools
+import json
+import logging
+import os
+import subprocess
+import sys
+from abc import ABC, abstractmethod
+from queue import Queue
+from threading import Event, Thread
+from typing import Any, Dict, Iterator, Optional
+
+from ._utils import to_attrdict
+
+
+class JsonRpcError(Exception):
+    """An error occurred in your request to the JSON-RPC API."""
+
+
+class RpcTransport(ABC):
+    """Chatmail RPC client's transport."""
+
+    @abstractmethod
+    def call(self, method: str, *args) -> Any:
+        """Request the RPC server to call a function and return its return value if any."""
+
+
+class _Result(Event):
+    def __init__(self) -> None:
+        self._value: Any = None
+        super().__init__()
+
+    def set(self, value: Any) -> None:  # noqa
+        self._value = value
+        super().set()
+
+    def wait(self) -> Any:  # noqa
+        super().wait()
+        return self._value
+
+
+class IOTransport:
+    """Chatmail RPC transport over IO using external deltachat-rpc-server program."""
+
+    def __init__(self, accounts_dir: Optional[str] = None, rpc_executable: str = "deltachat-rpc-server", **kwargs):
+        """The given arguments will be passed to subprocess.Popen()"""
+        self.logger = logging.getLogger("deltachat2.IOTransport")
+        if accounts_dir:
+            kwargs["env"] = {
+                **kwargs.get("env", os.environ),
+                "DC_ACCOUNTS_PATH": str(accounts_dir),
+            }
+        self.rpc_executable = rpc_executable
+        self._kwargs = kwargs
+        self.process: subprocess.Popen
+        self.id_iterator: Iterator[int]
+        # Map from request ID to the result.
+        self.pending_results: Dict[int, _Result]
+        self.request_queue: Queue
+        self.closing: bool
+        self.reader_thread: Thread
+        self.writer_thread: Thread
+
+    def start(self) -> None:
+        """Start the RPC server process."""
+        if sys.version_info >= (3, 11):
+            # Prevent subprocess from capturing SIGINT.
+            kwargs = {"process_group": 0, **self._kwargs}
+        else:
+            # `process_group` is not supported before Python 3.11.
+            kwargs = {"preexec_fn": os.setpgrp, **self._kwargs}  # noqa: PLW1509
+        self.process = subprocess.Popen(  # noqa: R1732
+            self.rpc_executable,
+            stdin=subprocess.PIPE,
+            stdout=subprocess.PIPE,
+            **kwargs,
+        )
+        self.id_iterator = itertools.count(start=1)
+        self.pending_results = {}
+        self.request_queue = Queue()
+        self.closing = False
+        self.reader_thread = Thread(target=self._reader_loop)
+        self.reader_thread.start()
+        self.writer_thread = Thread(target=self._writer_loop)
+        self.writer_thread.start()
+
+    def close(self) -> None:
+        """Terminate RPC server process and wait until the reader loop finishes."""
+        self.closing = True
+        self.call("stop_io_for_all_accounts")
+        assert self.process.stdin
+        self.process.stdin.close()
+        self.reader_thread.join()
+        self.request_queue.put(None)
+        self.writer_thread.join()
+
+    def __enter__(self):
+        self.start()
+        return self
+
+    def __exit__(self, _exc_type, _exc, _tb):
+        self.close()
+
+    def _reader_loop(self) -> None:
+        try:
+            assert self.process.stdout
+            while True:
+                line = self.process.stdout.readline()
+                if not line:  # EOF
+                    break
+                response = json.loads(line)
+                if "id" in response:
+                    self.pending_results.pop(response["id"]).set(response)
+                else:
+                    self.logger.warning("Got a response without ID: %s", response)
+        except Exception:
+            # Log an exception if the reader loop dies.
+            self.logger.exception("Exception in the reader loop")
+
+    def _writer_loop(self) -> None:
+        """Writer loop ensuring only a single thread writes requests."""
+        try:
+            assert self.process.stdin
+            while True:
+                request = self.request_queue.get()
+                if not request:
+                    break
+                data = (json.dumps(request) + "\n").encode()
+                self.process.stdin.write(data)
+                self.process.stdin.flush()
+        except Exception:
+            # Log an exception if the writer loop dies.
+            self.logger.exception("Exception in the writer loop")
+
+    def call(self, method: str, *args) -> Any:
+        """Request the RPC server to call a function and return its return value if any."""
+        request_id = next(self.id_iterator)
+        request = {
+            "jsonrpc": "2.0",
+            "method": method,
+            "params": args,
+            "id": request_id,
+        }
+        result = self.pending_results[request_id] = _Result()
+        self.request_queue.put(request)
+        response = result.wait()
+
+        if "error" in response:
+            raise JsonRpcError(response["error"])
+        if "result" in response:
+            return to_attrdict(response["result"])
+        return None