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.3.tar.gz
(11.9 kB
view hashes)
Built Distribution
Close
Hashes for pybotx-smartapp-rpc-0.3.3.tar.gz
Algorithm | Hash digest | |
---|---|---|
SHA256 | 9941b47e8db58b6454132f97faa1ec8e15a6ed776684b5a5d4934d3515998179 |
|
MD5 | 6a22657ab1065a8a81cfcaebe3db5cf6 |
|
BLAKE2b-256 | cea67b929e4ac2970d0139e55f766bc625d2e410b7e199fdaf1612d18606b7c3 |
Close
Hashes for pybotx_smartapp_rpc-0.3.3-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 4e7af4966618e87b9d7ffaa8e4c616be35eb02b5b6bfdedf5af802b7bcffc398 |
|
MD5 | 74030400bfb967c46f6a56716ddeb691 |
|
BLAKE2b-256 | 7b2e5d1503ebb70b627cb76e9791c597bbfc098c726bcf0e19994f7a7600f97f |