Fork: 0
root / gn-synapse-client-py
Download ZIP 1commit
Transfer to URL with SHA
latest commit 9671ed5b2b
@Eugene Sukhodolskiy Eugene Sukhodolskiy authored 1 day ago
examples/ plain Initial client library skeleton (v0.1.0) 1 day ago
src/ gnexus_synapse Initial client library skeleton (v0.1.0) 1 day ago
tests/ unit Initial client library skeleton (v0.1.0) 1 day ago
.gitignore Initial client library skeleton (v0.1.0) 1 day ago
README.md Initial client library skeleton (v0.1.0) 1 day ago
pyproject.toml Initial client library skeleton (v0.1.0) 1 day ago
README.md

gnexus-synapse — Python-клиент Synapse

Тонкий клиент Ingestion API v1 хаба уведомлений Gnexus Synapse (git+https://git.gnexus.space/git/root/gn-synapse.git): собрать конверт, проверить локально, послать — без ретраев и локальных очередей (очередь у Synapse своя, приём отвечает мгновенно). Контракт — docs/05-ingestion-api.md в репозитории gn-synapse.

Установка

pip install "git+https://git.gnexus.space/git/root/gn-synapse-client-py.git@v0.1.0"

Серверный репозиторий — https://git.gnexus.space/git/root/gn-synapse.git. Зависимость одна: httpx.

Quickstart

from gnexus_synapse import SynapseClient

# аргументы сильнее env; фолбэк — SYNAPSE_URL / SYNAPSE_API_KEY
syn = SynapseClient("http://synapse.gnexus.space:8013", "syn_...")

event = syn.send("bugtrail", "test", "failed",
                 payload={"user_id": "<sub gnexus-auth>", "error": "..."},
                 priority="high", dedup_key="run-2026-10-03")
print(event.id, event.status)  # queued

# пожарная отправка: ошибки Synapse не ломают сервис, пишутся в warning
syn.emit("gntodo", "task", "created", payload={"user_id": uid})

status = syn.status(event.id)          # статусы доставок по каналам
result = syn.send_batch([...])         # batch: result.accepted / result.rejected

env-фолбэк (для сервисов без своей конфигурации):

SYNAPSE_URL=http://synapse:8013
SYNAPSE_API_KEY=syn_...          # выдаёт админ (create-key / MCP key_issue), печатается один раз
SYNAPSE_DEFAULT_SOURCE=bugtrail  # необязательно: не указывать source в каждом вызове
SYNAPSE_TIMEOUT=10               # секунды

default_source — лучшая защита от частой ошибки 403: ключ принадлежит одному источнику, и source в конверте обязан совпадать с ним.

API

Метод Что делает Ошибки
send(source, subject, action, *, priority, payload, dedup_key, ttl_seconds, scheduled_at) POST /api/v1/events → Event(id, status, deduplicated) типизированные, см. ниже
emit(...) тот же, ловит SynapseError → warning в логгер, возвращает `Event \ None` не бросает
send_batch(events) массив конвертов; локально битые не посылаются; rejected — не исключение (частичный успех норма); транспортный сбой/5xx — исключение SynapseError
status(event_id) GET /api/v1/events/{id} → EventStatus (в т.ч. deliveries) SynapseNotFoundError
health() / ready() диагностика без ключа —
close() закрыть свой httpx.Client (инжектный не трогает) —

Исключения (все — наследники SynapseError с полями detail, status_code):

Класс Значение
SynapseConfigError нет url/api_key/source в аргументах и env
SynapseConnectionError Synapse недоступен (DNS/timeout/refused), __cause__ — оригинал
SynapseAuthError 401 — ключ отсутствует/неизвестен/отозван (или источник в архиве)
SynapseForbiddenError 403 — source конверта ≠ источник ключа; задайте default_source
SynapseValidationError 422 сервера или локальная ошибка конверта (status_code is None)
SynapseNotFoundError 404 — событие не у источника ключа
SynapseServerError 5xx — событие не принято, повторите при желании

Локальная валидация зеркалит серверную (^[a-z0-9]([a-z0-9._-]*[a-z0-9])?$, 1..64; priority low/normal/high/critical; ttl 1..604800; dedup_key ≤ 255; payload — JSON-объект), поэтому SynapseValidationError от клиента — почти всегда ошибка кода, ловить её нужно в тестах, а не в рантайме. Сервер игнорирует незарегистрированные пары (subject, action) — их регистрирует админ (add-type / MCP type_register).

Асинхронный близнец для FastAPI-сервисов:

from gnexus_synapse import AsyncSynapseClient
syn = AsyncSynapseClient()          # httpx.AsyncClient, await syn.aclose()
event = await syn.send("gntodo", "task", "created", payload={"user_id": uid})

Конвенции и оговорки

  • Дедуп best-effort (окно 24 ч, сервер не даёт гарантии уникальности): не стройте бизнес-логику на отсутствии дублей.
  • Дедуп вернул 202 с deduplicated=true — это id первого события, не ошибки.
  • payload.user_id — конвенция «о ком событие» (нестроковое значение → адресные доставки skipped): клиент пишет warning, но не блокирует.
  • scheduled_at — резерв контракта v1: передаётся, сервер пока не обрабатывает.
  • Свой httpx.Client создаётся с явным timeout (у httpx дефолт — «бесконечно»). Инжектный клиент (http_client=...) не трогается — таймаут настраивайте сами.
  • user_agent gnexus-synapse-py/<версия> — по нему Synapse видит, кто источался.

Разработка

uv venv -p 3.12 && uv pip install -e . -e '.[dev]'
pytest            # юнит-тесты (MockTransport)
ruff check src tests && mypy src
# интеграционный смок против живого стека:
SYNAPSE_URL=http://localhost:8013 SYNAPSE_API_KEY=syn_... python examples/plain/smoke.py