Megaplan Python SDK

Профессиональная Python-библиотека для работы с API Мегаплана версии 3.

О проекте

Современная библиотека для интеграции с CRM Мегаплан. Она предоставляет удобный и типобезопасный интерфейс для работы с задачами, проектами, сделками и другими сущностями через REST API.

Зачем нужна эта библиотека?

Работа с API Мегаплана напрямую требует знания множества технических деталей: правильной настройки OAuth2-авторизации, обработки токенов, формирования JSON-параметров в query string, обработки ошибок и пагинации. Эта библиотека берет на себя всю рутинную работу, позволяя разработчикам сосредоточиться на бизнес-логике.

Преимущества использования SDK

Вместо прямых HTTP-запросов вы получаете простой и понятный Python-интерфейс. Вместо ручной работы с токенами — автоматическую авторизацию и обновление токенов. Вместо парсинга JSON-ответов — типизированные Pydantic-модели с автодополнением в IDE. Вместо обработки ошибок вручную — понятные исключения с детальной информацией.

Библиотека полностью асинхронная, что позволяет эффективно работать с большими объемами данных и выполнять параллельные запросы. Встроенная логика повторных попыток при временных сбоях сервера делает интеграцию более надежной. Модульная архитектура позволяет легко расширять функциональность и добавлять поддержку новых модулей API.

Содержание

Основы

• Быстрый старт
• Авторизация
• Helper-функции
• Обработка ошибок

Работа с сущностями

• Общие паттерны
• Задачи
• Проекты
• Сделки
• Дела (Todos)
• Дела, привязанные к сущности
• Связи сущностей и их отслеживание
• Уведомления и упоминания
• База знаний

Продвинутые возможности

• Кэширование сущностей
• Глобальные дефолтные лимиты
• Автоматическая подгрузка связанных сущностей
• Работа с фильтрами
• Работа с комментариями
• Настройка HTTP-клиента
• Работа через прокси
• Ручное управление токенами
• Эндпоинты без ресурса: raw() и bulk()

Справочная информация

• Известные ограничения API
• Архитектура
• Требования
• Разработка

Возможности

• Полный CRUD для задач, проектов и сделок
• Метод get_full_details() — получение сущности со всеми связанными данными (комментарии, история, подзадачи и т.д.) за один вызов с параллельной загрузкой
• OAuth2-авторизация с автоматическим обновлением токенов
• Типобезопасность с Pydantic-моделями и полной типизацией
• Асинхронность — поддержка async/await во всех операциях
• Автоматические повторы при ошибках сервера (5xx)
• Кэширование сущностей с LRU и TTL для оптимизации запросов
• FilterBuilder для создания фильтров с fluent API
• Параметр expand для автоматической подгрузки связанных сущностей (подставляются на место ссылок)
• Уведомления с флагом isMention — надёжное отслеживание упоминаний
• Связи сущностей: текущие (linkedDeals/linkedTasks) и события привязки/отвязки из журнала
• Дела (client.todos) — чтение, запись и действия, плюс get_todos() у сделок, задач, проектов, сотрудников и контрагентов
• TodoSync — инкрементальная синхронизация дел там, где у сущности нет timeUpdated и фильтра «изменено после»
• Типизированный журнал (Changeset, BasedOnHistory) у сделок, задач, проектов, контрагентов и дел
• Загрузка вложений client.attachments.upload() и авторизованное скачивание
• Батч-вызовы client.bulk() и escape hatch client.raw() для непокрытых эндпоинтов
• Метод iterate() для автоматической пагинации больших списков
• Helper-функции для создания BaseEntity объектов
• Глобальные дефолтные лимиты для комментариев и истории
• Модульная архитектура для легкого расширения
• Комплексные тесты с покрытием 90%

Установка

pip install megaplan-sdk

Или из исходников:

git clone https://github.com/borzov/megaplan-sdk.git
cd megaplan-sdk
pip install -e .

Быстрый старт

import asyncio
from megaplan_sdk import MegaplanClient

async def main():
    # Создание клиента с учетными данными
    async with MegaplanClient(
        base_url="https://my.megaplan.ru",
        username="user@example.com",
        password="your_password"
    ) as client:

        # Получение списка задач
        tasks = await client.tasks.list(limit=10)
        for task in tasks:
            print(f"Задача: {task.name}")

        # Получение конкретной задачи
        task = await client.tasks.get(task_id=42)
        print(f"Детали задачи: {task.name}, Статус: {task.status}")

        # Упрощенное создание задачи
        new_task = await client.tasks.create_simple(
            "Новая задача",
            employees_resource=client.employees
        )
        print(f"Создана задача: {new_task.name}")

        # Получение задачи со всеми связанными данными за один вызов
        details = await client.tasks.get_full_details(
            task_id=42,
            include_comments=True,
            include_sub_tasks=True,
            include_responsible_details=True
        )
        print(f"Комментариев: {len(details.comments) if details.comments else 0}")
        print(f"Подзадач: {len(details.sub_tasks) if details.sub_tasks else 0}")

if __name__ == "__main__":
    asyncio.run(main())

Авторизация

SDK поддерживает OAuth2-авторизацию. Вы можете передать учетные данные или использовать предварительно полученный токен доступа:

# С логином и паролем (автоматическая авторизация)
client = MegaplanClient(
    base_url="https://my.megaplan.ru",
    username="user@example.com",
    password="password"
)

# С токеном доступа
client = MegaplanClient(
    base_url="https://my.megaplan.ru",
    access_token="your_access_token"
)

Helper-функции

SDK предоставляет удобные функции для создания BaseEntity объектов:

from megaplan_sdk import (
    make_employee_entity,
    make_project_entity,
    make_task_entity,
    make_deal_entity,
    make_contractor_entity,
)

# Вместо ручного создания {"contentType": "Employee", "id": 123}
employee_ref = make_employee_entity(123)
project_ref = make_project_entity(456)
task_ref = make_task_entity(789)
deal_ref = make_deal_entity(101)
contractor_ref = make_contractor_entity(202)

Обработка ошибок

SDK предоставляет специфичные типы исключений для различных сценариев ошибок:

from megaplan_sdk import (
    AuthenticationError,      # 401 - Ошибка аутентификации
    AuthorizationError,        # 403 - Ошибка авторизации (нет прав)
    NotFoundError,            # 404 - Ресурс не найден
    ValidationError,          # 422 - Ошибка валидации запроса
    RateLimitError,           # 429 - Превышен лимит запросов
    ServerError               # 5xx - Ошибка сервера
)

try:
    task = await client.tasks.get(task_id=999)
except NotFoundError:
    print("Задача не найдена")
except AuthenticationError:
    print("Ошибка аутентификации")
except ValidationError as e:
    print(f"Ошибки валидации: {e.errors}")
    # e.errors содержит список ошибок из API

Общие паттерны работы с сущностями

Большинство сущностей (задачи, проекты, сделки) поддерживают одинаковые операции CRUD и паттерны работы. В этом разделе описаны общие методы, которые применяются ко всем типам сущностей.

Базовые операции CRUD

Все ресурсы поддерживают стандартные операции:

Получение списка (list)

# Общий формат для всех ресурсов
entities = await client.{resource}.list(
    limit=None,              # int: Количество элементов на странице
    page_after=None,        # dict: Загрузить страницу, начиная с этой сущности
    page_before=None,       # dict: Загрузить страницу строго до этой сущности
    page_with=None,         # dict: Загрузить страницу с наличием этой сущности
    fields=None,            # any: Набор дополнительных полей
    sort_by=None,           # list[dict]: Массив полей сортировки
    only_requested_fields=None  # bool: Отдавать только перечисленные поля
)

Примеры:

# Получить все задачи
tasks = await client.tasks.list()

# Получить проекты с лимитом
projects = await client.projects.list(limit=50)

# Получить сделки с пагинацией
deals = await client.deals.list(limit=100, page_after={"contentType": "Deal", "id": 100})

Получение по ID (get)

entity = await client.{resource}.get({resource}_id=42)
# Возвращает: объект сущности со всеми полями

Примеры:

task = await client.tasks.get(task_id=42)
project = await client.projects.get(project_id=5)
deal = await client.deals.get(deal_id=200)

Батч-загрузка по ID (get_many)

Метод get_many() загружает несколько сущностей за один запрос и возвращает словарь {id: сущность}:

# Загрузить несколько задач по списку ID
tasks_map = await client.tasks.get_many([101, 102, 103])
# Возвращает: dict[int, Task]
for task_id, task in tasks_map.items():
    print(f"[{task_id}] {task.name}")

# Аналогично для сделок
deals_map = await client.deals.get_many([201, 202])
# Возвращает: dict[int, Deal]

# И для сотрудников (через параллельные одиночные запросы)
employees_map = await client.employees.get_many([301, 302, 303])
# Возвращает: dict[int, Employee]

Примечание: tasks.get_many и deals.get_many используют батч-эндпоинт POST /api/v3/bulk/getEntitiesByLinks. employees.get_many выполняет параллельные одиночные запросы (bulk-эндпоинт для Employee не поддерживается сервером).

Создание (create)

entity = await client.{resource}.create({resource}_data={
    "name": "Название",  # Обязательное поле
    # ... другие поля
})
# Возвращает: созданная сущность

Примеры:

# Простое создание задачи
task = await client.tasks.create({"name": "Новая задача"})

# Создание проекта
project = await client.projects.create({"name": "Новый проект"})

# Создание сделки (требует program)
deal = await client.deals.create({
    "name": "Новая сделка",
    "program": {"contentType": "Program", "id": 10}
})

Обновление (update)

entity = await client.{resource}.update(
    {resource}_id=42,
    {resource}_data={
        "name": "Обновленное название",
        # ... другие поля для обновления
    }
)
# Возвращает: обновленная сущность

Примеры:

task = await client.tasks.update(task_id=42, task_data={"status": "completed"})
project = await client.projects.update(project_id=5, project_data={"name": "Новое название"})
deal = await client.deals.update(deal_id=200, deal_data={"price": {"currency": "RUB", "value": 60000}})

Удаление (delete)

