aiopyrus

Асинхронная Python-библиотека для Pyrus API. Стиль — как у aiogram. Под капотом — HTTPX.

English version

Три режима работы

UserClient — скрипты от своего имени (async)

Автоматизация задач, выгрузки, массовые операции — от имени вашего аккаунта Pyrus. Не нужно регистрировать бота, не нужен публичный сервер.

import asyncio
from aiopyrus import UserClient

async def main():
    async with UserClient(login="user@example.com", security_key="KEY") as client:
        profile = await client.get_profile()
        print(f"Привет, {profile.first_name}!")

        ctx = await client.task_context(12345678)
        print(ctx.get("Статус задачи", "не задан"))

asyncio.run(main())

SyncClient — простые скрипты без async/await

Та же функциональность, что у UserClient, но без async/await. Для скриптов, Jupyter-ноутбуков и простых интеграций.

from aiopyrus import SyncClient

with SyncClient(login="user@example.com", security_key="KEY") as client:
    profile = client.get_profile()
    print(f"Привет, {profile.first_name}!")

    ctx = client.task_context(12345678)
    print(ctx["Статус задачи"])

PyrusBot — бот на вебхуках / polling

Обработка входящих задач, автоматическое согласование, роутинг — aiogram-style.

bot = PyrusBot(login="bot@example", security_key="SECRET")
dp = Dispatcher()

Подробнее о ботах — ниже.

Где взять security_key

