feat: simpler permissions
This commit is contained in:
Vlad Stan
@@ -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",
|
||||
]
|
||||
|
||||
@@ -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)
|
||||
@@ -1,8 +0,0 @@
|
||||
from .catalog import get_extension_capability_registry
|
||||
from .registry import CapabilityDefinition, CapabilityRegistry
|
||||
|
||||
__all__ = [
|
||||
"CapabilityDefinition",
|
||||
"CapabilityRegistry",
|
||||
"get_extension_capability_registry",
|
||||
]
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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()],
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
"""SDK generators for extension API contracts."""
|
||||
@@ -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> = 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]] = {}
|
||||
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<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 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())
|
||||
Reference in New Issue
Block a user