await client.{resource}.delete({resource}_id=42)
# Возвращает: None

Примеры:

await client.tasks.delete(task_id=42)
await client.projects.delete(project_id=5)
await client.deals.delete(deal_id=200)

Пагинация

SDK поддерживает несколько способов работы с большими списками:

Ручная пагинация

# Пагинация "после" определенной сущности
entities = await client.tasks.list(
    limit=50,
    page_after={"contentType": "Task", "id": 100}
)

# Пагинация "до" определенной сущности
entities = await client.tasks.list(
    limit=50,
    page_before={"contentType": "Task", "id": 200}
)

# Пагинация "с" определенной сущностью
entities = await client.tasks.list(
    limit=50,
    page_with={"contentType": "Task", "id": 150}
)

Автоматическая пагинация с iterate()

Метод iterate() автоматически обрабатывает пагинацию и возвращает все элементы:

# Итерация по всем задачам
async for task in client.tasks.iterate(limit=100):
    print(task.name)

# Итерация по всем проектам
async for project in client.projects.iterate(limit=50):
    print(project.name)

# Итерация по всем сделкам
async for deal in client.deals.iterate(limit=200):
    print(deal.name)

Получение полной информации (get_full_details)

Метод get_full_details() позволяет получить сущность со всеми связанными данными за один вызов. Все запросы выполняются параллельно для максимальной производительности.

Общий формат:

details = await client.{resource}.get_full_details(
    {resource}_id=42,
    include_comments=True,          # Загрузить комментарии
    include_history=True,            # Загрузить историю изменений
    comments_limit=50,               # Лимит комментариев (опционально)
    history_limit=100                # Лимит записей истории (опционально)
    # ... другие специфичные параметры для каждого типа
)

Примеры для разных типов:

# Задача со всеми данными
task_details = await client.tasks.get_full_details(
    task_id=42,
    include_comments=True,
    include_sub_tasks=True,
    include_responsible_details=True
)

# Проект со всеми данными
project_details = await client.projects.get_full_details(
    project_id=5,
    include_deals=True,
    include_issues=True,
    include_comments=True
)

# Сделка со всеми данными
deal_details = await client.deals.get_full_details(
    deal_id=200,
    include_comments=True,
    include_status_history=True,
    include_contractor_details=True
)

Доступ к данным:

# Основная сущность
print(details.task.name)      # для задач
print(details.project.name)   # для проектов
print(details.deal.name)      # для сделок

# Связанные данные
if details.comments:
    for comment in details.comments:
        print(comment.content)

if details.history:
    print(f"Записей в истории: {len(details.history)}")

Записи в details.history — типизированные (Changeset, BasedOnHistory; неизвестные серверу типы остаются dict), как и у get_history(). Параметра raw= у get_full_details() нет: если нужен сырой журнал, читайте его отдельным вызовом get_history(entity_id, raw=True).

Подробнее о специфичных параметрах для каждого типа сущностей см. в соответствующих разделах:

• Задачи
• Проекты
• Сделки

Работа с задачами

Примечание: Базовые операции CRUD (list, get, create, update, delete) и пагинация описаны в разделе Общие паттерны работы с сущностями.

Специфичные параметры для задач

Получение списка задач с фильтрацией

Метод list() поддерживает дополнительные параметры для задач:

tasks = await client.tasks.list(
    filter=None,              # TaskFilter: ID фильтра (int/str) или FilterBuilder объект
    statuses=None,            # list[str]: Статусы задач для фильтрации
    q=None,                   # str: Поиск по полю name (с 0.4.0 — реальный серверный фильтр)
    q_in=None,                # list[str]: Поля поиска, по умолчанию ["name"] (например, ["name", "statement"])
    sort_by=None,             # list[dict]: Сортировка; по умолчанию — timeCreated DESC (используйте sort_by=[] для отключения)
    # ... остальные параметры из общих паттернов
)

Поведение с 0.4.0: без sort_by задачи сортируются по timeCreated DESC (как в UI). Чтобы убрать сортировку: sort_by=[]. Константа DEFAULT_SORT_RECENT из megaplan_sdk содержит значение по умолчанию.

Параметр q теперь отправляется серверу как фильтр по полю name. Значение q_in расширяет список полей поиска (например, q_in=["name", "statement"]). Поля description и subject сервером не поддерживаются → NotImplementedError.

Примеры использования фильтров:

# С фильтром по статусам
tasks = await client.tasks.list(
    statuses=["assigned", "in_progress"],
    limit=50
)

# С фильтром по ID (int или str)
tasks = await client.tasks.list(filter=123)
tasks = await client.tasks.list(filter="incoming")

# С FilterBuilder для текстового поиска (рекомендуется)
from megaplan_sdk import TaskFilterBuilder

# Простой поиск по названию
filter_obj = TaskFilterBuilder().field("name").contains("договор").build()
tasks = await client.tasks.list(filter=filter_obj)

# Несколько условий с AND
filter_obj = (
    TaskFilterBuilder()
    .field("name").contains("договор")
    .and_()
    .field("name").starts_with("Важный")
    .build()
)
tasks = await client.tasks.list(filter=filter_obj)

# Условия с OR
filter_obj = (
    TaskFilterBuilder()
    .field("name").contains("договор")
    .or_()
    .field("name").contains("соглашение")
    .build()
)
tasks = await client.tasks.list(filter=filter_obj)

Подробнее о работе с фильтрами см. раздел Работа с фильтрами.

Поля модели Task

• id: int - Идентификатор задачи
• name: str - Название задачи
• description: str - Описание
• status: str - Статус задачи
• responsible: BaseEntity - Ответственный (Employee)
• owner: BaseEntity - Владелец (Employee)
• deadline: str - Срок выполнения
• actual_finish: str - Фактическая дата завершения
• parent: BaseEntity - Родительская задача/проект
• project: BaseEntity - Проект
• priority: str - Приоритет
• tags: list[BaseEntity] - Теги
• attaches: list[BaseEntity] - Вложения (файлы)
• todos: list[BaseEntity] - Подзадачи-чеклисты
• time_created: str - Дата создания (API поле timeCreated)
• time_updated: str - Дата обновления (API поле timeUpdated)
• activity: str | None - Дата последней активности (API поле activity)
• last_comment_time_created: str | None - Время последнего комментария (API поле lastCommentTimeCreated)
• status_change_time: str | None - Время смены статуса (API поле statusChangeTime)
• actual_start: str | None - Фактическое время начала (API поле actualStart)
• last_view: str | None - Время последнего просмотра (API поле lastView)

Фильтрация задач по временным полям

Временны́е поля (activity, lastCommentTimeCreated и др.) не возвращаются в tasks.list() по умолчанию. Используйте константу DEFAULT_TASK_LIST_FIELDS, чтобы запросить их явно:

from megaplan_sdk import DEFAULT_TASK_LIST_FIELDS

# Запросить задачи с временными полями (activity, lastCommentTimeCreated и др.)
tasks = await client.tasks.list(
    limit=50,
    fields=list(DEFAULT_TASK_LIST_FIELDS),
)
for task in tasks:
    print(f"{task.name}: активность {task.activity}")

Примечание: Сортировка по "timeUpdated" не поддерживается API — используйте "activity":

tasks = await client.tasks.list(
    sort_by=[{"fieldName": "activity", "order": "desc"}],
    fields=list(DEFAULT_TASK_LIST_FIELDS),
)

Упрощенные методы создания

Помимо стандартного create(), задачи поддерживают упрощенные методы:

# Создание задачи с текущим пользователем как ответственным
task = await client.tasks.create_simple(
    "Новая задача",
    employees_resource=client.employees  # Автоматически определит текущего пользователя
)

# Создание задачи с указанным ответственным
task = await client.tasks.create_simple(
    "Новая задача",
    responsible_id=123
)

# Создание задачи внутри проекта (автоматически устанавливает связь)
task = await client.tasks.create_in_project(
    "Задача в проекте",
    project_id=456,
    employees_resource=client.employees
)

Примечание: Стандартный метод create() также поддерживается. При создании задачи автоматически устанавливаются isUrgent=False и isTemplate=False, если они не указаны явно.

Получение подзадач

subtasks = await client.tasks.get_sub_tasks(
    task_id=10,                               # int: Идентификатор задачи
    filters=None,                             # list[dict]: Фильтры типов результатов
    limit=None,                               # int: Количество элементов
    page_after=None,                          # dict: Пагинация после
    page_before=None,                         # dict: Пагинация до
    page_with=None,                           # dict: Пагинация с
    fields=None,                              # any: Дополнительные поля
    sort_by=None,                             # list[dict]: Сортировка
    only_requested_fields=None                # bool: Только запрошенные поля
)
# Возвращает: list[Task] - список подзадач

# Получение актуальных подзадач
actual_subtasks = await client.tasks.get_actual_sub_tasks(
    task_id=10,
    # ... те же параметры
)
# Возвращает: list[Task] - список актуальных подзадач

Получение доступных родителей

Методы для получения доступных надзадач и надпроектов (для выбора родителя при создании или перемещении задачи):

# Глобальный поиск доступных родителей для новой задачи
# Возвращает список Task и Project объектов
parents = await client.tasks.get_available_parents(
    is_template=False,                        # bool: Фильтр по шаблонам
    limit=10,                                 # int: Количество элементов
)
for parent in parents:
    print(f"{type(parent).__name__}: {parent.name}")  # "Task: ..." или "Project: ..."

# Доступные родители для существующей задачи
# Исключает саму задачу и её потомков
parents = await client.tasks.get_available_parents_for(
    task_id=123,
    is_template=False,
    limit=10,
)

Примечание: Методы возвращают смешанный список объектов Task и Project, так как задача может быть вложена как в другую задачу, так и в проект.

Получение всех участников задачи

Метод get_all_participants() возвращает полный список участников задачи (ответственный, соисполнители, аудиторы, владелец) в одном запросе:

participants = await client.tasks.get_all_participants(
    task_id=123,
    limit=None,                               # int: Количество элементов
    # ... стандартные параметры пагинации
)
# Возвращает: list[Employee | ContractorHuman | Group]

