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.4.4.tar.gz
(11.8 kB
view hashes)
Built Distribution
Close
Hashes for pybotx-smartapp-rpc-0.4.4.tar.gz
Algorithm | Hash digest | |
---|---|---|
SHA256 | 8b97f59d5206e4247d3e26827f85ff0b5361a4e53618ab5464cfce4e28280364 |
|
MD5 | 86b8828809588ec33314ab22ea0886e7 |
|
BLAKE2b-256 | cc1e6701cafa60ecfe72ad7de00d02490510e23a99d6b3ad7c9275f44dd0dd44 |
Close
Hashes for pybotx_smartapp_rpc-0.4.4-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | fa19ba6ceb3626fadc89c2da471c7a8129ecd6cf03ebfe4634b8f3df7931b748 |
|
MD5 | d2ddb50bc0bc2a37fd68ed6664c20612 |
|
BLAKE2b-256 | f1faadfa54d71bca78e3c016dc57ef1bb0cca80638cd1db099ec4e108f0bc08a |