refactor: move code generator to tools
This commit is contained in:
Vlad Stan
@@ -0,0 +1,310 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
from collections import defaultdict
|
||||
from collections.abc import Sequence
|
||||
from pathlib import Path
|
||||
from types import UnionType
|
||||
from typing import Any, Literal, Union, get_args, get_origin
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
from lnbits.core.extensions import (
|
||||
ExtensionAPI,
|
||||
ExtensionAPIMethod,
|
||||
get_extension_api_method,
|
||||
list_extension_api_methods,
|
||||
)
|
||||
|
||||
|
||||
def generate_typescript_sdk(
|
||||
api_cls: type[ExtensionAPI] | None = None,
|
||||
method_ids: Sequence[str] | None = None,
|
||||
) -> str:
|
||||
api_cls = api_cls or ExtensionAPI
|
||||
methods = _select_methods(api_cls, method_ids)
|
||||
models = _collect_models(methods)
|
||||
|
||||
lines = [
|
||||
"/* Generated by LNbits ExtensionAPI codegen. */",
|
||||
"/* Do not edit by hand. */",
|
||||
"",
|
||||
"export type MaybePromise<T> = T | Promise<T>",
|
||||
"",
|
||||
]
|
||||
|
||||
for model in models:
|
||||
lines.extend(_render_model_type(model))
|
||||
lines.append("")
|
||||
|
||||
lines.extend(_render_method_metadata(methods))
|
||||
lines.append("")
|
||||
lines.extend(_render_host_type(methods))
|
||||
lines.append("")
|
||||
lines.extend(_render_sdk_type(methods))
|
||||
lines.append("")
|
||||
|
||||
lines.extend(_render_create_sdk(methods))
|
||||
|
||||
return "\n".join(lines).rstrip() + "\n"
|
||||
|
||||
|
||||
def write_typescript_sdk(
|
||||
path: str | Path,
|
||||
api_cls: type[ExtensionAPI] | None = None,
|
||||
method_ids: Sequence[str] | None = None,
|
||||
) -> None:
|
||||
Path(path).write_text(
|
||||
generate_typescript_sdk(api_cls, method_ids), encoding="utf-8"
|
||||
)
|
||||
|
||||
|
||||
def _select_methods(
|
||||
api_cls: type[ExtensionAPI], method_ids: Sequence[str] | None
|
||||
) -> list[ExtensionAPIMethod]:
|
||||
if not method_ids:
|
||||
return list_extension_api_methods(api_cls)
|
||||
return [get_extension_api_method(method_id, api_cls) for method_id in method_ids]
|
||||
|
||||
|
||||
def _collect_models(methods: Sequence[ExtensionAPIMethod]) -> list[type[BaseModel]]:
|
||||
models: dict[str, type[BaseModel]] = {}
|
||||
pending = [
|
||||
model
|
||||
for method in methods
|
||||
for model in (method.request_model, method.response_model)
|
||||
]
|
||||
|
||||
while pending:
|
||||
model = pending.pop()
|
||||
if model.__name__ in models:
|
||||
continue
|
||||
models[model.__name__] = model
|
||||
for field in model.__fields__.values():
|
||||
pending.extend(_nested_model_types(field.outer_type_))
|
||||
return [models[name] for name in sorted(models)]
|
||||
|
||||
|
||||
def _nested_model_types(type_: Any) -> list[type[BaseModel]]:
|
||||
models: list[type[BaseModel]] = []
|
||||
if _is_model_type(type_):
|
||||
models.append(type_)
|
||||
for arg in get_args(type_):
|
||||
models.extend(_nested_model_types(arg))
|
||||
return models
|
||||
|
||||
|
||||
def _render_model_type(model: type[BaseModel]) -> list[str]:
|
||||
name = _model_name(model)
|
||||
fields = model.__fields__
|
||||
if not fields:
|
||||
return [f"export type {name} = Record<string, never>"]
|
||||
|
||||
lines = [f"export type {name} = {{"]
|
||||
for field_name, field in fields.items():
|
||||
optional = "?" if not field.required else ""
|
||||
ts_type = _python_type_to_ts(field.outer_type_, field.allow_none)
|
||||
lines.append(f" {_camel(field_name)}{optional}: {ts_type}")
|
||||
lines.append("}")
|
||||
return lines
|
||||
|
||||
|
||||
def _python_type_to_ts(type_: Any, allow_none: bool = False) -> str:
|
||||
origin = get_origin(type_)
|
||||
args = get_args(type_)
|
||||
|
||||
if origin in (UnionType, Union):
|
||||
ts = " | ".join(
|
||||
_python_type_to_ts(arg) for arg in args if arg is not type(None)
|
||||
)
|
||||
if type(None) in args:
|
||||
ts = f"{ts} | null"
|
||||
return ts
|
||||
|
||||
if origin is Literal:
|
||||
return " | ".join(_literal_to_ts(arg) for arg in args)
|
||||
|
||||
if _is_model_type(type_):
|
||||
ts = _model_name(type_)
|
||||
elif origin in (list, Sequence):
|
||||
item_type = _python_type_to_ts(args[0]) if args else "unknown"
|
||||
ts = f"{item_type}[]"
|
||||
elif origin is dict:
|
||||
key_type = _python_type_to_ts(args[0]) if args else "string"
|
||||
value_type = _python_type_to_ts(args[1]) if len(args) > 1 else "unknown"
|
||||
ts = (
|
||||
f"Record<{key_type}, {value_type}>"
|
||||
if key_type == "string"
|
||||
else f"{{ [key: string]: {value_type} }}"
|
||||
)
|
||||
elif _is_subclass(type_, str):
|
||||
ts = "string"
|
||||
elif _is_subclass(type_, bool):
|
||||
ts = "boolean"
|
||||
elif _is_subclass(type_, int) or _is_subclass(type_, float):
|
||||
ts = "number"
|
||||
else:
|
||||
ts = "unknown"
|
||||
|
||||
return f"{ts} | null" if allow_none else ts
|
||||
|
||||
|
||||
def _literal_to_ts(value: Any) -> str:
|
||||
if isinstance(value, str):
|
||||
return f'"{value}"'
|
||||
if isinstance(value, bool):
|
||||
return "true" if value else "false"
|
||||
if value is None:
|
||||
return "null"
|
||||
return str(value)
|
||||
|
||||
|
||||
def _is_subclass(type_: Any, class_: type) -> bool:
|
||||
try:
|
||||
return isinstance(type_, type) and issubclass(type_, class_)
|
||||
except TypeError:
|
||||
return False
|
||||
|
||||
|
||||
def _is_model_type(type_: Any) -> bool:
|
||||
return _is_subclass(type_, BaseModel)
|
||||
|
||||
|
||||
def _render_method_metadata(
|
||||
methods: Sequence[ExtensionAPIMethod],
|
||||
) -> list[str]:
|
||||
lines = ["export const extensionApiMethods = ["]
|
||||
for method in methods:
|
||||
permission = (
|
||||
f'"{method.required_permission}"' if method.required_permission else "null"
|
||||
)
|
||||
lines.extend(
|
||||
[
|
||||
" {",
|
||||
f' id: "{method.method_id}",',
|
||||
f' namespace: "{method.namespace}",',
|
||||
f' sdkName: "{method.sdk_name}",',
|
||||
f' pythonName: "{method.python_name}",',
|
||||
f' hostName: "{method.host_name}",',
|
||||
f' hostJsName: "{_camel(method.host_name)}",',
|
||||
f" requiredPermission: {permission},",
|
||||
" },",
|
||||
]
|
||||
)
|
||||
lines.append("] as const")
|
||||
return lines
|
||||
|
||||
|
||||
def _render_host_type(methods: Sequence[ExtensionAPIMethod]) -> list[str]:
|
||||
lines = ["export type ExtensionHost = {"]
|
||||
for method in sorted(methods, key=lambda item: item.host_name):
|
||||
request = _model_name(method.request_model)
|
||||
response = _model_name(method.response_model)
|
||||
if _is_empty_model(method.request_model):
|
||||
lines.append(f" {_camel(method.host_name)}(): MaybePromise<{response}>")
|
||||
else:
|
||||
lines.append(
|
||||
f" {_camel(method.host_name)}"
|
||||
f"(input: {request}): MaybePromise<{response}>"
|
||||
)
|
||||
lines.append("}")
|
||||
return lines
|
||||
|
||||
|
||||
def _render_sdk_type(methods: Sequence[ExtensionAPIMethod]) -> list[str]:
|
||||
namespaces = _methods_by_namespace(methods)
|
||||
lines = ["export type ExtensionSdk = {"]
|
||||
for namespace, namespace_methods in namespaces.items():
|
||||
lines.append(f" {namespace}: {{")
|
||||
for method in namespace_methods:
|
||||
request = _model_name(method.request_model)
|
||||
response = _model_name(method.response_model)
|
||||
if _is_empty_model(method.request_model):
|
||||
lines.append(f" {method.sdk_name}(): Promise<{response}>")
|
||||
else:
|
||||
lines.append(
|
||||
f" {method.sdk_name}(input: {request}): Promise<{response}>"
|
||||
)
|
||||
lines.append(" }")
|
||||
lines.append("}")
|
||||
return lines
|
||||
|
||||
|
||||
def _render_create_sdk(methods: Sequence[ExtensionAPIMethod]) -> list[str]:
|
||||
namespaces = _methods_by_namespace(methods)
|
||||
lines = [
|
||||
"export function createExtensionSdk(",
|
||||
" host: ExtensionHost",
|
||||
"): ExtensionSdk {",
|
||||
" return {",
|
||||
]
|
||||
for namespace, namespace_methods in namespaces.items():
|
||||
lines.append(f" {namespace}: {{")
|
||||
for method in namespace_methods:
|
||||
host_name = _camel(method.host_name)
|
||||
if _is_empty_model(method.request_model):
|
||||
signature = f"{method.sdk_name}()"
|
||||
host_call = f"host.{host_name}()"
|
||||
else:
|
||||
signature = f"{method.sdk_name}(input)"
|
||||
host_call = f"host.{host_name}(input)"
|
||||
lines.extend(
|
||||
[
|
||||
f" async {signature} {{",
|
||||
f" return {host_call}",
|
||||
" },",
|
||||
]
|
||||
)
|
||||
lines.append(" },")
|
||||
lines.extend([" }", "}"])
|
||||
return lines
|
||||
|
||||
|
||||
def _methods_by_namespace(
|
||||
methods: Sequence[ExtensionAPIMethod],
|
||||
) -> dict[str, list[ExtensionAPIMethod]]:
|
||||
namespaces: dict[str, list[ExtensionAPIMethod]] = defaultdict(list)
|
||||
for method in sorted(methods, key=lambda item: (item.namespace, item.sdk_name)):
|
||||
namespaces[method.namespace].append(method)
|
||||
return dict(sorted(namespaces.items()))
|
||||
|
||||
|
||||
def _model_name(model: type[BaseModel]) -> str:
|
||||
return model.__name__
|
||||
|
||||
|
||||
def _camel(value: str) -> str:
|
||||
head, *tail = value.split("_")
|
||||
return head + "".join(part.capitalize() for part in tail)
|
||||
|
||||
|
||||
def _is_empty_model(model: type[BaseModel]) -> bool:
|
||||
return not model.__fields__
|
||||
|
||||
|
||||
def main(argv: Sequence[str] | None = None) -> int:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Generate a TypeScript SDK from the LNbits ExtensionAPI."
|
||||
)
|
||||
parser.add_argument(
|
||||
"--method",
|
||||
action="append",
|
||||
dest="method_ids",
|
||||
help="ExtensionAPI method id to include. Can be passed multiple times.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--out",
|
||||
help="Output file. If omitted, the generated SDK is printed to stdout.",
|
||||
)
|
||||
args = parser.parse_args(argv)
|
||||
|
||||
sdk = generate_typescript_sdk(method_ids=args.method_ids)
|
||||
if args.out:
|
||||
Path(args.out).write_text(sdk, encoding="utf-8")
|
||||
else:
|
||||
print(sdk, end="")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user