for participant in participants:
    if hasattr(participant, 'display_name'):
        print(participant.display_name())

Типы участников:

• Employee — сотрудник организации
• ContractorHuman — контрагент-физлицо
• Group — группа участников (например, отдел)

Получение задач на уровне дерева

tasks = await client.tasks.tree_level(
    filter=None,                              # TaskFilter: Фильтр
    limit=None,                               # int: Количество элементов
    page_after=None,                          # dict: Пагинация
    page_before=None,                         # dict: Пагинация
    page_with=None,                           # dict: Пагинация
    fields=None,                              # any: Дополнительные поля
    sort_by=None,                             # list[dict]: Сортировка
    only_requested_fields=None                # bool: Только запрошенные поля
)
# Возвращает: list[Task | Project] - список задач/проектов текущего уровня

Получение полной информации о задаче

Метод get_full_details() для задач поддерживает следующие специфичные параметры:

details = await client.tasks.get_full_details(
    task_id=42,
    include_sub_tasks=True,              # Загрузить подзадачи
    include_actual_sub_tasks=True,       # Загрузить актуальные подзадачи
    include_comments=True,                # Загрузить комментарии
    include_history=True,                 # Загрузить историю изменений
    include_auditors=True,                # Загрузить список аудиторов
    include_executors=True,               # Загрузить соисполнителей
    include_milestones=True,              # Загрузить вехи
    include_responsible_details=True,     # Загрузить полные данные ответственного
    include_owner_details=True,           # Загрузить полные данные постановщика
    comments_limit=50,                    # Лимит комментариев (опционально)
    history_limit=100                    # Лимит записей истории (опционально)
)

Поля объекта TaskFullDetails:

• task: Task - Основная задача
• sub_tasks: list[Task] | None - Подзадачи
• actual_sub_tasks: list[Task] | None - Актуальные подзадачи
• comments: list[Comment] | None - Комментарии
• history: list[dict] | None - История изменений
• auditors: list[dict] | None - Аудиторы
• executors: list[dict] | None - Соисполнители
• milestones: list[Milestone] | None - Вехи
• responsible_details: Employee | None - Полные данные ответственного
• owner_details: Employee | None - Полные данные постановщика

Примечание: Общее описание метода get_full_details() и примеры использования см. в разделе Общие паттерны работы с сущностями.

Работа с вехами (Milestones)

Вехи можно получать и создавать для задач и проектов.

Получение вех

# Получить вехи задачи
milestones = await client.tasks.get_milestones(
    task_id=123,
    limit=50  # Опционально
)

# Получить вехи проекта
milestones = await client.projects.get_milestones(
    project_id=456,
    limit=50  # Опционально
)

# Вехи также доступны через get_full_details()
details = await client.tasks.get_full_details(
    task_id=123,
    include_milestones=True
)
if details.milestones:
    for milestone in details.milestones:
        print(f"{milestone.name}: {milestone.type}")

Создание вехи

from megaplan_sdk.models.milestone import Milestone

# Создать веху для задачи
milestone = await client.tasks.add_milestone(
    task_id=123,
    milestone_data={
        "name": "Release 1.0",
        "description": "Release milestone description",  # Обязательное поле
        "type": "report",  # Обязательное: "report", "reminder", или "note"
        "date": "2026-02-01T10:00:00Z"  # Обязательное: ISO 8601 формат
    }
)

# Или использовать модель Milestone
milestone = await client.tasks.add_milestone(
    task_id=123,
    milestone_data=Milestone(
        name="Release 1.0",
        description="Release milestone description",
        type="report",
        date="2026-02-01T10:00:00Z"
    )
)

# Создать веху для проекта
milestone = await client.projects.add_milestone(
    project_id=456,
    milestone_data={
        "description": "Phase 1 completion",
        "type": "reminder",
        "date": "2026-03-15T14:00:00Z"
    }
)

Обязательные поля при создании вехи:

• description: str - Описание вехи
• type: str - Тип вехи: "report", "reminder", или "note"
• date: str | DateTime | dict - Дата и время вехи (ISO 8601 строка или объект DateTime)

Поля модели Milestone:

• id: int - Идентификатор вехи
• name: str | None - Название вехи
• description: str | None - Описание
• completed: bool | None - Признак завершенности
• type: str | None - Тип вехи
• date: str | DateTime | dict | None - Дата и время
• owner: BaseEntity | None - Создатель (Employee)
• responsible: BaseEntity | None - Ответственный (Employee)
• task: BaseEntity | None - Связанная задача
• project: BaseEntity | None - Связанный проект

Примечание: Метод get_milestones() может вернуть пустой список для некоторых задач/проектов из-за ограничений API (ошибка 500). Это обрабатывается автоматически.

Работа с проектами

Примечание: Базовые операции CRUD (list, get, create, update, delete) описаны в разделе Общие паттерны работы с сущностями.

Важно: Проекты не поддерживают фильтрацию через API (параметр filter недоступен).

Поля модели Project

• id: int - Идентификатор проекта
• name: str - Название проекта
• description: str - Описание
• status: str - Статус проекта
• owner: BaseEntity - Владелец (Employee)
• responsible: BaseEntity - Ответственный (Employee)
• deadline: str - Срок выполнения
• actual_finish: str - Фактическая дата завершения
• parent: BaseEntity - Родительский проект
• priority: str - Приоритет
• tags: list[BaseEntity] - Теги
• attaches: list[BaseEntity] - Вложения
• todos: list[BaseEntity] - Подзадачи-чеклисты
• time_created: str - Дата создания (API поле timeCreated)
• time_updated: str - Дата обновления (API поле timeUpdated)

Упрощенные методы создания

Помимо стандартного create(), проекты поддерживают упрощенный метод:

# Создание проекта с текущим пользователем как владельцем и ответственным
project = await client.projects.create_simple(
    "Новый проект",
    employees_resource=client.employees  # Автоматически определит текущего пользователя
)

# Создание проекта с указанными владельцем и ответственным
project = await client.projects.create_simple(
    "Новый проект",
    owner_id=123,
    responsible_id=123
)

Примечание: При создании проекта автоматически устанавливается isTemplate=False, если не указано явно.

Получение сделок проекта

deals = await client.projects.get_deals(
    project_id=5,                             # int: Идентификатор проекта
    limit=None,                               # int: Количество элементов
    page_after=None,                          # dict: Пагинация после
    page_before=None,                         # dict: Пагинация до
    page_with=None,                           # dict: Пагинация с
    fields=None,                              # any: Дополнительные поля
    sort_by=None,                             # list[dict]: Сортировка
    only_requested_fields=None                # bool: Только запрошенные поля
)
# Возвращает: list[Deal] - список связанных сделок

Получение задач проекта

issues = await client.projects.get_issues(
    project_id=5,                             # int: Идентификатор проекта
    limit=None,                               # int: Количество элементов
    page_after=None,                          # dict: Пагинация
    page_before=None,                         # dict: Пагинация
    page_with=None,                           # dict: Пагинация
    fields=None,                              # any: Дополнительные поля
    sort_by=None,                             # list[dict]: Сортировка
    only_requested_fields=None                # bool: Только запрошенные поля
)
# Возвращает: list[Task] - список задач проекта

# Получение актуальных задач проекта
actual_issues = await client.projects.get_actual_issues(
    project_id=5,
    # ... те же параметры
)
# Возвращает: list[Task] - список актуальных задач проекта

Получение доступных родителей

Методы для получения доступных родительских проектов (для выбора родителя при создании или перемещении проекта):

# Глобальный поиск доступных родительских проектов
parents = await client.projects.get_available_parents(
    is_template=False,                        # bool: Фильтр по шаблонам
    limit=10,                                 # int: Количество элементов
)
for parent in parents:
    print(f"Project: {parent.name}")

# Доступные родители для существующего проекта
# Исключает сам проект и его потомков
parents = await client.projects.get_available_parents_for(
    project_id=456,
    is_template=False,
    limit=10,
)

Примечание: В отличие от задач, проекты могут быть вложены только в другие проекты, поэтому возвращается список объектов Project.

Получение всех участников проекта

Метод get_all_participants() возвращает полный список участников проекта в одном запросе:

participants = await client.projects.get_all_participants(
    project_id=123,
    limit=None,                               # int: Количество элементов
)
# Возвращает: list[Employee | ContractorHuman | Group]

for participant in participants:
    print(f"{type(participant).__name__}: {participant.display_name()}")

Получение полной информации о проекте

Метод get_full_details() для проектов поддерживает следующие специфичные параметры:

details = await client.projects.get_full_details(
    project_id=5,
    include_deals=True,                    # Загрузить связанные сделки
    include_issues=True,                    # Загрузить задачи проекта
    include_actual_issues=True,            # Загрузить актуальные задачи
    include_comments=True,                  # Загрузить комментарии
    include_history=True,                   # Загрузить историю изменений
    include_auditors=True,                  # Загрузить список аудиторов
    include_executors=True,                 # Загрузить соисполнителей
    include_milestones=True,                # Загрузить вехи
    include_responsible_details=True,        # Загрузить полные данные ответственного
    include_owner_details=True,             # Загрузить полные данные владельца
    comments_limit=50,                      # Лимит комментариев (опционально)
    history_limit=100                      # Лимит записей истории (опционально)
)

Поля объекта ProjectFullDetails:

• project: Project - Основной проект
• deals: list[Deal] | None - Связанные сделки
• issues: list[Task] | None - Задачи проекта
• actual_issues: list[Task] | None - Актуальные задачи
• comments: list[Comment] | None - Комментарии
• history: list[dict] | None - История изменений
• auditors: list[dict] | None - Аудиторы
• executors: list[dict] | None - Соисполнители
• milestones: list[Milestone] | None - Вехи
• responsible_details: Employee | None - Полные данные ответственного
• owner_details: Employee | None - Полные данные владельца

Примечание: Общее описание метода get_full_details() и примеры использования см. в разделе Общие паттерны работы с сущностями.

Работа с вехами (Milestones)

Вехи для проектов работают аналогично вехам для задач. См. раздел Работа с вехами в разделе "Работа с задачами" для подробностей.

Работа со сделками