1. В Pyrus нажмите Настройки (шестерёнка слева внизу)
2. Перейдите в Авторизация (pyrus.com/t#authorize)
3. Скопируйте Секретный API ключ

Теперь можно запускать скрипты от своего имени:

client = UserClient(login="you@company.com", security_key="<скопированный ключ>")

Главная фишка — TaskContext

Работайте с задачами по именам полей из интерфейса Pyrus — без знания field_id, choice_id, person_id.

ctx = await client.task_context(12345678)

status   = ctx["Статус задачи"]        # multiple_choice → str
executor = ctx["Исполнитель"]           # person → "Имя Фамилия"

ctx.set("Статус задачи", "В работе")    # имя варианта → choice_id автоматически
ctx.set("Исполнитель", "Данил Колбасенко")  # имя → person_id автоматически
await ctx.answer("Принято в работу")

Установка

pip install aiopyrus

Python 3.10+

TaskContext — справочник методов

Таблицы — как DataFrame

Поле типа «таблица» открывается через await ctx.table("Имя") и работает по именам колонок — без знания row_id, col_id и форматов ячеек. Значения резолвятся автоматически (имя человека → person_id, True/False → галочка, строка каталога → item_id). Правки ленивые — уходят одним запросом при ctx.answer(), вместе с обычными fill().

ctx = await client.task_context(12345678)
tbl = await ctx.table("План работ")

# ── Чтение — как pandas / список ──
print(tbl.columns)          # ['Что сделать', 'Ответственный', 'Выполнено']
print(len(tbl))             # 7
print(tbl)                  # красивый ASCII-рендер всей таблицы
for row in tbl:
    print(row["Что сделать"], "→", row["Ответственный"], row["Выполнено"])

# ── Фильтрация — как SQLAlchemy ──
todo = tbl.where(**{"Выполнено": False})          # bool → точное совпадение
mine = tbl.where(**{"Ответственный": "Колбасенко"})  # str → подстрока
row  = tbl.find(**{"Что сделать": "Ревью задачи"})

# ── Правка — как Excel ──
row["Выполнено"] = True                            # отметить галочку
for r in tbl.where(**{"Выполнено": False}):
    r["Ответственный"] = "Данил Колбасенко"        # имя → person_id авто

# ── Добавить / удалить строку ──
tbl.add(**{
    "Что сделать": "03/26 Новая задача",
    "Ответственный": "Данил Колбасенко",
    "Выполнено": False,
})
tbl.find(**{"Что сделать": "устаревшая запись"}).delete()

# ── Отправка — всё одним запросом ──
await ctx.answer("Обновил план")

Имена колонок с пробелами передавай через **{"Имя колонки": значение}. Полный пример — examples/14_table_editing.py.

Бот на вебхуках

import asyncio
from aiopyrus import PyrusBot, Dispatcher, Router, FormFilter, StepFilter
from aiopyrus.utils.context import TaskContext

bot = PyrusBot(login="bot@example", security_key="SECRET")
dp = Dispatcher()
router = Router()

@router.task_received(FormFilter(321), StepFilter(2))
async def on_invoice(ctx: TaskContext):
    amount = float(ctx.get("Сумма", "0"))
    if amount > 100_000:
        await ctx.reject("Сумма превышает лимит.")
    else:
        ctx.set("Статус", "Одобрено")
        await ctx.approve("Одобрено автоматически.")

dp.include_router(router)
asyncio.run(dp.start_webhook(bot, host="0.0.0.0", port=8080, path="/pyrus"))

Лимит размера вебхука

Pyrus изредка присылает вебхуки больше 2 МБ (задачи с длинной историей). Сервер принимает до 16 МиБ по умолчанию; лимит настраивается:

asyncio.run(dp.start_webhook(
    bot, port=8080, path="/pyrus",
    max_request_size=64 * 1024 * 1024,  # 64 МиБ
))

Если перед ботом стоит nginx / Angie — поднимите лимит и там (client_max_body_size 16m;), иначе 413 вернёт прокси, не дойдя до бота.

Бот на polling (без публичного сервера)

asyncio.run(
    dp.start_polling(
        bot,
        form_id=321,
        steps=2,
        interval=30.0,       # секунды между запросами
        skip_old=True,        # не обрабатывать существующие задачи
    )
)

Работает за файрволом, не требует публичный URL.

Polling: защита от повторного срабатывания

Polling отслеживает last_modified_date каждой задачи. Если хендлер изменяет задачу (ctx.set(...), ctx.answer(...)), её last_modified_date обновится — и следующий poll вызовет хендлер снова. Это может привести к дублированию комментариев.

Защита — FieldValueFilter в декораторе:

@dp.task_received(
    FormFilter(321), StepFilter(2),
    FieldValueFilter(field_name="Статус", value="Открыта"),
    FieldValueFilter(field_name="Исполнитель", value=None),
)
async def on_new_task(ctx: TaskContext):
    ctx.set("Статус", "В работе")
    ctx.set("Исполнитель", "Данил Колбасенко")
    await ctx.approve("Принято")
    # После этого статус уже не "Открыта" → фильтр отсечёт при повторном poll

Фильтры

from aiopyrus import FormFilter, StepFilter, FieldValueFilter, EventFilter, F

# Классические — по id
@router.task_received(FormFilter(321), StepFilter(2))

# Или по названию формы (резолвится в id один раз при старте через bot.get_forms())
@router.task_received(FormFilter("Заявки на доступ"), StepFilter(2))
@router.task_received(FormFilter([321, "Согласование договора"]))  # mix

# По значению поля
@router.task_received(FieldValueFilter(field_name="Тип", value="Баг"))

# Magic F
@router.task_received(F.form_id.in_([321, 322]), F.text.contains("срочно"))

# Композиция: &, |, ~
@router.task_received(FormFilter(321) & StepFilter(2) & ~FieldValueFilter(field_name="Статус", value="Закрыт"))

# Временные (для polling)
from aiopyrus.bot.filters import ModifiedAfterFilter, CreatedAfterFilter
@router.task_received(ModifiedAfterFilter())  # только задачи, изменённые после старта бота

# По ожидающему согласованию (approver — person_id ИЛИ role_id)
from aiopyrus import ApprovalPendingFilter
@router.task_received(ApprovalPendingFilter(5555))                    # текущий шаг
@router.task_received(ApprovalPendingFilter([5555, 6666]))            # любой из списка (OR)
@router.task_received(ApprovalPendingFilter([5555, 6666], any_step=True))  # на ЛЮБОМ шаге маршрута

Роли vs люди: если согласование назначено на роль, передавай в ApprovalPendingFilter role_id, а не person_id членов роли (await client.get_roles()).

Поиск задач на согласовании без бота

ApprovalPendingFilter — бот-фильтр (работает с payload.task). Для поиска по реестру от имени UserClient (реестр не отдаёт approvals) есть helper — он делает two-step (get_registers → get_tasks → фильтр):

# Задачи форм 321 и 322, ждущие согласования у роли 5555 или 6666
tasks = await client.find_pending_approvals([5555, 6666], forms=[321, 322])
for t in tasks:
    print(t.id, t.current_step, t.get_approver_names(t.current_step))

Для cross-form поиска без перечисления форм — client.search_tasks_internal(approver_ids=[...]) (внутренний WCF-эндпоинт, experimental).

Middleware

from aiopyrus import BaseMiddleware

class LoggingMiddleware(BaseMiddleware):
    async def __call__(self, handler, payload, bot, data):
        print(f"Task {payload.task_id}")
        return await handler(payload, bot, data)

dp.middleware(LoggingMiddleware())

Inbox vs Register vs get_task — что возвращает API

Разные эндпоинты Pyrus возвращают разный объём данных в задаче:

Что это значит для фильтрации:

• FormFilter и StepFilter не сработают на данных из inbox (всё None).
• start_polling(form_id=...) автоматически подставляет form_id — фильтры работают.
• Для inbox-поллинга нужен enrich=True (дополнительный get_task() на каждую задачу).

Рекомендация: если знаете формы — используйте start_polling(form_id=[id1, id2]). Inbox-поллинг подходит только для сценария «все входящие без фильтрации».

Данные организации

async with UserClient(login=LOGIN, security_key=KEY) as client:
    # Реестр с фильтрами
    tasks = await client.get_register(321, steps=[1, 2], due_filter="overdue")

    # CSV-экспорт реестра
    csv_text = await client.get_register_csv(321, steps=[1, 2])

    # Реестры нескольких форм параллельно
    regs = await client.get_registers([321, 322, 323])
    for form_id, tasks in regs.items():
        print(f"Форма {form_id}: {len(tasks)} задач")

    # Потоковое чтение большого реестра (10 000+ задач, без загрузки в память)
    async for task in client.stream_register(321, steps=[1, 2]):
        print(task.id, task.current_step)

    # Фильтрация при стриминге (клиентская, для условий которые сервер не умеет)
    async for task in client.stream_register(321, predicate=lambda t: t.text):
        print(task.id)

    # Параллельный поиск по нескольким формам
    all_tasks = await client.search_tasks({321: [1, 2], 322: None})

    # Списки задач (проекты / канбан-доски)
    lists = await client.get_lists()
    list_tasks = await client.get_task_list(lists[0].id)

    # Каталоги
    catalogs = await client.get_catalogs()
    cat = await client.get_catalog(999)
    item = cat.find_item("Москва")

    # Участники — умный поиск (имя/email/логин/ID)
    person = await client.find_member("Данил Колбасенко")
    person = await client.find_member("100500")           # числовая строка → ID
    person = await client.find_member(100500)             # int → ID
    members = await client.get_members()

    # Строгий поиск (когда в компании несколько тёзок)
    person = await client.find_member_by_id(100500)        # только по ID
    person = await client.find_member_by_email("kolbasenko@corp.ru")  # только по email
    found = await client.find_members_by_emails(["alice@corp.ru", "bob@corp.ru"])

    # Аватар
    uploaded = await client.upload_file("photo.jpg")
    await client.set_avatar(person.id, uploaded.guid)

    # Роли
    roles = await client.get_roles()

    # Файлы
    uploaded = await client.upload_file("/path/to/file.pdf")
    content = await client.download_file("guid")

    # Прикрепить файл к комментарию
    await client.comment_task(task_id, text="Документ", attachments=[uploaded.guid])

    # Прикрепить файл к полю типа file
    await client.comment_task(task_id, field_updates=[
        {"id": 686, "value": [{"guid": uploaded.guid}]},
    ])

    # Печатные формы (PDF)
    pdf = await client.download_print_form(task_id=12345678, print_form_id=1)

    # Объявления
    announcements = await client.get_announcements()

Батч-операции

Параллельное выполнение через asyncio.gather:

async with UserClient(login=LOGIN, security_key=KEY) as client:
    # Получить несколько задач параллельно (ошибки пропускаются)
    tasks = await client.get_tasks([1001, 1002, 1003])

    # Создать несколько задач (типизированные модели)
    from aiopyrus import NewTask, NewRole, MemberUpdate
    results = await client.create_tasks([
        NewTask(form_id=321, fields=[{"id": 1, "value": "A"}]),
        NewTask(text="Простая задача"),
    ])

    # Прокомментировать несколько задач через TaskContext
    ctxs = await client.task_contexts([1001, 1002])
    ctxs[0].set("Статус", "Выполнена")
    ctxs[1].set("Статус", "Отклонена")
    await asyncio.gather(
        ctxs[0].approve("Одобрено"),
        ctxs[1].reject("Отклонено"),
    )

    # Реестры нескольких форм параллельно
    regs = await client.get_registers([321, 322, 323])

    # Батч-операции с ролями и участниками
    await client.create_roles([NewRole(name="Admins", member_ids=[1, 2]), NewRole(name="Users")])
    await client.update_members([MemberUpdate(member_id=100, position="Lead"), MemberUpdate(member_id=200, position="Dev")])

Утилиты

FieldUpdate — конструктор обновлений полей

from aiopyrus import FieldUpdate

# Ручные фабрики
updates = [
    FieldUpdate.text(field_id=1, value="Москва"),
    FieldUpdate.choice(field_id=2, choice_id=3),
    FieldUpdate.person(field_id=3, person_id=100500),
    FieldUpdate.checkmark(field_id=4, checked=True),
    FieldUpdate.catalog(field_id=5, item_id=42),
]

# Автоопределение формата по типу поля
task = await client.get_task(12345678)
updates = [
    FieldUpdate.from_field(task.get_field("Статус"), 3),            # choice_id
    FieldUpdate.from_field(task.get_field("Исполнитель"), 100500),   # person_id
    FieldUpdate.from_field(task.get_field("Описание"), "Текст"),     # text
]
await client.comment_task(task.id, field_updates=updates)

URL-хелперы

# Ссылка на задачу в браузере
url = client.get_task_url(12345678)
# → "https://pyrus.com/t#id12345678"

# Ссылка на форму
url = client.get_form_url(321)
# → "https://pyrus.com/form/321"

Работают и для on-premise: https://pyrus.mycompany.ru/t#id12345678.

Прочие утилиты

from aiopyrus import get_flat_fields, format_mention, select_fields

# Рекурсивный flatten вложенных полей (title-секции, таблицы)
flat = get_flat_fields(task.fields)

# HTML @упоминание для formatted_text
html = format_mention(100500, header="Данил Колбасенко")
await client.comment_task(task_id, formatted_text=html)

# Выборка полей из списка моделей
tasks = await client.get_register(321)
slim = select_fields(tasks, {"id", "current_step", "fields"})

Тестирование

from aiopyrus import create_mock_client
from aiopyrus.types import Task

# AsyncMock с spec=UserClient
mock = create_mock_client(
    get_task=Task(id=12345678, text="Test"),
    get_members=[],
)

task = await mock.get_task(12345678)
assert task.id == 12345678
mock.get_task.assert_awaited_once_with(12345678)

# Поддержка async context manager
async with mock as client:
    await client.get_inbox()

Управление этапами согласования

# Пересогласование (вернуть шаг в «ожидание»)
await client.comment_task(task_id, approvals_rerequested=[[100500]])

# Добавить согласующего на этап
await client.comment_task(task_id, approvals_added=[[{"id": 100500}]])

# Убрать согласующего с этапа
await client.comment_task(task_id, approvals_removed=[{"id": 100500}])

Боты Pyrus комбинируют approvals_removed + approvals_added для переключения задачи между этапами.

Хелперы согласования на Task

Модель Task предоставляет методы для работы с этапами согласования:

task = await client.get_task(12345678)

# Все согласующие на этапе 1
entries = task.get_approvals(1)

# Только утвердившие
approved = task.get_approvals(1, choice="approved")

# Ожидающие
waiting = task.get_approvals(2, choice=ApprovalChoice.waiting)

# Словарь {номер_этапа: [ApprovalEntry, ...]}
by_step = task.approvals_by_step

# Удобные методы
names  = task.get_approver_names(1)                    # ["Данил Колбасенко", "Иван Иванов"]
emails = task.get_approver_emails(1, choice="approved") # ["kolbasenko@corp.ru"]
ids    = task.get_approver_ids(2)                       # [100500]

Журнал событий (on-premise)

Аудит-эндпоинты доступны только в серверной версии Pyrus. Все возвращают CSV.

# Журнал безопасности (логины, пароли, роли — 113 типов событий)
csv = await client.get_event_history(after=1000, count=500)

# История доступа к файлам
csv = await client.get_file_access_history(count=1000)

# Доступ к задачам / экспорт задач / скачивание реестров
csv = await client.get_task_access_history()
csv = await client.get_task_export_history()
csv = await client.get_registry_download_history()

Подробнее: https://pyrus.com/ru/help/api/event-log

Rate limiting

bot = PyrusBot(
    login="bot@example",
    security_key="SECRET",
    requests_per_minute=30,
    requests_per_10min=4000,
)

Встроенный rate limiter с экспоненциальным backoff. Лимиты Pyrus API: 5000 запросов / 10 мин.

Дебаг — что именно отправляется в Pyrus

Все HTTP-запросы и ответы пишутся в стандартный logging — нужно только поднять уровень:

import logging
logging.basicConfig(
    level=logging.DEBUG,
    format="%(asctime)s %(levelname)-5s %(name)s: %(message)s",
)
# Или точечно — только aiopyrus, не httpx/asyncio:
logging.getLogger("aiopyrus").setLevel(logging.DEBUG)

Будет видно метод, полный URL, какие ключи отправили, статус ответа и тайминг:

DEBUG aiopyrus.session: → POST https://api.pyrus.com/v4/tasks/12345678/comments  body=['text', 'approval_choice']
DEBUG aiopyrus.session:    status=200  keys=['task']  rl_remaining=4997
DEBUG aiopyrus.session: ← POST tasks/12345678/comments  213ms

Loggers по компонентам:

Чтобы заодно увидеть HTTP на уровне httpx (заголовки, низкоуровневые connect/close):

logging.getLogger("httpx").setLevel(logging.DEBUG)

Полный пример с фильтрацией по компонентам — examples/13_debug_logging.py.

On-premise

client = UserClient(
    login="user@corp.ru",
    security_key="KEY",
    base_url="https://pyrus.mycompany.ru",
    ssl_verify=False,  # самоподписанные сертификаты
)

Proxy

client = UserClient(
    login="user@example.com",
    security_key="KEY",
    proxy="http://proxy.corp:8080",
)

Примеры

В папке examples/ — 12 файлов от простого к сложному:

FAQ

Чем aiopyrus отличается от официального pyrus-api?

pyrus-api — синхронная обёртка от Pyrus на requests. aiopyrus — полностью асинхронный фреймворк на httpx с системой роутеров, фильтров и middleware как в aiogram. Работа с полями задач идёт по именам из интерфейса, а не по field_id.

Нужен ли публичный сервер для бота?

Нет. Есть polling-режим (dp.start_polling(...)) — бот сам опрашивает Pyrus по таймеру. Работает за файрволом, NAT, VPN.

Поддерживаются ли on-premise инсталляции Pyrus?

Да. Передайте base_url при создании клиента:

client = UserClient(
    login="user@corp.ru",
    security_key="KEY",
    base_url="https://pyrus.mycompany.ru",
    ssl_verify=False,  # для самоподписанных сертификатов
)

Можно ли использовать без бота, просто как API-клиент?

Да, именно для этого есть UserClient — скрипты от имени вашего аккаунта без регистрации бота.

Лицензия

MIT