diff --git a/lnbits/core/extensions/__init__.py b/lnbits/core/extensions/__init__.py index 8752368df..8e8fd4e92 100644 --- a/lnbits/core/extensions/__init__.py +++ b/lnbits/core/extensions/__init__.py @@ -1 +1,19 @@ """Extension runtime contracts.""" + +from .api import ( + ExtensionAPI, + ExtensionAPIMethod, + extension_api_contract, + extension_api_method, + get_extension_api_method, + list_extension_api_methods, +) + +__all__ = [ + "ExtensionAPI", + "ExtensionAPIMethod", + "extension_api_contract", + "extension_api_method", + "get_extension_api_method", + "list_extension_api_methods", +] diff --git a/lnbits/core/extensions/api.py b/lnbits/core/extensions/api.py new file mode 100644 index 000000000..a44f066e7 --- /dev/null +++ b/lnbits/core/extensions/api.py @@ -0,0 +1,397 @@ +from __future__ import annotations + +import inspect +import logging +import secrets +import time +from collections.abc import Awaitable, Callable, Iterable +from dataclasses import dataclass +from functools import wraps +from typing import Literal, TypeVar, get_type_hints + +from pydantic import BaseModel, Field + +logger = logging.getLogger("lnbits.extensions") + +_EXTENSION_API_METHOD_ATTR = "__lnbits_extension_api_method__" +_RequestModel = TypeVar("_RequestModel", bound=BaseModel) +_ResponseModel = TypeVar("_ResponseModel", bound=BaseModel) + + +class EmptyRequest(BaseModel): + pass + + +class KvGetRequest(BaseModel): + key: str = Field(..., min_length=1, max_length=512) + + +class KvGetResponse(BaseModel): + value: str | None = None + + +class KvSetRequest(BaseModel): + key: str = Field(..., min_length=1, max_length=512) + value: str = Field(..., max_length=65536) + + +class KvSetResponse(BaseModel): + ok: bool = True + + +class KvListRequest(BaseModel): + prefix: str = Field(..., min_length=1, max_length=512) + + +class KvListResponse(BaseModel): + keys: list[str] = Field(default_factory=list) + + +class CreateInvoiceRequest(BaseModel): + wallet_id: str = Field(..., min_length=1, max_length=128) + amount_sat: int = Field(..., gt=0) + memo: str = Field(..., max_length=512) + tag: str = Field(..., min_length=1, max_length=64) + extra: dict[str, str] = Field(default_factory=dict) + + +class CreateInvoiceResponse(BaseModel): + payment_hash: str + payment_request: str + checking_id: str + + +class WatchPaymentRequest(BaseModel): + payment_hash: str = Field(..., min_length=1, max_length=128) + callback_export: str = Field(..., min_length=1, max_length=128) + + +class WatchPaymentResponse(BaseModel): + ok: bool = True + + +class RandomIdRequest(BaseModel): + prefix: str = Field(..., min_length=1, max_length=32) + + +class RandomIdResponse(BaseModel): + id: str + + +class NowResponse(BaseModel): + timestamp: int + + +class LogRequest(BaseModel): + level: Literal["debug", "info", "warning", "error"] = "info" + message: str = Field(..., min_length=1, max_length=2048) + + +class LogResponse(BaseModel): + ok: bool = True + + +@dataclass(frozen=True) +class ExtensionAPIMethodExport: + method_id: str + namespace: str + name: str + host_name: str + sdk_name: str + description: str + required_permission: str | None = None + public_context: bool = False + + +@dataclass(frozen=True) +class ExtensionAPIMethod: + method_id: str + namespace: str + name: str + python_name: str + host_name: str + sdk_name: str + description: str + request_model: type[BaseModel] + response_model: type[BaseModel] + required_permission: str | None = None + public_context: bool = False + + @property + def sdk_qualified_name(self) -> str: + return f"{self.namespace}.{self.sdk_name}" + + +def extension_api_method( + *, + method_id: str, + namespace: str, + name: str, + host_name: str, + sdk_name: str, + description: str, + required_permission: str | None = None, + public_context: bool = False, +) -> Callable[ + [Callable[["ExtensionAPI", _RequestModel], Awaitable[_ResponseModel]]], + Callable[["ExtensionAPI", _RequestModel], Awaitable[_ResponseModel]], +]: + export = ExtensionAPIMethodExport( + method_id=method_id, + namespace=namespace, + name=name, + host_name=host_name, + sdk_name=sdk_name, + description=description, + required_permission=required_permission, + public_context=public_context, + ) + + def decorator( + function: Callable[["ExtensionAPI", _RequestModel], Awaitable[_ResponseModel]], + ) -> Callable[["ExtensionAPI", _RequestModel], Awaitable[_ResponseModel]]: + @wraps(function) + async def wrapper( + self: "ExtensionAPI", request: _RequestModel + ) -> _ResponseModel: + self.require_permission(required_permission) + return await function(self, request) + + setattr(wrapper, _EXTENSION_API_METHOD_ATTR, export) + return wrapper + + return decorator + + +class ExtensionAPI: + def __init__( + self, + extension_id: str, + permissions: Iterable[str], + *, + user_id: str | None = None, + wallet_id: str | None = None, + ) -> None: + self.extension_id = extension_id + self.permissions = set(permissions) + self.user_id = user_id + self.wallet_id = wallet_id + + def require_permission(self, permission: str | None) -> None: + if permission and permission not in self.permissions: + raise PermissionError( + f"Extension '{self.extension_id}' is missing permission '{permission}'." + ) + + @extension_api_method( + method_id="storage.get", + namespace="storage", + name="Get storage value", + host_name="kv_get", + sdk_name="get", + description="Read one value from the extension storage namespace.", + required_permission="ext.storage.read_write", + public_context=True, + ) + async def storage_get(self, request: KvGetRequest) -> KvGetResponse: + self._raise_unwired_runtime("storage_get") + + @extension_api_method( + method_id="storage.set", + namespace="storage", + name="Set storage value", + host_name="kv_set", + sdk_name="set", + description="Write one value to the extension storage namespace.", + required_permission="ext.storage.read_write", + ) + async def storage_set(self, request: KvSetRequest) -> KvSetResponse: + self._raise_unwired_runtime("storage_set") + + @extension_api_method( + method_id="storage.list", + namespace="storage", + name="List storage keys", + host_name="kv_list", + sdk_name="list", + description="List keys under a prefix in the extension storage namespace.", + required_permission="ext.storage.read_write", + public_context=True, + ) + async def storage_list(self, request: KvListRequest) -> KvListResponse: + self._raise_unwired_runtime("storage_list") + + @extension_api_method( + method_id="wallet.create_invoice", + namespace="wallet", + name="Create invoice", + host_name="create_invoice", + sdk_name="createInvoice", + description="Create an incoming Lightning invoice for an allowed wallet.", + required_permission="wallet.create_invoice", + public_context=True, + ) + async def wallet_create_invoice( + self, request: CreateInvoiceRequest + ) -> CreateInvoiceResponse: + self._raise_unwired_runtime("wallet_create_invoice") + + @extension_api_method( + method_id="payments.watch", + namespace="payments", + name="Watch payment", + host_name="watch_payment", + sdk_name="watch", + description="Subscribe the extension to a payment state callback.", + required_permission="payments.watch", + ) + async def payments_watch(self, request: WatchPaymentRequest) -> WatchPaymentResponse: + self._raise_unwired_runtime("payments_watch") + + @extension_api_method( + method_id="system.random_id", + namespace="system", + name="Random ID", + host_name="random_id", + sdk_name="id", + description="Create a random extension-local identifier.", + public_context=True, + ) + async def system_random_id(self, request: RandomIdRequest) -> RandomIdResponse: + return RandomIdResponse( + id=f"{request.prefix}_{secrets.token_urlsafe(12).replace('-', '_')}" + ) + + @extension_api_method( + method_id="system.now", + namespace="system", + name="Current timestamp", + host_name="now", + sdk_name="now", + description="Return the current Unix timestamp.", + public_context=True, + ) + async def system_now(self, request: EmptyRequest) -> NowResponse: + return NowResponse(timestamp=int(time.time())) + + @extension_api_method( + method_id="system.log", + namespace="system", + name="Log message", + host_name="log", + sdk_name="log", + description="Write a bounded message to the extension log.", + public_context=True, + ) + async def system_log(self, request: LogRequest) -> LogResponse: + log = getattr(logger, request.level) + log("extension:%s %s", self.extension_id, request.message) + return LogResponse() + + def _raise_unwired_runtime(self, method_name: str) -> None: + raise NotImplementedError( + f"ExtensionAPI.{method_name} must be wired to LNbits services before use." + ) + + +def list_extension_api_methods( + api_cls: type[ExtensionAPI] = ExtensionAPI, +) -> list[ExtensionAPIMethod]: + methods: list[ExtensionAPIMethod] = [] + + for python_name, function in inspect.getmembers(api_cls, inspect.isfunction): + export = getattr(function, _EXTENSION_API_METHOD_ATTR, None) + if not export: + continue + + request_model, response_model = _get_method_models(function) + methods.append( + ExtensionAPIMethod( + method_id=export.method_id, + namespace=export.namespace, + name=export.name, + python_name=python_name, + host_name=export.host_name, + sdk_name=export.sdk_name, + description=export.description, + request_model=request_model, + response_model=response_model, + required_permission=export.required_permission, + public_context=export.public_context, + ) + ) + + return sorted(methods, key=lambda method: method.method_id) + + +def get_extension_api_method( + method_id: str, + api_cls: type[ExtensionAPI] = ExtensionAPI, +) -> ExtensionAPIMethod: + for method in list_extension_api_methods(api_cls): + if method.method_id == method_id: + return method + raise KeyError(f"Unknown extension API method '{method_id}'.") + + +def extension_api_contract( + api_cls: type[ExtensionAPI] = ExtensionAPI, +) -> dict[str, object]: + return { + "version": 1, + "methods": [ + { + "id": method.method_id, + "namespace": method.namespace, + "name": method.name, + "python_name": method.python_name, + "host_name": method.host_name, + "sdk_name": method.sdk_name, + "sdk_qualified_name": method.sdk_qualified_name, + "description": method.description, + "required_permission": method.required_permission, + "public_context": method.public_context, + "request_schema": method.request_model.schema( + ref_template="#/definitions/{model}" + ), + "response_schema": method.response_model.schema( + ref_template="#/definitions/{model}" + ), + } + for method in list_extension_api_methods(api_cls) + ], + } + + +def _get_method_models( + function: Callable[..., object], +) -> tuple[type[BaseModel], type[BaseModel]]: + signature = inspect.signature(function) + request_parameters = [ + parameter + for parameter in signature.parameters.values() + if parameter.name != "self" + ] + if len(request_parameters) != 1: + raise TypeError( + f"Extension API method '{function.__name__}' must accept one request model." + ) + + hints = get_type_hints(function) + request_model = hints.get(request_parameters[0].name) + response_model = hints.get("return") + + if not _is_pydantic_model(request_model): + raise TypeError( + f"Extension API method '{function.__name__}' request must be a BaseModel." + ) + if not _is_pydantic_model(response_model): + raise TypeError( + f"Extension API method '{function.__name__}' response must be a BaseModel." + ) + + return request_model, response_model + + +def _is_pydantic_model(value: object) -> bool: + return isinstance(value, type) and issubclass(value, BaseModel) diff --git a/lnbits/core/extensions/capabilities/__init__.py b/lnbits/core/extensions/capabilities/__init__.py deleted file mode 100644 index 4ade34043..000000000 --- a/lnbits/core/extensions/capabilities/__init__.py +++ /dev/null @@ -1,8 +0,0 @@ -from .catalog import get_extension_capability_registry -from .registry import CapabilityDefinition, CapabilityRegistry - -__all__ = [ - "CapabilityDefinition", - "CapabilityRegistry", - "get_extension_capability_registry", -] diff --git a/lnbits/core/extensions/capabilities/catalog.py b/lnbits/core/extensions/capabilities/catalog.py deleted file mode 100644 index c1413761d..000000000 --- a/lnbits/core/extensions/capabilities/catalog.py +++ /dev/null @@ -1,120 +0,0 @@ -from .models import ( - CreateInvoiceRequest, - CreateInvoiceResponse, - EmptyRequest, - KvGetRequest, - KvGetResponse, - KvListRequest, - KvListResponse, - KvSetRequest, - KvSetResponse, - LogRequest, - LogResponse, - NowResponse, - RandomIdRequest, - RandomIdResponse, - WatchPaymentRequest, - WatchPaymentResponse, -) -from .registry import CapabilityDefinition, CapabilityRegistry - - -def get_extension_capability_registry() -> CapabilityRegistry: - registry = CapabilityRegistry() - - for capability in [ - CapabilityDefinition( - id="storage.get", - namespace="storage", - name="Get storage value", - host_name="kv_get", - sdk_name="get", - description="Read one value from the extension storage namespace.", - request_model=KvGetRequest, - response_model=KvGetResponse, - required_permission="ext.storage.read_write", - public_context=True, - ), - CapabilityDefinition( - id="storage.set", - namespace="storage", - name="Set storage value", - host_name="kv_set", - sdk_name="set", - description="Write one value to the extension storage namespace.", - request_model=KvSetRequest, - response_model=KvSetResponse, - required_permission="ext.storage.read_write", - ), - CapabilityDefinition( - id="storage.list", - namespace="storage", - name="List storage keys", - host_name="kv_list", - sdk_name="list", - description="List keys under a prefix in the extension storage namespace.", - request_model=KvListRequest, - response_model=KvListResponse, - required_permission="ext.storage.read_write", - public_context=True, - ), - CapabilityDefinition( - id="wallet.create_invoice", - namespace="wallet", - name="Create invoice", - host_name="create_invoice", - sdk_name="createInvoice", - description="Create an incoming Lightning invoice for an allowed wallet.", - request_model=CreateInvoiceRequest, - response_model=CreateInvoiceResponse, - required_permission="wallet.create_invoice", - public_context=True, - ), - CapabilityDefinition( - id="payments.watch", - namespace="payments", - name="Watch payment", - host_name="watch_payment", - sdk_name="watch", - description="Subscribe the extension to a payment state callback.", - request_model=WatchPaymentRequest, - response_model=WatchPaymentResponse, - required_permission="payments.watch", - ), - CapabilityDefinition( - id="system.random_id", - namespace="system", - name="Random ID", - host_name="random_id", - sdk_name="id", - description="Create a random extension-local identifier.", - request_model=RandomIdRequest, - response_model=RandomIdResponse, - public_context=True, - ), - CapabilityDefinition( - id="system.now", - namespace="system", - name="Current timestamp", - host_name="now", - sdk_name="now", - description="Return the current Unix timestamp.", - request_model=EmptyRequest, - response_model=NowResponse, - public_context=True, - ), - CapabilityDefinition( - id="system.log", - namespace="system", - name="Log message", - host_name="log", - sdk_name="log", - description="Write a bounded message to the extension log.", - request_model=LogRequest, - response_model=LogResponse, - public_context=True, - ), - ]: - registry.register(capability) - - return registry diff --git a/lnbits/core/extensions/capabilities/models.py b/lnbits/core/extensions/capabilities/models.py deleted file mode 100644 index 94673633f..000000000 --- a/lnbits/core/extensions/capabilities/models.py +++ /dev/null @@ -1,76 +0,0 @@ -from typing import Literal - -from pydantic import BaseModel, Field - - -class EmptyRequest(BaseModel): - pass - - -class KvGetRequest(BaseModel): - key: str = Field(..., min_length=1, max_length=512) - - -class KvGetResponse(BaseModel): - value: str | None = None - - -class KvSetRequest(BaseModel): - key: str = Field(..., min_length=1, max_length=512) - value: str = Field(..., max_length=65536) - - -class KvSetResponse(BaseModel): - ok: bool = True - - -class KvListRequest(BaseModel): - prefix: str = Field(..., min_length=1, max_length=512) - - -class KvListResponse(BaseModel): - keys: list[str] = [] - - -class CreateInvoiceRequest(BaseModel): - wallet_id: str = Field(..., min_length=1, max_length=128) - amount_sat: int = Field(..., gt=0) - memo: str = Field(..., max_length=512) - tag: str = Field(..., min_length=1, max_length=64) - extra: dict[str, str] = Field(default_factory=dict) - - -class CreateInvoiceResponse(BaseModel): - payment_hash: str - payment_request: str - checking_id: str - - -class WatchPaymentRequest(BaseModel): - payment_hash: str = Field(..., min_length=1, max_length=128) - callback_export: str = Field(..., min_length=1, max_length=128) - - -class WatchPaymentResponse(BaseModel): - ok: bool = True - - -class RandomIdRequest(BaseModel): - prefix: str = Field(..., min_length=1, max_length=32) - - -class RandomIdResponse(BaseModel): - id: str - - -class NowResponse(BaseModel): - timestamp: int - - -class LogRequest(BaseModel): - level: Literal["debug", "info", "warning", "error"] = "info" - message: str = Field(..., min_length=1, max_length=2048) - - -class LogResponse(BaseModel): - ok: bool = True diff --git a/lnbits/core/extensions/capabilities/registry.py b/lnbits/core/extensions/capabilities/registry.py deleted file mode 100644 index 93a9c2867..000000000 --- a/lnbits/core/extensions/capabilities/registry.py +++ /dev/null @@ -1,76 +0,0 @@ -from __future__ import annotations - -from dataclasses import dataclass -from typing import Any - -from pydantic import BaseModel - - -@dataclass(frozen=True) -class CapabilityDefinition: - id: str - namespace: str - name: str - host_name: str - sdk_name: str - description: str - request_model: type[BaseModel] - response_model: type[BaseModel] - required_permission: str | None = None - public_context: bool = False - - @property - def sdk_qualified_name(self) -> str: - return f"{self.namespace}.{self.sdk_name}" - - def request_schema(self) -> dict[str, Any]: - return self.request_model.schema(ref_template="#/definitions/{model}") - - def response_schema(self) -> dict[str, Any]: - return self.response_model.schema(ref_template="#/definitions/{model}") - - def to_codegen_contract(self) -> dict[str, Any]: - return { - "id": self.id, - "namespace": self.namespace, - "name": self.name, - "host_name": self.host_name, - "sdk_name": self.sdk_name, - "sdk_qualified_name": self.sdk_qualified_name, - "description": self.description, - "required_permission": self.required_permission, - "public_context": self.public_context, - "request_schema": self.request_schema(), - "response_schema": self.response_schema(), - } - - -class CapabilityRegistry: - def __init__(self) -> None: - self._capabilities: dict[str, CapabilityDefinition] = {} - - def register(self, capability: CapabilityDefinition) -> None: - if capability.id in self._capabilities: - raise ValueError(f"Capability '{capability.id}' already registered.") - self._capabilities[capability.id] = capability - - def get(self, capability_id: str) -> CapabilityDefinition | None: - return self._capabilities.get(capability_id) - - def require(self, capability_id: str) -> CapabilityDefinition: - capability = self.get(capability_id) - if not capability: - raise KeyError(f"Unknown capability '{capability_id}'.") - return capability - - def all(self) -> list[CapabilityDefinition]: - return sorted(self._capabilities.values(), key=lambda c: c.id) - - def by_permission(self, permission_id: str) -> list[CapabilityDefinition]: - return [c for c in self.all() if c.required_permission == permission_id] - - def to_codegen_contract(self) -> dict[str, Any]: - return { - "version": 1, - "capabilities": [c.to_codegen_contract() for c in self.all()], - } diff --git a/lnbits/core/extensions/codegen/__init__.py b/lnbits/core/extensions/codegen/__init__.py new file mode 100644 index 000000000..e1de7faff --- /dev/null +++ b/lnbits/core/extensions/codegen/__init__.py @@ -0,0 +1 @@ +"""SDK generators for extension API contracts.""" diff --git a/lnbits/core/extensions/codegen/typescript.py b/lnbits/core/extensions/codegen/typescript.py new file mode 100644 index 000000000..f7e6b8cc6 --- /dev/null +++ b/lnbits/core/extensions/codegen/typescript.py @@ -0,0 +1,290 @@ +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 | Promise", + "", + ] + + 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]] = {} + for method in methods: + models[method.request_model.__name__] = method.request_model + models[method.response_model.__name__] = method.response_model + return [models[name] for name in sorted(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"] + + 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 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 _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},", + f" publicContext: {_bool(method.public_context)},", + " },", + ] + ) + 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 _bool(value: bool) -> str: + return "true" if value else "false" + + +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())