Примечание: Базовые операции CRUD (list, get, create, update, delete) описаны в разделе Общие паттерны работы с сущностями.

Специфичные параметры для сделок

Получение списка сделок с фильтрацией

Метод list() поддерживает дополнительные параметры для сделок:

deals = await client.deals.list(
    filter=None,              # TradeFilter: ID фильтра (int/str) или FilterBuilder объект
    status=None,              # ProgramState: Статус программы для фильтрации
    base_on=None,             # BaseEntity: Базовая сущность для фильтрации
    q=None,                   # str: Поиск по полю name (с 0.4.0 — реальный серверный фильтр)
    q_in=None,                # list[str]: Поля поиска, по умолчанию ["name"]
    sort_by=None,             # list[dict]: Сортировка; по умолчанию — timeCreated DESC (sort_by=[] для отключения)
    # ... остальные параметры из общих паттернов
)

Поведение с 0.4.0: без sort_by сделки сортируются по timeCreated DESC. Используйте sort_by=[] для отключения сортировки по умолчанию.

Параметр q теперь отправляется серверу как фильтр по полю name. Значение q_in расширяет список полей поиска. Поля description и subject сервером не поддерживаются → NotImplementedError.

Примеры использования фильтров:

# С фильтром по ID
deals = await client.deals.list(filter=123)
deals = await client.deals.list(filter="active")

# С FilterBuilder для текстового поиска (рекомендуется)
from megaplan_sdk import TradeFilterBuilder

# Простой поиск по названию
filter_obj = TradeFilterBuilder().field("name").contains("Leader").build()
deals = await client.deals.list(filter=filter_obj)

# Несколько условий
filter_obj = (
    TradeFilterBuilder()
    .field("name").contains("Leader")
    .and_()
    .field("name").starts_with("Важная")
    .build()
)
deals = await client.deals.list(filter=filter_obj)

Подробнее о работе с фильтрами см. раздел Работа с фильтрами.

Поля модели Deal

• id: int - Идентификатор сделки
• name: str - Название сделки
• number: str - Номер сделки
• short_description: str - Краткое описание
• program: BaseEntity - Программа (схема сделки)
• state: ProgramState - Текущий статус в программе
• contractor: BaseEntity - Контрагент (ContractorCompany/ContractorHuman)
• manager: BaseEntity - Ответственный (Employee, API поле manager)
• price: Money - Сумма сделки (объект с полями currency, value)
• cost: Money - Стоимость
• debt: Money - Долг
• result: str - Результат ("positive", "negative", null)
• currency: BaseEntity - Валюта
• deadline: str - Срок
• description: str - Описание (только в deals.get(), не в списке)
• tags: list[BaseEntity] - Теги
• attaches: list[BaseEntity] - Вложения
• time_created: str - Дата создания (API поле timeCreated)
• time_updated: str - Дата обновления (API поле timeUpdated)
• state_time_updated: str - Дата последнего изменения статуса

Примечание: Поля description, deadline и пользовательские поля доступны только при запросе отдельной сделки через deals.get(id), но не в списке.

Важно: При создании сделки обязательно указывать поле program (программа/схема сделки).

Специфичные методы сделок

Применение перехода (изменение статуса)

deal = await client.deals.apply_transition(
    deal_id=200,                              # int: Идентификатор сделки
    transition_id=5                           # int: Идентификатор перехода
)
# Возвращает: Deal - обновленная сделка с новым статусом

Применение триггера

deal = await client.deals.apply_trigger(
    deal_id=200,                              # int: Идентификатор сделки
    trigger_id=3                               # int: Идентификатор триггера
)
# Возвращает: Deal - обновленная сделка

Получение всех участников сделки

Метод get_all_participants() возвращает полный список участников сделки:

participants = await client.deals.get_all_participants(
    deal_id=200,
    limit=None,                               # int: Количество элементов
)
# Возвращает: list[Employee]

for employee in participants:
    print(employee.display_name())

Примечание: В отличие от задач и проектов, сделки возвращают только сотрудников (Employee).

Получение аудиторов сделки

auditors = await client.deals.get_auditors(deal_id=200)
# Параметры:
#   deal_id: int - Идентификатор сделки
# Возвращает: list[dict] - список аудиторов

Получение истории изменения статуса

history = await client.deals.get_status_history(deal_id=200)
# Параметры:
#   deal_id: int - Идентификатор сделки
# Возвращает: list[dict] - список записей истории статусов

Проверка существования сделки

exists = await client.deals.check_exists(deal_params={
    "name": "Название сделки",
    "contractor": {"contentType": "ContractorCompany", "id": 100}
    # ... другие параметры для проверки
})
# Параметры:
#   deal_params: dict - Параметры для проверки
# Возвращает: bool - True если сделка существует, False иначе

Получение полной информации о сделке

Метод get_full_details() для сделок поддерживает следующие специфичные параметры:

details = await client.deals.get_full_details(
    deal_id=200,
    include_comments=True,                # Загрузить комментарии
    include_history=True,                  # Загрузить историю изменений
    include_status_history=True,           # Загрузить историю статусов
    include_auditors=True,                  # Загрузить список аудиторов
    include_manager_details=True,          # Загрузить полные данные ответственного
    include_contractor_details=True,       # Загрузить полные данные контрагента
    include_related_tasks=True,            # Загрузить связанные задачи (/deal/{id}/linkedTasks)
    comments_limit=50,                     # Лимит комментариев (опционально)
    history_limit=100                      # Лимит записей истории (опционально)
)

Поля объекта DealFullDetails:

• deal: Deal - Основная сделка
• comments: list[Comment] | None - Комментарии
• history: list[dict] | None - История изменений
• status_history: list[dict] | None - История статусов
• auditors: list[dict] | None - Аудиторы
• manager_details: Employee | None - Полные данные ответственного
• contractor_details: Contractor | None - Полные данные контрагента
• related_tasks: list[Task] | None - Связанные задачи

Примечание: Общее описание метода get_full_details() и примеры использования см. в разделе Общие паттерны работы с сущностями.

Работа с делами (Todos)

"Дела" (Todo, эндпоинт /api/v3/todo) — самостоятельная сущность, не алиас задач: у неё свой диапазон id, свой набор полей и свои ограничения. Чтение и запись — через client.todos, как и у остальных ресурсов:

todos = await client.todos.list(limit=20)
todo = await client.todos.get(todo_id=88)
found = await client.todos.search(q="созвон")   # list[BaseEntity], не list[Todo]

async for todo in client.todos.iterate(limit=100):
    ...

created = await client.todos.create(name="Позвонить клиенту")
await client.todos.update(todo_id=created.id, todo_data={"name": "Новое название"})
await client.todos.delete(todo_id=created.id)

Действия: finish / renew / take

Отдельных маршрутов /finish, /renew, /take в API нет — они 404. Все три действия идут через один общий POST /todo/{id}/doAction с телом Todo*ActionRequest; SDK прячет это за тремя методами:

finished = await client.todos.finish(todo_id=88, result_text="Готово, клиент подтвердил")
renewed = await client.todos.renew(todo_id=88)      # вернуть завершённое дело обратно
taken = await client.todos.take(todo_id=88)         # назначить себя ответственным

result_text не становится полем дела — сервер публикует его как обычный Comment на деле (client.todos.get_comments(todo_id)), поле Todo.resultText не существует.

Журнал и связи

Как и у сделок/задач/проектов/контрагентов, у дел есть типизированный журнал и события связей:

history = await client.todos.get_history(todo_id=88, limit=50)     # Changeset | BasedOnHistory
events = await client.todos.get_link_events(todo_id=88, since_id=1096)
linked_deals = await client.todos.get_linked_deals(todo_id=88)
linked_tasks = await client.todos.get_linked_tasks(todo_id=88)

Дела, привязанные к сущности

У сделок, задач, проектов, сотрудников и контрагентов есть get_todos() — дела, привязанные к конкретной карточке:

deal_todos = await client.deals.get_todos(deal_id=219)
task_todos = await client.tasks.get_todos(task_id=1006256, limit=50)
project_todos = await client.projects.get_todos(project_id=47)
employee_todos = await client.employees.get_todos(employee_id=1000003)

# У контрагентов — тот же нюанс с подтипом, что и у журнала
contractor_todos = await client.contractors.get_todos(
    contractor_id=1001786,
    content_type="ContractorCompany",
)

Инкрементальная синхронизация

У Todo нет timeUpdated и нет фильтра "изменено после" — узнать через API "что изменилось" напрямую нельзя. Основной канал — вебхуки потока "события" приложения (on_after_create/on_after_update/on_after_drop), а megaplan_sdk.sync.TodoSync — страховка и первичная загрузка поверх них.

from megaplan_sdk.sync import TodoSync, TodoSyncState

sync = TodoSync(client.todos, window_days=30)

# Первый прогон: состояния нет, все дела приедут как created
changes = await sync.poll()
save_state(changes.state.to_dict())          # хранение состояния — на вашей стороне

# Следующий прогон: передаём сохранённое состояние
changes = await sync.poll(TodoSyncState.from_dict(load_state()))

if changes.looks_truncated:
    # ответ сервера подозрительно пуст — ничего не удаляем, ждём следующего опроса
    return

for todo in changes.created + changes.updated:
    upsert(todo)                             # именно upsert, а не insert
for todo_id in changes.deleted:
    drop(todo_id)

Две вещи, о которые проще всего споткнуться: created означает «нет в вашем предыдущем состоянии», а не «создано на сервере» (после потери состояния дело приедет повторно), а deleted — «сервер перестал отдавать этот id», и при looks_truncated=True удалять нельзя вообще. Каждый poll() читает весь список дел аккаунта постранично по 100: window_days ограничивает объём сохраняемого состояния, а не число запросов.

Подробный рецепт с приёмником вебхука, сохранением состояния и разбором семантики deleted/looks_truncated: examples/cookbook/todo-sync.md.

Смежная тема — как узнавать о привязке/отвязке сделок, задач и других сущностей друг к другу через журнал (get_link_events), когда вебхука на это тоже нет: examples/cookbook/link-tracking.md.

