eXpress SmartApp JSON-RPC library
Project description
BotX-SmartApp-RPC
Библиотека, позволяющая писать смартаппы, используя наш JSONRPC-like протокол
Установка
Используя poetry
:
poetry add pybotx-smartapp-rpc
Добавление RPC методов
- Создайте класс для входящих аргументов:
from pybotx_smartap_rpc import RPCArgsBaseModel
...
class SumArgs(RPCArgsBaseModel):
a: int
b: int
- Создайте RPC метод:
from pybotx_smartapp_rpc import SmartApp, RPCRouter, RPCResultResponse
...
rpc = RPCRouter()
...
@rpc.method("sum")
async def sum(
smartapp: SmartApp, rpc_arguments: SumArgs
) -> RPCResultResponse[int]:
return RPCResultResponse(result=rpc_arguments.a + rpc_arguments.b)
# Так же у метода может не быть аргументов:
@rpc.method("answer")
async def answer(smartapp: SmartApp) -> RPCResultResponse[int]:
return RPCResultResponse(result=42)
- Создайте экземпляр
SmartAppRPC
и подключите роутер из прошлого пункта:
from pybotx_smartapp_rpc import SmartAppRPC
from anywhere import methods
...
smartapp = SmartAppRPC(routers=[methods.rpc])
- Сделайте хендлер для
smartapp_event
и вызывайте в нем хендлер библиотеки
@collector.smartapp_event
async def handle_smartapp_event(event: SmartAppEvent, bot: Bot) -> None:
await smartapp.handle_smartapp_event(event, bot)
Продвинутая работа с библиотекой
- В
RPCResultResponse
можно передаватьbotx.File
файлы.
@rpc.method("get-pdf")
async def get_pdf(
smartapp: SmartApp, rpc_arguments: GetPDFArgs
) -> RPCResultResponse[None]:
...
return RPCResultResponse(result=None, files=[...])
- В
SmartAppRPC
,RPCRouter
иRPCRouter.method
можно передать мидлвари, сначала будут вызваны мидлвари приложения, затем мидлвари роутера и в конце мидлвари метода.
smartapp = SmartAppRPC(..., middlewares=[...])
...
rpc = RPCRouter(middlewares=[...])
...
@rpc.method("sum", middlewares=[...])
RPCArgsBaseModel
это алиас дляpydantic.BaseModel
, вы можете использовать все возможности исходного класса.
from uuid import UUID
...
class DelUserArgs(RPCArgsBaseModel):
# pydantic сериализует входящую строку в UUID
user_huid: UUID
- Через объект
smartapp
, передаваемый в хендлер можно получить доступ кevent
иbot
.
...
@rpc.method("del-user")
async def del_user(
smartapp: SmartApp, rpc_arguments: DelUserArgs
) -> RPCResultResponse[None]:
await smartapp.bot.send_message(
body="Done",
bot_id=smartapp.event.bot.id,
chat_id=smartapp.event.chat.id,
)
...
- Используя метод
smartapp.send_event
можно отправлять RPC ивенты сref: null
.
Это может пригодиться при необходимости отправки уведомления не в ответ на RPC запрос.
@rpc.method("notify-me")
async def notify_me(
smartapp: SmartApp, rpc_arguments: NotifyMeArgs
) -> RPCResultResponse[None]:
...
await smartapp.send_event("notified", files=[notify_file])
...
- Используя метод
smartapp.send_push
можно отправлять пуш уведомлений на клиент. И обновлять счетчик уведомлений на икноке смартапа.
@rpc.method("notify-me")
async def notify_me(
smartapp: SmartApp, rpc_arguments: NotifyMeArgs
) -> RPCResultResponse[None]:
await smartapp.send_push(42, "You have 42 new emails!")
...
- В мидлварях можно создавать новые объекты в
smartapp.state
, чтобы потом использовать их в хендлерах.
async def user_middleware(smartapp: SmartApp, rpc_arguments: RPCArgsBaseModel, call_next: Callable) -> RPCResponse[User]:
smartapp.state.user = await User.get(smartapp.message.user_huid)
return await call_next(smartapp, rpc_arguments)
@rpc.method("get-user-fullname")
async def get_user_fullname(smartapp: SmartApp) -> RPCResultResponse[str]:
return RPCResultResponse(result=smartapp.state.user.fullname)
- Можно выбрасывать пользовательские RPC ошибки, которые будут отправлены как ответ на RPC запрос.
from pybotx_smartapp_rpc import RPCErrorExc, RPCError
...
@rpc.method("return-error")
async def return_error(smartapp: SmartApp, rpc_arguments: RaiseOneErrorArgs) -> None:
# one error
raise RPCErrorExc(
RPCError(
reason="It's error reason",
id="CUSTOM_ERROR",
meta={"args": rpc_arguments.dict()},
)
)
# or list of errors
raise RPCErrorExc(
[
RPCError(
reason="It's error reason",
id="CUSTOM_ERROR",
meta={"args": rpc_arguments.dict()},
),
RPCError(
reason="It's one more error reason",
id="CUSTOM_ERROR_NUMBER_TWO",
meta={"args": rpc_arguments.dict()},
)
]
)
- Можно добавить хендлер на определенный тип исключений. В него будут отправлять исключения того же и дочерних классов.
Хендлер обязан возвращать
RPCErrorResponse
, ошибки из которого будут отправлены источнику запроса.
from pybotx_smartapp_rpc import SmartAppRPC, RPCErrorResponse
...
async def key_error_handler(exc: KeyError, smartapp: SmartApp) -> RPCErrorResponse:
key = exc.args[0]
return RPCErrorResponse(
errors=[
RPCError(
reason=f"Key {key} not found.",
id="KEY_ERROR",
meta={"key": key},
),
]
)
smartapp = SmartAppRPC(..., exception_handlers={KeyError: key_error_handler})
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
pybotx-smartapp-rpc-0.5.1.tar.gz
(11.9 kB
view hashes)
Built Distribution
Close
Hashes for pybotx-smartapp-rpc-0.5.1.tar.gz
Algorithm | Hash digest | |
---|---|---|
SHA256 | 745e8b660b1fdbb4eaf96754553d1a08ea7fc9ee608c9ec2a7c9c126a722b000 |
|
MD5 | bc999205bdb493c8e44a8843017ccb68 |
|
BLAKE2b-256 | cac314561109355c45da71aaaa57939badc679542f918a75eeb40eb1e9739c34 |
Close
Hashes for pybotx_smartapp_rpc-0.5.1-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | dc30ebce4fbd6045fe6c95fbf34227d5a99c4276b1487b4679f8ffe2b1b511fc |
|
MD5 | 93e476456c18c23a00d3a389b4c70d3c |
|
BLAKE2b-256 | 6d0396a575d95ba18f27eceb00b379f314212403d20f2a1f9186f9181dc4f27e |