Webhook Handler API¶

API для обработки входящих webhook-уведомлений от YooKassa.

Библиотека предоставляет два подхода к обработке webhook-уведомлений:

WebhookHandler — базовый обработчик для интеграции с любым веб-фреймворком
WebhookServer — готовый сервер на aiohttp для быстрого старта

WebhookHandler¶

Базовый класс для обработки webhook-уведомлений. Работает с любым веб-фреймворком.

class aioyookassa.core.webhook_handler.WebhookHandler(validator: WebhookIPValidator | None = None, logger: Logger | None = None)[source]¶

Bases: object

Handler for processing YooKassa webhook notifications.

Provides functionality to: - Parse webhook notifications - Validate IP addresses (optional) - Register callbacks for specific events - Automatically parse event objects into typed Pydantic models

Initialize webhook handler.

Parameters:

validator – IP validator instance. If None, creates default validator with YooKassa official IP ranges.
logger – Logger instance. If None, uses default logger.

register_callback(events: WebhookEvent | str | List[WebhookEvent | str], callback: Callable | None = None) → Callable[source]¶

Decorator for registering event callbacks.

Supports: - Single event: @handler.register_callback(WebhookEvent.PAYMENT_SUCCEEDED) - Multiple events: @handler.register_callback([event1, event2]) - Pattern matching: @handler.register_callback(“payment.*”)

Parameters:

events – Event(s) to register callback for.
callback – Callback function (used when called as decorator).

Returns:

Decorator function or callback.

add_callback(events: WebhookEvent | str | List[WebhookEvent | str], callback: Callable) → None[source]¶

Parameters:

events – Event(s) to register callback for.
callback – Callback function to register.

parse_notification(data: dict) → WebhookNotification[source]¶

Parse raw webhook data into WebhookNotification model.

Parameters:: data – Raw JSON data from webhook request.
Returns:: Parsed WebhookNotification instance.
Raises:: InvalidWebhookDataError – If data is invalid.

Process webhook notification and return typed event object.

Automatically parses the notification object into the appropriate Pydantic model based on event type and calls registered callbacks.

Parameters:: notification – Parsed webhook notification.
Returns:: Typed event object (Payment, Refund, Payout, Deal, PaymentMethod).

Методы¶

parse_notification¶

Парсит сырые данные webhook-уведомления в типизированную модель.

from aioyookassa.core.webhook_handler import WebhookHandler

handler = WebhookHandler()
data = await request.json()
notification = handler.parse_notification(data)

print(f"Event: {notification.event}")
print(f"Type: {notification.type}")

handle_notification¶

Обрабатывает уведомление и возвращает типизированный объект события.

notification = handler.parse_notification(data)
event_object = await handler.handle_notification(notification)

# event_object будет типизированным объектом:
# - Payment для payment.* событий
# - Refund для refund.* событий
# - Payout для payout.* событий
# - Deal для deal.closed
# - PaymentMethod для payment_method.active

Регистрация callbacks¶

Декоратор для регистрации обработчиков событий.

from aioyookassa.types.enum import WebhookEvent
from aioyookassa.types.payment import Payment

@handler.register_callback(WebhookEvent.PAYMENT_SUCCEEDED)
async def on_payment_succeeded(payment: Payment):
    print(f"Payment {payment.id} succeeded")
    # Ваша бизнес-логика

Регистрация для нескольких событий:

@handler.register_callback([
    WebhookEvent.PAYMENT_SUCCEEDED,
    WebhookEvent.PAYMENT_CANCELED
])
async def on_payment_status_change(payment: Payment):
    print(f"Payment {payment.id} status changed")

Регистрация с паттерном (wildcard):

@handler.register_callback("payment.*")
async def handle_all_payments(payment: Payment):
    print(f"Payment event: {payment.id}")

add_callback¶

Обычный метод для регистрации callbacks без декоратора.

async def my_handler(payment: Payment):
    print(f"Payment {payment.id} processed")

handler.add_callback(WebhookEvent.PAYMENT_SUCCEEDED, my_handler)

WebhookIPValidator¶

Валидатор IP-адресов для проверки, что запросы приходят от YooKassa.

class aioyookassa.core.webhook_validator.WebhookIPValidator(allowed_ips: List[str] | None = None, logger: Logger | None = None)[source]¶

Bases: object

Validator for YooKassa webhook IP addresses.

Validates that incoming webhook requests come from authorized YooKassa IP ranges. Supports both IPv4 and IPv6 addresses, CIDR ranges, and individual IPs.

Initialize IP validator.

Parameters:

allowed_ips – List of allowed IP addresses or CIDR ranges. If None, uses default YooKassa IP ranges.
logger – Logger instance. If None, uses default logger.

is_allowed(ip: str) → bool[source]¶

Check if IP address is allowed.

Parameters:: ip – IP address to check (IPv4 or IPv6).
Returns:: True if IP is allowed, False otherwise.

Использование¶

from aioyookassa.core.webhook_validator import WebhookIPValidator

