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