Связи сущностей и их отслеживание

Текущие связи

# Связанные сделки и задачи (подресурсы карточки сделки)
linked_deals = await client.deals.get_linked_deals(deal_id=86)
linked_tasks = await client.deals.get_linked_tasks(deal_id=86)
actual_tasks = await client.deals.get_actual_linked_tasks(deal_id=86)
based_on_ids = await client.deals.get_based_on_linked_deals(deal_id=86)  # list[int]

# Для многих сделок сразу — один HTTP-запрос вместо N
links = await client.deals.get_linked_deals_many([86, 219, 220])

Фильтра «задачи по сделке» в API нет (любой формат baseOn игнорируется или отдаёт 422), но подресурс /deal/{id}/linkedTasks работает и отдаёт ровно те задачи, что учтены в tasksCount.

Кто и когда изменил связь

В API нет вебхука на привязку/отвязку: потоки событий приложений (dev.megaplan.ru/apps/events.html) знают только on_after_create, on_after_update, on_after_drop, а карточка сделки не отдаёт список связей — только счётчики. Но журнал сущности пишет каждую привязку и отвязку записью BasedOnHistory, поэтому дельту не нужно вычислять сравнением двух состояний:

events = await client.deals.get_link_events(deal_id=219)

for event in events:
    action = "отвязал" if event.unlink else "привязал"
    print(f"{event.time.value}: {action} {event.other.content_type}#{event.other.id}")

# Инкрементальный опрос: храните наибольший id и передавайте его обратно
last_seen = max(e.id for e in events)
new_events = await client.deals.get_link_events(deal_id=219, since_id=last_seen)

Рабочая схема интеграции: вебхук deals.on_after_update — только триггер «в сделке что-то изменилось», а факт связи берётся из журнала по since_id.

get_link_events() есть не только у сделок — тот же метод есть у задач, проектов, контрагентов и дел. Подробный рецепт с обоснованием, почему API v1 (saveRelation/removeRelation) не подходит, и с примером опроса нескольких типов сущностей: examples/cookbook/link-tracking.md.

Журнал сущности

# Типизированные записи: Changeset (изменения полей), BasedOnHistory (связи);
# неизвестные типы остаются сырыми dict — новый тип на сервере ничего не ломает
history = await client.deals.get_history(deal_id=219, limit=50)

for entry in history:
    if isinstance(entry, Changeset):
        for change in entry.changes or []:
            print(change.field, change.old_value, "→", change.new_value)

# Автопагинация по всему журналу
async for entry in client.deals.iterate_history(deal_id=219, limit=100):
    ...

# Прежнее поведение (list[dict]) — если разбор не нужен
raw = await client.deals.get_history(deal_id=219, raw=True)

get_history(), iterate_history() и get_link_events() доступны у сделок, задач, проектов, контрагентов и дел — сигнатуры одинаковые, меняется только имя параметра с идентификатором.

У контрагентов есть нюанс: сервер не умеет разрешать абстрактный тип Contractor и отвечает 500 на /contractor/{id}/history. SDK ходит по маршруту конкретного подтипа, поэтому методы принимают необязательный content_type — если подтип уже известен (например, из list()), лишнего запроса не будет, иначе SDK сам прочитает карточку:

# Подтип известен — один запрос
history = await client.contractors.get_history(
    contractor_id=1001786,
    content_type="ContractorCompany",   # или "ContractorHuman"
)

# Подтип неизвестен — SDK определит его сам, ценой одного лишнего запроса
events = await client.contractors.get_link_events(contractor_id=1001786)

Для массивов сервер пишет две записи FieldChange — добавленные элементы (old_value=None) и удалённые (new_value=None), — так что изменение уже приходит как дельта.

Ограничения журнала (проверено на боевом аккаунте): серверного фильтра «только связи» нет (filters=["BasedOnHistory"] → 422; работает только Self/Item, причём Self как раз отсекает записи о связях), а history/search их не индексирует — фильтровать нужно на клиенте.

Методы v1 Deal/saveRelation и Deal/removeRelation (и поле RelatedObjects) относятся к API v1 и в SDK не поддерживаются: SDK работает только с v3, где публичного метода создания связи нет — только чтение.

Уведомления и упоминания

/api/v3/notification — единственный надёжный источник факта «пользователя упомянули»: сервер сам ставит isMention, а регэксп по ФИО пропускает упоминания, отрендеренные как HTML.

mentions = await client.notifications.list(limit=60, only_mentions=True)

for note in mentions:
    ref = note.entity_ref          # NotificationEntityRef(entity_type='task', entity_id=..., comment_anchor=...)
    print(note.sender.display_name(), "→", ref.entity_type, ref.entity_id)
    if note.subject_comment:       # subject полиморфен: Comment / Deal / Task / Todo
        print(note.subject_comment.content)

# Все упоминания, а не только внутри одной страницы
async for note in client.notifications.iterate(limit=100, only_mentions=True):
    ...

counter = await client.notifications.counter()      # {attributes: ["mention"], count: N}
types = await client.notifications.activity_types()

Сервер принимает только isActive и пагинацию (остальные поля → 422), поэтому only_mentions фильтрует на клиенте: в list() — внутри полученной страницы, в iterate() — после пагинации, так что короткая страница не обрывает обход.

Эндпоинты без ресурса: client.raw() и client.bulk()

# Любой эндпоинт, для которого в SDK ещё нет ресурса — с рефрешем токена,
# ретраями и разбором meta.errors. Литерал ?{"limit": 60} строится сам.
body = await client.raw("GET", "/api/v3/todo", query={"limit": 20})
todos = body["data"]

# N вызовов одним HTTP-запросом; порядок сохраняется, статус у каждого свой
results = await client.bulk([
    {"method": "GET", "url": f"/api/v3/deal/{deal_id}/linkedDeals"}
    for deal_id in (86, 219, 220)
])
for result in results:
    if result.is_success:
        handle(result.data)
    else:
        log.warning("call failed: %s %s", result.status, result.errors)

Работа с базой знаний

SDK предоставляет доступ к разделам и статьям Базы знаний Мегаплана через два ресурса: client.knowledge_base (разделы) и client.knowledge_article (статьи).

# Список разделов Базы знаний (плоский)
sections = await client.knowledge_base.list()
for s in sections:
    print(s.id, s.name)

# Один раздел с HTML-содержимым
section = await client.knowledge_base.get(11)
print(section.content)

# Итерация по всем разделам с автопагинацией
async for section in client.knowledge_base.iterate():
    print(section.id, section.name)

# Статья по ID (parent всегда None — используйте base)
article = await client.knowledge_article.get(33)
print(article.name, "→ раздел:", article.base.name if article.base else None)

# Экспериментально: раздел вместе со статьями (через парсинг HTML-ссылок)
bundle = await client.knowledge_base.get_with_articles(2)
for a in bundle.articles:
    print(a.id, a.name)

Известные ограничения API (серверная сторона):

• Нет листинга статей: эндпоинт GET /api/v3/knowledgeArticle отсутствует (возвращает 404). Единственный способ обнаружения статей в разделе — метод get_with_articles(), который парсит HTML-ссылки из поля content раздела. Это экспериментальный и хрупкий метод — формат ссылок может измениться на стороне сервера.
• Фильтр parent не работает: knowledge_base.list() всегда возвращает плоский список всех разделов; передача parent в фильтре игнорируется сервером. Иерархии разделов нет.
• Поле parent у статьи всегда null: для определения принадлежности статьи к разделу используйте article.base (не article.parent).

Продвинутые возможности

Кэширование сущностей

SDK автоматически кэширует справочные сущности (сотрудники, контрагенты, отделы) для уменьшения количества API запросов и повышения производительности.

Включение кэша

async with MegaplanClient(
    base_url="https://my.megaplan.ru",
    username="user@example.com",
    password="password",
    enable_cache=True,      # Включить кэш (по умолчанию True)
    cache_ttl=300,          # Время жизни кэша: 5 минут (по умолчанию)
    cache_max_size=1000,    # Макс. размер кэша: 1000 сущностей (по умолчанию)
) as client:
    # Кэш работает автоматически при использовании expand
    tasks_full = await client.tasks.list(limit=10, expand=["responsible", "owner"])

    # Повторная загрузка тех же сотрудников использует кэш
    tasks_full_2 = await client.tasks.list(limit=10, expand=["responsible"])

Управление кэшем

# Очистить весь кэш
client.clear_cache()

# Очистить кэш для конкретного типа сущностей
client.clear_cache_type("Employee")
client.clear_cache_type("Department")
client.clear_cache_type("Contractor")

# Получить статистику кэша
if client._cache:
    stats = client._cache.stats()
    print(f"Кэшировано сущностей: {stats['size']}")
    print(f"Типы: {stats['types']}")  # {"Employee": 15, "Department": 3}

Особенности кэширования

• При достижении cache_max_size удаляются наименее используемые сущности
• Сущности автоматически удаляются из кэша через cache_ttl секунд
• Кэш работает автоматически, не требуя изменений в коде
• При использовании expand уникальные сущности загружаются параллельно

Глобальные дефолтные лимиты

SDK позволяет задать глобальные дефолтные значения для параметров comments_limit и history_limit на уровне клиента. Эти значения будут применяться ко всем вызовам get_full_details() для задач, проектов и сделок, если не переопределены явно.

Установка глобальных дефолтов

async with MegaplanClient(
    base_url="https://my.megaplan.ru",
    username="user@example.com",
    password="password",
    default_comments_limit=50,     # Дефолт для комментариев
    default_history_limit=100,     # Дефолт для истории
) as client:
    # Использует дефолты (50 комментариев, 100 записей истории)
    details = await client.tasks.get_full_details(
        task_id=123,
        include_comments=True,
        include_history=True,
    )

    # Явный параметр переопределяет глобальный дефолт
    details = await client.tasks.get_full_details(
        task_id=456,
        include_comments=True,
        comments_limit=10,  # Используется 10, а не дефолт 50
    )

    # Без указания лимита API использует свой дефолт
    details = await client.projects.get_full_details(
        project_id=5,
        include_comments=True,
        # comments_limit не указан, используется глобальный дефолт 50
    )