validator = WebhookIPValidator()

# Проверка IP
if validator.is_allowed("185.71.76.1"):
    print("IP is allowed")

# Кастомный список IP (для тестирования)
custom_validator = WebhookIPValidator(
    allowed_ips=["127.0.0.1", "192.168.1.0/24"]
)

WebhookServer¶

Готовый сервер на aiohttp для обработки webhook-уведомлений.

class aioyookassa.contrib.webhook_server.WebhookServer(handler: WebhookHandler | None = None, validator: WebhookIPValidator | None = None, validate_ip: bool = True, logger: Logger | None = None)[source]¶

Bases: object

Ready-to-use aiohttp server for handling YooKassa webhook notifications.

Provides a complete HTTP server setup with IP validation and event handling.

Initialize webhook server.

Parameters:

handler – WebhookHandler instance. If None, creates new one.
validator – IP validator instance. If None, uses default.
validate_ip – Whether to validate IP addresses. Default: True.
logger – Logger instance. If None, uses default logger.

create_app() → Application[source]¶

Create aiohttp application with webhook endpoint.

Returns:: Configured aiohttp Application instance.

run(host: str = '0.0.0.0', port: int = 8080, path: str = '/webhook') → None[source]¶

Run webhook server.

Parameters:

host – Host to bind to. Default: 0.0.0.0.
port – Port to bind to. Default: 8080.
path – Webhook endpoint path. Default: /webhook.

Использование¶

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

from aioyookassa.contrib.webhook_server import WebhookServer
from aioyookassa.core.webhook_handler import WebhookHandler
from aioyookassa.types.enum import WebhookEvent

handler = WebhookHandler()

@handler.register_callback(WebhookEvent.PAYMENT_SUCCEEDED)
async def on_payment(payment):
    print(f"Payment {payment.id} succeeded")

server = WebhookServer(handler=handler)
server.run(host="0.0.0.0", port=8080)

Создание приложения для интеграции:

app = server.create_app()
# Можно добавить дополнительные роуты, middleware и т.д.

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

Интеграция с aiohttp¶

from aiohttp import web
from aioyookassa.core.webhook_handler import WebhookHandler
from aioyookassa.types.enum import WebhookEvent

handler = WebhookHandler()

@handler.register_callback(WebhookEvent.PAYMENT_SUCCEEDED)
async def on_payment_succeeded(payment):
    await process_payment(payment)

async def webhook_endpoint(request):
    # Валидация IP
    if not handler.validator.is_allowed(request.remote):
        raise web.HTTPForbidden()

    # Обработка
    data = await request.json()
    notification = handler.parse_notification(data)
    await handler.handle_notification(notification)

    return web.Response(status=200)

app = web.Application()
app.router.add_post("/webhook", webhook_endpoint)
web.run_app(app)

Интеграция с FastAPI¶

from fastapi import FastAPI, Request, HTTPException
from aioyookassa.core.webhook_handler import WebhookHandler
from aioyookassa.types.enum import WebhookEvent

handler = WebhookHandler()
app = FastAPI()

@handler.register_callback(WebhookEvent.PAYMENT_SUCCEEDED)
async def on_payment_succeeded(payment):
    await process_payment(payment)

@app.post("/webhook")
async def webhook_endpoint(request: Request):
    # Валидация IP
    if not handler.validator.is_allowed(request.client.host):
        raise HTTPException(status_code=403)

    # Обработка
    data = await request.json()
    notification = handler.parse_notification(data)
    await handler.handle_notification(notification)

    return {"status": "ok"}

Валидация IP¶

По умолчанию валидация IP включена и использует официальные IP-адреса YooKassa:

185.71.76.0/27
185.71.77.0/27
77.75.153.0/25
77.75.156.11
77.75.156.35
77.75.154.128/25
2a02:5180::/32

Для отключения валидации (только для разработки!):

from aioyookassa.core.webhook_validator import WebhookIPValidator

# Разрешить все IP (не рекомендуется для production)
validator = WebhookIPValidator(allowed_ips=["0.0.0.0/0"])
handler = WebhookHandler(validator=validator)

Поддерживаемые события¶

payment.waiting_for_capture — платеж ожидает подтверждения
payment.succeeded — платеж успешно выполнен
payment.canceled — платеж отменен
payment_method.active — способ оплаты активирован
refund.succeeded — возврат успешно выполнен
payout.succeeded — выплата успешно выполнена
payout.canceled — выплата отменена
deal.closed — безопасная сделка закрыта

Исключения¶

Webhook-related exceptions.

exception aioyookassa.exceptions.webhooks.InvalidWebhookIPError(message: str = 'IP address not in whitelist')[source]¶: Raised when webhook request comes from an IP address that is not in the allowed whitelist.

exception aioyookassa.exceptions.webhooks.InvalidWebhookDataError(message: str = 'Invalid webhook notification data')[source]¶: Raised when webhook notification data is invalid or cannot be parsed.