eXpress SmartApp JSON-RPC library
Project description
BotX-SmartApp-RPC
Библиотека, позволяющая писать смартаппы, используя наш JSONRPC-like протокол
Установка
В pyproject.toml
:
- Переключите
pybotx
на веткуnext
:botx = { git = "https://github.com/ExpressApp/pybotx.git", rev = "next" }
- Добавьте зависимость:
botx-smartapp-rpc = { git = "https://gitlab.ccsteam.ru/rnd/botx-smartapp-rpc", rev = "master" }
Добавление 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.update_app_counter
можно обновлять счетчик уведомлений на иконке смартаппа.
@rpc.method("notify-me")
async def notify_me(
smartapp: SmartApp, rpc_arguments: NotifyMeArgs
) -> RPCResultResponse[None]:
await smartapp.update_app_counter(42)
...
- В мидлварях можно создавать новые объекты в
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.3.6.tar.gz
(11.9 kB
view hashes)
Built Distribution
Close
Hashes for pybotx-smartapp-rpc-0.3.6.tar.gz
Algorithm | Hash digest | |
---|---|---|
SHA256 | b3dc42b6e07f2c274ed10ae397b4adb3a9226be279fabe6ce4c228202fd5755f |
|
MD5 | b1a1c0c0e1407d5087f2b38b9cfd460a |
|
BLAKE2b-256 | 4865c3cf3f80e38d5d7f6d21cd82fd4f0ac321e1008107e70cb581d0a646cc40 |
Close
Hashes for pybotx_smartapp_rpc-0.3.6-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 4f5c9fa2863736af9c01120fbf97999767a7a5a9b70d95aacd44b7739fa8cf28 |
|
MD5 | 7c726a8bfdb8517a4609d5388e24afc5 |
|
BLAKE2b-256 | 262b975731beb12a23d8701ed8292c18c46f3ed2d5d5193bc862e118fe36bba2 |