Приоритет значений

Система применяет лимиты в следующем порядке (от высшего к низшему приоритету):

1. Явно указанный параметр в методе get_full_details() - всегда имеет наивысший приоритет
2. Глобальный дефолт из MegaplanClient - применяется если параметр не указан явно
3. API default (None) - API использует свои дефолты (обычно без ограничения) если не установлен глобальный дефолт

# Пример приоритетов
client = MegaplanClient(
    base_url="https://my.megaplan.ru",
    access_token="token",
    default_comments_limit=50,  # Глобальный дефолт
)

# Приоритет 1: Явный параметр (загрузит 100)
details = await client.tasks.get_full_details(
    task_id=1,
    include_comments=True,
    comments_limit=100,  # Явно указано
)

# Приоритет 2: Глобальный дефолт (загрузит 50)
details = await client.tasks.get_full_details(
    task_id=2,
    include_comments=True,
    # comments_limit не указан, используется дефолт 50
)

# Приоритет 3: API default без глобального дефолта
client_no_defaults = MegaplanClient(
    base_url="https://my.megaplan.ru",
    access_token="token",
    # default_comments_limit не установлен
)
details = await client_no_defaults.tasks.get_full_details(
    task_id=3,
    include_comments=True,
    # API использует свой дефолт (обычно без ограничения)
)

Когда использовать глобальные дефолты

Глобальные дефолты полезны в следующих случаях:

• Ограничение объема загружаемых данных для ускорения запросов
• Уменьшение размера ответов API для экономии трафика
• Применение одинаковых лимитов ко всем операциям без дублирования кода
• Предотвращение загрузки слишком большого количества комментариев/истории

# Пример для высоконагруженного приложения
client = MegaplanClient(
    base_url="https://my.megaplan.ru",
    access_token="token",
    default_comments_limit=20,   # Ограничение для быстрых ответов
    default_history_limit=50,    # Контроль объема данных
)

# Все вызовы автоматически используют лимиты
tasks_details = await client.tasks.get_full_details(
    task_id=100,
    include_comments=True,
    include_history=True,
)

deals_details = await client.deals.get_full_details(
    deal_id=200,
    include_comments=True,
    include_history=True,
)

projects_details = await client.projects.get_full_details(
    project_id=300,
    include_comments=True,
    include_history=True,
)

Автоматическая подгрузка связанных сущностей

Параметр expand позволяет автоматически подгружать связанные сущности (сотрудников, контрагентов, отделы) вместо получения только ID.

Использование expand в задачах

# Без expand - получаем только базовую информацию
tasks = await client.tasks.list(limit=10)
for task in tasks:
    print(task.responsible)  # BaseEntity(id=123, contentType='Employee')

# С expand - сущности подставляются на место ссылок, тип не меняется
tasks = await client.tasks.list(limit=10, expand=["responsible", "owner"])
for task in tasks:
    if task.responsible:
        # Доступ к полным данным сотрудника
        print(task.responsible.display_name())
        # Вывод: "Максим Борзов (Генеральный директор)"

Изменение в 0.6.0 (ломающее): list()/iterate() с expand= больше не возвращают *FullDetails. Возвращается тот же тип (list[Task], list[Deal], list[Project]), а загруженные сущности подставляются вместо голых ссылок. Полей responsible_details/manager_details в списочном контексте больше нет — они остались только у get_full_details().

Поддерживаемые поля для expand в задачах:

• responsible - ответственный сотрудник
• owner - автор/постановщик задачи

Использование expand в сделках

deals = await client.deals.list(limit=10, expand=["manager", "contractor"])

for deal in deals:
    print(f"Сделка: {deal.name}")

    if deal.manager:
        print(f"Ответственный: {deal.manager.display_name()}")

    if deal_full.contractor_details:
        print(f"Контрагент: {deal_full.contractor_details.display_name()}")

    # Статус сделки с читаемым выводом
    if deal.state:
        print(f"Статус: {deal.state}")  # Использует __str__ из ProgramState

Поддерживаемые поля для expand в сделках:

• manager - ответственный сотрудник
• contractor - контрагент

Использование expand в проектах

projects_full = await client.projects.list(limit=10, expand=["responsible", "owner"])

for project_full in projects_full:
    if project_full.responsible_details:
        print(f"Ответственный: {project_full.responsible_details.display_name()}")

Поддерживаемые поля для expand в проектах:

• responsible - ответственный сотрудник
• owner - владелец проекта

Использование expand в сотрудниках

employees = await client.employees.list(limit=10, expand=["department", "manager"])

for employee in employees:
    # Используем helper метод для форматированного вывода
    print(f"Сотрудник: {employee.display_name()}")

    # Отдел подгружен как полный объект Department
    if employee.department and hasattr(employee.department, 'name'):
        print(f"Отдел: {employee.department.name}")

    # Руководитель подгружен как полный объект Employee
    if employee.manager and hasattr(employee.manager, 'display_name'):
        print(f"Руководитель: {employee.manager.display_name()}")

Поддерживаемые поля для expand в сотрудниках:

• department - отдел сотрудника
• manager - непосредственный руководитель

Использование expand в делах

todos = await client.todos.list(limit=10, expand=["responsible", "user_created"])

for todo in todos:
    if todo.responsible:
        print(f"Ответственный: {todo.responsible.display_name()}")

    if todo.user_created:
        print(f"Постановщик: {todo.user_created.display_name()}")

Поддерживаемые поля для expand в делах:

• responsible - ответственный за дело
• user_created - постановщик дела

Helper методы для читаемого вывода

Модели содержат удобные методы для форматированного вывода:

# Employee
employee.full_name()           # "Максим Борзов"
employee.full_name(include_middle=True)  # "Максим Александрович Борзов"
employee.display_name()        # "Максим Борзов (Генеральный директор)"
str(employee)                  # То же, что display_name()

# Contractor
contractor.display_name()      # "ООО Рога и Копыта" или "Contractor#123"
str(contractor)                # То же, что display_name()

# Department
str(department)                # "IT отдел" или "Department#5"

# ProgramState (статус сделки)
str(deal.state)                # "Переговоры" или "State#10"

Производительность

Использование expand значительно сокращает количество API запросов:

# БЕЗ expand: 1 запрос на список + N запросов на каждого уникального сотрудника
tasks = await client.tasks.list(limit=100)  # 1 запрос
for task in tasks:
    if task.responsible:
        # Нужно загрузить сотрудника отдельно (100+ запросов)
        employee = await client.employees.get(task.responsible.id)

# С expand: 1 запрос на список + 1 батч запросов на уникальных сотрудников
tasks_full = await client.tasks.list(limit=100, expand=["responsible"])
# Всего: 2 запроса (список задач + батч сотрудников)
# Повторные сотрудники берутся из кэша!

Пример:

• 100 задач с 5 уникальными ответственными
• Без expand: 101 запрос (1 список + 100 запросов на сотрудников)
• С expand: 2 запроса (1 список + 1 батч на 5 сотрудников)
• Экономия: 99 запросов (98%)

fields= vs expand= для связанных сущностей

Сервер Мегаплана встраивает повторяющуюся связанную сущность (owner, responsible, manager, contractor) полностью только при первом вхождении в ответе list(); повторы приходят как голая ссылка {contentType, id} без name. Это относится и к fields=[...], и к expand=[...], но ведут себя они по-разному:

# fields=["owner"] — заказывает поле, но не решает дедупликацию:
# только первое вхождение каждого owner будет с именем
tasks = await client.tasks.list(limit=50, fields=["owner"])

# expand=["owner"] — SDK догружает КАЖДОГО owner отдельным батч-запросом
# с кэшированием, поэтому все задачи получают полное имя
tasks_full = await client.tasks.list(limit=50, expand=["owner"])

С 0.6.0 дедупликация чинится автоматически: SDK дозаполняет повторные ссылки из полного вхождения в том же ответе — включая случай, когда полный объект пришёл в другом поле другой записи. Никаких словарей {id: name} на стороне потребителя и никаких предупреждений в логе больше не нужно (#BUG-4). expand= по-прежнему нужен, когда требуется не имя, а вся сущность целиком (должность, отдел, контакты).

Работа с фильтрами

SDK предоставляет удобный FilterBuilder для создания фильтров с использованием fluent API. Фильтры поддерживаются для задач (TaskFilter) и сделок (TradeFilter). Проекты не поддерживают фильтрацию через API.

Базовое использование

from megaplan_sdk import TaskFilterBuilder, TradeFilterBuilder

# Простой текстовый поиск в задачах
filter_obj = TaskFilterBuilder().field("name").contains("договор").build()
tasks = await client.tasks.list(filter=filter_obj)

# Простой текстовый поиск в сделках
filter_obj = TradeFilterBuilder().field("name").contains("Leader").build()
deals = await client.deals.list(filter=filter_obj)

Доступные операции для строковых полей

# Поиск подстроки (рекомендуется для текстового поиска)
filter_obj = TaskFilterBuilder().field("name").contains("договор").build()

# Поиск по началу строки
filter_obj = TaskFilterBuilder().field("name").starts_with("Важный").build()

# Точное совпадение
filter_obj = TaskFilterBuilder().field("status").equals("active").build()

# Исключение подстроки
filter_obj = TaskFilterBuilder().field("name").not_contains("архив").build()

# Не равно
filter_obj = TaskFilterBuilder().field("status").not_equals("completed").build()

Комбинирование условий

# Несколько условий с AND
filter_obj = (
    TaskFilterBuilder()
    .field("name").contains("договор")
    .and_()
    .field("name").starts_with("Важный")
    .build()
)

# Условия с OR
filter_obj = (
    TaskFilterBuilder()
    .field("name").contains("договор")
    .or_()
    .field("name").contains("соглашение")
    .build()
)

Использование фильтров по ID

Помимо FilterBuilder, можно использовать сохраненные фильтры по их ID:

# Фильтр по числовому ID
tasks = await client.tasks.list(filter=123)

# Фильтр по строковому ID
tasks = await client.tasks.list(filter="incoming")
deals = await client.deals.list(filter="active")

Управление фильтрами

SDK предоставляет методы для работы с сохраненными фильтрами:

# Получить список всех фильтров для задач
filters = await client.filters.list("task")

# Получить конкретный фильтр
filter_obj = await client.filters.get("task", filter_id=123)

# Создать новый фильтр
new_filter = await client.filters.create(
    "task",
    filter_id="my_custom_filter",
    filter_config={
        "config": {
            "contentType": "FilterConfig",
            "termGroup": {
                "contentType": "FilterTermGroup",
                "join": "and",
                "terms": [
                    {
                        "contentType": "FilterTermString",
                        "field": "name",
                        "comparison": "contains",
                        "value": "договор"
                    }
                ]
            }
        }
    }
)

# Обновить существующий фильтр
updated = await client.filters.update("task", filter_id=123, filter_config={...})

# Экспортировать фильтр
export_data = await client.filters.export("task", filter_id=123)

Работа с комментариями

Комментарии доступны для задач, проектов и сделок через client.comments.

Получение комментариев

# Комментарии задачи (entity_type по умолчанию "task")
comments = await client.comments.list(entity_id=42)

# Комментарии проекта
comments = await client.comments.list(entity_id=5, entity_type="project")

# Комментарии сделки
comments = await client.comments.list(entity_id=200, entity_type="deal")

# Автоматическая пагинация
async for comment in client.comments.iterate(entity_id=42):
    print(comment.content)

Подгрузка авторов через expand

API Мегаплана не раскрывает поле owner (автор комментария) в списке комментариев. Используйте expand=["owner"], чтобы SDK дозагрузил авторов отдельными запросами (с кэшированием):

# Комментарии задачи с именами авторов
comments = await client.comments.list(entity_id=42, expand=["owner"])
for comment in comments:
    author_name = comment.owner.name if comment.owner else "неизвестен"
    print(f"{author_name}: {comment.content}")

Создание комментария

# Комментарий к задаче
comment = await client.comments.create(entity_id=42, content="Текст комментария")

# Комментарий к проекту
comment = await client.comments.create(entity_id=5, content="Текст", entity_type="project")

# Комментарий к сделке
comment = await client.comments.create(entity_id=200, content="Текст", entity_type="deal")

# С учётом трудозатрат: work — отработанное время в ЧАСАХ (work=2.5 ⇒ 2 ч 30 мин).
# Сериализуется как value = int(work * 3600) секунд; сервер квантует до минут.
comment = await client.comments.create(entity_id=42, content="Сделано", work=2.5)
print(comment.work_time.value)    # 9000 (секунды), плюс .minutes / .hours

Удаление комментариев: на большинстве инсталляций Megaplan запрещено политикой даже автору — comments.delete() ожидаемо вернёт 403 (#19).

Ограничение: Комментарии контрагентов не поддерживаются API (возвращает 500). Используйте комментарии в связанных сделках или задачах.

Вложения

Ссылки на файлы (Comment.attaches, Task.attaches и т.д.) требуют авторизованного скачивания через client.attachments — путь из ссылки относительный и не работает без Bearer-заголовка.

from pathlib import Path

# Скачать вложение целиком в память
data = await client.attachments.download(comment.attaches[0])
Path("report.pdf").write_bytes(data)

# Стриминг для больших файлов
async with client.attachments.stream(attach) as response:
    async for chunk in response.aiter_bytes():
        f.write(chunk)

download()/stream() принимают модель вложения (BaseEntity с path в model_extra), dict с path/url, или сам путь строкой.

Загрузка файла — upload() — отправляет POST /api/file (вне префикса /api/v3) и возвращает ссылку {"contentType": "File", "id": ...}, готовую для передачи в attaches любой сущности (например, комментарию):

# Загрузить файл и прикрепить его к комментарию
ref = await client.attachments.upload("report.pdf")
comment = await client.comments.create(
    entity_id=task.id,
    content="Отчёт во вложении",
    attaches=[ref],
)

Файл читается целиком в память, но в отдельном потоке (asyncio.to_thread), поэтому event loop не блокируется. Стриминга и ограничения на размер нет — очень большой файл целиком окажется в памяти процесса.

Настройка HTTP-клиента

client = MegaplanClient(
    base_url="https://my.megaplan.ru",
    username="user@example.com",
    password="password",
    timeout=60.0,                             # float: Таймаут запросов в секундах (по умолчанию 30.0)
    max_retries=5                              # int: Максимальное количество повторов при 5xx ошибках (по умолчанию 3)
)

Работа через прокси

SDK поддерживает работу через HTTP/HTTPS/SOCKS5 прокси-серверы. Это полезно для корпоративных сетей, где все запросы должны проходить через прокси.

# HTTP прокси с аутентификацией
async with MegaplanClient(
    base_url="https://my.megaplan.ru",
    username="user@example.com",
    password="password",
    proxy="http://login:pass@proxy.corp.local:8080",
) as client:
    tasks = await client.tasks.list()

# HTTP прокси без аутентификации
client = MegaplanClient(
    base_url="https://my.megaplan.ru",
    access_token="token",
    proxy="http://proxy.corp.local:8080",
)

# HTTPS прокси
client = MegaplanClient(
    base_url="https://my.megaplan.ru",
    access_token="token",
    proxy="https://proxy.corp.local:8080",
)

# SOCKS5 прокси (требует httpx[socks])
client = MegaplanClient(
    base_url="https://my.megaplan.ru",
    access_token="token",
    proxy="socks5://user:pass@proxy.corp.local:1080",
)

Поддерживаемые форматы прокси:

• http://proxy:port - HTTP прокси без аутентификации
• http://user:password@proxy:port - HTTP прокси с аутентификацией
• https://proxy:port - HTTPS прокси
• socks5://user:password@proxy:port - SOCKS5 прокси (требует pip install httpx[socks])

Ручное управление токенами

authenticate() и refresh_token() возвращают модель AuthTokenResponse (access_token, refresh_token, expires_in, token_type, scope), а не голую строку — сервер ротирует refresh_token при каждом вызове refresh_token(), и только последний возвращённый токен гарантированно валиден.

# Получить токен доступа
token = await client.auth.authenticate("user@example.com", "password")
# Возвращает: AuthTokenResponse
print(token.access_token, token.expires_in)

# Обновить токен
new_token = await client.auth.refresh_token(refresh_token="refresh_token")
# Параметры:
#   refresh_token: str | None - Токен обновления (опционально, используется сохраненный)
# Возвращает: AuthTokenResponse

# Сохранение ротированных токенов во внешнем хранилище
token = await client.auth.refresh_token(refresh_token=stored_refresh)
save_secrets(
    access_token=token.access_token,
    refresh_token=token.refresh_token,
    expires_in=token.expires_in,
)

# Установить токен вручную
client.set_access_token("your_token")

# Очистить токены
client.auth.clear_tokens()

Автообновление токена

С 0.6.2 SDK обновляет access-токен сам: проверяет срок перед запросом и повторяет запрос один раз после обновления, если сервер ответил 401. Отдельно ничего вызывать не нужно — кроме stream_binary() (скачивание вложений): там обновление работает только проактивно, повтор после 401 на уже открытом стриме не выполняется.

Срок действия токена не сохраняется между запусками: конструктор MegaplanClient принимает access_token/refresh_token, но не их срок действия, поэтому после перезапуска процесса SDK не знает, когда токен истекает, и до первого 401 работает только реактивная половина автообновления — первый запрос платит за это одним лишним обращением к серверу.

Сервер ротирует refresh_token при каждом обновлении — старый после этого отдаёт 400 invalid_grant. Чтобы пережить перезапуск процесса, сохраняйте пару из колбэка:

import json
from pathlib import Path

from megaplan_sdk import MegaplanClient
from megaplan_sdk.models.auth import AuthTokenResponse

TOKENS = Path("tokens.json")


def save_tokens(token: AuthTokenResponse) -> None:
    """Persist the rotated pair; the old refresh token is dead after this."""
    TOKENS.write_text(json.dumps({
        "access_token": token.access_token,
        "refresh_token": token.refresh_token,
    }))


saved = json.loads(TOKENS.read_text()) if TOKENS.exists() else {}

async with MegaplanClient(
    "https://example.megaplan.ru",
    access_token=saved.get("access_token"),
    refresh_token=saved.get("refresh_token"),
    # Первый запуск: tokens.json ещё нет, saved пуст — без пароля клиент
    # останется без единого токена и первый же запрос упадёт с 401.
    username=None if saved else "user@example.com",
    password=None if saved else "password",
    on_token_refresh=save_tokens,
) as client:
    tasks = await client.tasks.list(limit=10)

Колбэк может быть и асинхронным. Исключение внутри него логируется и не прерывает запрос. При ручном вызове authenticate()/refresh_token() это не страшно — токен-пара в любом случае возвращается методом вызывающему коду. Но при автоматическом обновлении внутри _request() (проактивном или по 401) это невосполнимая потеря: ротированная пара приложению никак не передаётся, а прежний refresh_token к этому моменту уже мёртв на сервере — после перезапуска процесса восстановить сессию будет нечем. Поэтому сохранение токенов в колбэке должно быть надёжным и не должно падать молча.

Если refresh_token отклонён сервером, SDK бросает AuthenticationError с указанием переавторизоваться: пароль в памяти не хранится, поэтому сделать это автоматически SDK не может.

Прямые запросы через _http

Для эндпоинтов, не покрытых ресурсами SDK, можно использовать client._http напрямую — но никогда не собирайте query-строку вручную: Мегаплан ожидает параметры как JSON внутри query-строки (?{"limit":5}), а не как обычные ?key=value (#20). Метод get() сам кодирует params в нужном формате:

# Правильно — params кодируется в JSON автоматически
response = await client._http.get("/api/v3/some/endpoint", params={"limit": 5})

# Неправильно — вручную собранная строка `?limit=5` сервер не поймёт
# response = await client._http.get("/api/v3/some/endpoint?limit=5")

Справочная информация

Известные ограничения API

Некоторые эндпоинты Megaplan API имеют ограничения или известные проблемы:

Абстрактный маршрут контрагента

GET /contractor/{id}/history и GET /contractor/{id}/todos отвечают 500 («There is no model class for ...Entity\Contractor»): сервер не создаёт экземпляр полиморфного Contractor. Маршруты конкретных подтипов (/contractorCompany/..., /contractorHuman/...) работают, и SDK ходит именно по ним — см. content_type у get_history()/iterate_history()/ get_link_events()/get_todos()/search_history().

Отвязка сущностей не попадает в журнал

Привязка пишет в журнал запись BasedOnHistory — её видно с обеих сторон связи и это подтверждено на живом аккаунте. А вот отвязка события не порождает: способ, при котором в журнале появляется запись с unlink=true, найти не удалось. Если вам нужен факт отвязки, сравнивайте текущий список связей (get_linked_deals()/get_linked_tasks()) со своим предыдущим снимком.

Дела: непроверенные параметры

Подтверждены живьём: todos.busy_days() без фильтра, finish(result_text=...) и finish(status_id=...). Не проверялись: рабочий формат filter у busy_days() (диапазон дат подобрать не удалось), а также result_attaches и notify_contractors у finish(). Поле when при создании дела сервер отклоняет с 422 в любой известной форме, поэтому create() его не принимает.

Комментарии контрагентов

API возвращает ошибку 500 при попытке получить или создать комментарии для контрагентов. Для отслеживания взаимодействия с контрагентами используйте:

• Журнал действий (action history)
• Комментарии в связанных сделках
• Комментарии в связанных задачах

# Это НЕ работает - вернет 500 ошибку
# comments = await client.contractors.get_comments(contractor_id=123)

# Вместо этого используйте комментарии в сделках контрагента
deals = await client.contractors.get_deals(contractor_id=123)
for deal in deals:
    comments = await client.deals.get_comments(deal.id)

Получение сделок контрагента

SDK предоставляет удобный метод get_deals() для получения сделок контрагента:

# Получить все сделки контрагента
deals = await client.contractors.get_deals(
    contractor_id=123,
    limit=50  # Опционально
)

for deal in deals:
    print(f"[{deal.id}] {deal.name}")
    if deal.state:
        print(f"  Статус: {deal.state}")

Это удобнее, чем использование FilterBuilder:

# Альтернатива через FilterBuilder (более сложный способ)
from megaplan_sdk import TradeFilterBuilder
filter_obj = TradeFilterBuilder().field("contractor").equals(
    {"contentType": "Contractor", "id": 123}
).build()
deals = await client.deals.list(filter=filter_obj)

Поиск сотрудников

Серверная фильтрация на эндпоинте /employee не работает вообще: q / filter молча игнорируются (200 OK), а department_id / status отвергаются (422). Начиная с 0.4.1 эти параметры убраны из сигнатуры, и любая попытка их передать кидает NotImplementedError. Фильтруйте на клиенте:

# Вызовет NotImplementedError
# employees = await client.employees.list(q="Иван Иванов")
# employees = await client.employees.list(department_id=1000004)

# Рекомендуется: загрузить страницу и фильтровать локально
employees = await client.employees.list(limit=500)
working = [e for e in employees if e.is_working]                       # статус-поля #13
by_dept = [e for e in employees if getattr(e.department, "id", None) == 1000004]

# Или get_many по известным ID
employees_map = await client.employees.get_many([123, 456])

Полезные статус-поля Employee (без серверного фильтра): is_working, fire_in_progress, can_login, status.name («В штате»). Поля isDropped у Employee нет (в отличие от Task/Deal/Project).

Текущий пользователь

me = await client.employees.get_current()   # → /api/v3/currentUser
print(me.display_name(), me.email)
# employees.get("me") намеренно кидает ValueError с подсказкой на get_current()

Проверка существования сделки (check_exists)

Метод check_exists() для сделок может возвращать ошибки 500 или 422 из-за ограничений API. SDK автоматически обрабатывает эти ошибки и возвращает False. Для проверки существования сделки рекомендуется использовать альтернативные методы:

# Может вернуть 500/422 ошибку
# exists = await client.deals.check_exists(query="Deal name")

# Альтернатива: используйте поиск через list()
deals = await client.deals.list(q="Deal name", limit=1)
exists = len(deals) > 0

# Или используйте FilterBuilder
from megaplan_sdk import TradeFilterBuilder
filter_obj = TradeFilterBuilder().field("name").equals("Deal name").build()
deals = await client.deals.list(filter=filter_obj, limit=1)
exists = len(deals) > 0

Примечание: SDK автоматически нормализует BaseEntity объекты (конвертирует строковые ID в int), но это не решает проблему багов API.

Параметр statuses для задач

Параметр statuses для фильтрации задач по статусам может возвращать ошибку 422 ValidationError из-за ограничений API. Рекомендуется использовать FilterBuilder для надежной фильтрации:

# Может вернуть 422 ошибку
# tasks = await client.tasks.list(statuses=["assigned", "in_progress"])

# Рекомендуется: используйте FilterBuilder
from megaplan_sdk import TaskFilterBuilder
filter_obj = TaskFilterBuilder().field_enum("status").in_list(["assigned", "in_progress"]).build()
tasks = await client.tasks.list(filter=filter_obj)

Параметр baseOn для сделок

Параметр baseOn для фильтрации сделок по связанной сущности может возвращать ошибку 422 ValidationError из-за ограничений API. SDK автоматически нормализует BaseEntity объекты (конвертирует строковые ID в int), но это не всегда решает проблему:

# Может вернуть 422 ошибку
# deals = await client.deals.list(base_on={"contentType": "Contractor", "id": 123})

# Альтернатива: используйте FilterBuilder
from megaplan_sdk import TradeFilterBuilder
filter_obj = TradeFilterBuilder().field("contractor").equals({"contentType": "Contractor", "id": 123}).build()
deals = await client.deals.list(filter=filter_obj)

Контрагенты и плательщики

Пагинация контрагентов сохраняет конкретный подтип последнего элемента:

async for contractor in client.contractors.iterate(limit=100):
    print(contractor.display_name())

matches = await client.contractors.search("Название", limit=20)
# Старый list(q=...) сохранён как сокращение для поиска.
profile = await client.contractors.get(contractor_id)
payers = await client.contractors.get_payers(contractor_id)

Поиск возвращает компании и физлиц, сокращённые карточки без плательщиков. Поиск по ИНН ненадёжен: для точного сопоставления используйте client.payers.iterate(). Пагинационный курсор требует ContractorCompany или ContractorHuman, а не абстрактный Contractor. Поиск q нельзя сочетать с category_id; обычный список по категории остаётся доступным.

Реквизиты хранятся в типизированных profile.payers; основного плательщика сервер не определяет. Старые поля компании inn/kpp/ogrn/legal_* сохранены для совместимости, но устарели и не вычисляются из первого плательщика. Контакты представлены ContactInfo/Address, страна плательщика имеет строковый ID.

client.payers: list/iterate/get/create/update/delete/extra_fields. client.contractors: search/iterate_search/get_payers/possible_duplicates/ find_duplicates/get_similar; слияние не выполняется.

preview = await client.payers.update(payer_id, {"inn": "1234567890"}, dry_run=True)
for change in preview.changes:
    print(change.field, change.before, change.after)

# После рассмотрения изменений:
updated = await client.payers.apply_update(
    preview, audit_path="/existing/private/directory/profiles.jsonl", actor="operator-id",
)

Предварительный просмотр не пишет на сервер. Apply перечитывает объект, отклоняет конфликт и сохраняет JSONL-намерение до POST, результат после POST. Пара GET/POST не атомарна. При сетевой ошибке исход записи может быть неизвестен: новые POST не повторяются автоматически, кроме авторефреша после отклонённого 401. Поддержку частичного обновления и очистки полей нужно проверить на отдельном тестовом объекте перед пакетной актуализацией. Обычный contractors.update(id, dict) сохраняет историческое поведение; безопасный просмотр/apply доступен также для контрагентов.

Полные примеры, аудит, конфликты и ограничения.

Ограничение архивного списка задач

На проверенном аккаунте список задач без фильтра показывает активную выборку; архивный обход может пропускать карточки, доступные по прямому ID. Параметр statuses и термин status внутри фильтра имеют разное серверное поведение. SDK пока не предоставляет include_archive и не обещает полную выгрузку. При смешанной выдаче идентифицируйте записи по (contentType, id).

Нормализация BaseEntity

SDK автоматически нормализует BaseEntity объекты во всех параметрах:

• Конвертирует строковые ID в int (где возможно)
• Обеспечивает правильный формат contentType и id
• Применяется к параметрам: page_after, page_before, page_with, baseOn, вложенным объектам в deal для check_exists()

Это помогает избежать некоторых ошибок валидации, но не решает все проблемы API.

Архитектура

SDK спроектирован с учетом модульности:

• Resources (TasksResource, ProjectsResource, etc.) - Обработка операций API
• Models (Pydantic) - Типобезопасные структуры данных
• HTTPClient - Низкоуровневые HTTP-операции с retry и авторизацией
• AuthManager - Управление OAuth2-токенами

Расширение SDK

Для добавления нового ресурса:

1. Создайте модель в src/megaplan_sdk/models/
2. Создайте ресурс в src/megaplan_sdk/resources/, наследуя от BaseResource
3. Добавьте ресурс в MegaplanClient:

class MegaplanClient:
    def __init__(self, ...):
        # ... существующий код ...
        self.new_resource = NewResource(self._http)

Требования

• Python 3.11+
• httpx >= 0.25.0
• pydantic >= 2.0.0

Разработка

Установка для разработки

git clone https://github.com/borzov/megaplan-sdk.git
cd megaplan-sdk
pip install -e ".[dev]"

Запуск тестов

pytest

С покрытием:

pytest --cov=megaplan_sdk --cov-report=html

Проверка типов

mypy megaplan_sdk

Линтинг

ruff check megaplan_sdk
ruff format megaplan_sdk

Лицензия

MIT License

Ссылки

• Документация API Мегаплана
• Официальный PHP SDK

Вклад в проект

Вклад приветствуется! Пожалуйста, не стесняйтесь отправлять Pull Request.