Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 12 additions & 1 deletion src/shade/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,16 @@
ShadeError,
SignatureVerificationError,
)
from .models import AssetBalance, Balance, Merchant, ShadeObject, Transfer, TransferStatus
from .models import (
AssetBalance,
Balance,
Merchant,
ShadeObject,
Transfer,
TransferStatus,
WebhookEvent,
WebhookEventType,
)

__version__ = "0.1.0"

Expand All @@ -43,6 +52,8 @@
"SyncHTTPClient",
"Transfer",
"TransferStatus",
"WebhookEvent",
"WebhookEventType",
"config",
"api_base",
"environment",
Expand Down
3 changes: 3 additions & 0 deletions src/shade/models/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
from .base import ShadeObject
from .merchant import Merchant
from .transfer import Transfer, TransferStatus
from .webhook import WebhookEvent, WebhookEventType

__all__ = [
"AssetBalance",
Expand All @@ -13,4 +14,6 @@
"ShadeObject",
"Transfer",
"TransferStatus",
"WebhookEvent",
"WebhookEventType",
]
88 changes: 88 additions & 0 deletions src/shade/models/webhook.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,88 @@
"""
Webhook event model.

Represents a parsed, verified webhook event delivered by the Shade platform.
Field names are converted from ``camelCase`` (backend/JSON) to ``snake_case``
(Python) via pydantic field aliases, matching the convention established by
:class:`~shade.models.transfer.Transfer`.

At this layer ``data`` is deliberately left as the raw decoded JSON object. The
resource layer (``Webhook.construct_event()``) is responsible for coercing it
into the corresponding typed model based on ``type``.
"""
from __future__ import annotations

from datetime import datetime
from enum import Enum
from typing import Any

from pydantic import Field, StrictBool

from .base import ShadeObject


class WebhookEventType(str, Enum):
"""Event types delivered by the Shade platform.

Members subclass :class:`str`, so they compare equal to the wire value::

if event.type == WebhookEventType.PAYMENT_COMPLETED:
...

The list is not exhaustive by design: :class:`WebhookEvent` stores ``type``
as a plain ``str``, so an event type added server-side still parses and can
be compared against a literal until a constant is added here.
"""

PAYMENT_COMPLETED = "payment.completed"
PAYMENT_CANCELLED = "payment.cancelled"
PAYMENT_EXPIRED = "payment.expired"
PAYMENT_PARTIALLY_PAID = "payment.partially_paid"
INVOICE_PAID = "invoice.paid"
INVOICE_SENT = "invoice.sent"
INVOICE_CANCELLED = "invoice.cancelled"
TRANSFER_COMPLETED = "transfer.completed"
TRANSFER_FAILED = "transfer.failed"
SWAP_COMPLETED = "swap.completed"
SWAP_SLIPPAGE_EXCEEDED = "swap.slippage_exceeded"


class WebhookEvent(ShadeObject):
"""A parsed, verified webhook event.

The expected payload is a JSON object of the shape::

{
"id": "evt_123",
"type": "payment.completed",
"data": {"id": "pay_123", ...},
"createdAt": "2026-07-20T12:00:00Z",
"livemode": false
}

Build one with :meth:`ShadeObject.from_dict`, which maps ``createdAt`` to
:attr:`created_at` and parses it into a :class:`~datetime.datetime`.
``livemode`` distinguishes a production event from a test-mode one.

``id``, ``type``, ``data``, ``created_at`` and ``livemode`` are all
required; a payload missing any of them — or carrying an unparseable
timestamp — raises
:class:`~shade.errors.InvalidRequestError` rather than producing a
half-populated event. Unknown extra keys are preserved, per
:class:`~shade.models.base.ShadeObject`.

Attributes:
id: Unique identifier of the event.
type: Event type string, e.g. ``"payment.completed"``. Compare against
:class:`WebhookEventType` members.
data: The event payload, left as the raw decoded JSON object. Typed
model coercion happens in the resource layer, not here.
created_at: When the platform emitted the event.
livemode: ``True`` for a live event, ``False`` for a test-mode one.
"""

id: str
type: str
data: dict[str, Any]
created_at: datetime = Field(alias="createdAt")
livemode: StrictBool
149 changes: 149 additions & 0 deletions tests/test_webhook_event.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,149 @@
from datetime import datetime, timezone

import pytest

import shade
from shade import InvalidRequestError, ShadeObject, WebhookEvent, WebhookEventType

PAYMENT_DATA = {
"id": "pay_123",
"amount": "150.25",
"asset": "USDC",
"status": "completed",
}


def _api_response(**overrides):
"""A representative camelCase backend payload."""
data = {
"id": "evt_123",
"type": "payment.completed",
"data": dict(PAYMENT_DATA),
"createdAt": "2026-07-20T12:00:00Z",
"livemode": True,
}
data.update(overrides)
return data


def test_from_dict_populates_all_fields():
event = WebhookEvent.from_dict(_api_response())

assert event.id == "evt_123"
assert event.type == "payment.completed"
assert event.data == PAYMENT_DATA
assert event.created_at == datetime(2026, 7, 20, 12, 0, tzinfo=timezone.utc)
assert event.livemode is True


def test_data_stays_a_raw_dict():
event = WebhookEvent.from_dict(_api_response())

assert isinstance(event.data, dict)
assert not isinstance(event.data, ShadeObject)
assert event.data["status"] == "completed"


def test_non_dict_data_raises():
with pytest.raises(InvalidRequestError):
WebhookEvent.from_dict(_api_response(data=["a", "b"]))


def test_payload_is_not_mutated():
payload = _api_response()
snapshot = {
**payload,
"data": dict(payload["data"]),
}

WebhookEvent.from_dict(payload)

assert payload == snapshot


def test_created_at_parsed_from_iso_string():
event = WebhookEvent.from_dict(_api_response(createdAt="2026-01-02T03:04:05Z"))
assert event.created_at == datetime(2026, 1, 2, 3, 4, 5, tzinfo=timezone.utc)


def test_created_at_accepts_snake_case_key():
payload = _api_response()
del payload["createdAt"]
payload["created_at"] = "2026-07-20T12:00:00Z"

event = WebhookEvent.from_dict(payload)
assert event.created_at == datetime(2026, 7, 20, 12, 0, tzinfo=timezone.utc)


def test_invalid_created_at_raises():
with pytest.raises(InvalidRequestError) as exc_info:
WebhookEvent.from_dict(_api_response(createdAt="not-a-timestamp"))

assert "createdAt" in exc_info.value.field_errors


@pytest.mark.parametrize("livemode,expected", [(True, True), (False, False)])
def test_livemode_reflects_payload(livemode, expected):
event = WebhookEvent.from_dict(_api_response(livemode=livemode))
assert event.livemode is expected


def test_livemode_string_value_raises():
with pytest.raises(InvalidRequestError):
WebhookEvent.from_dict(_api_response(livemode="false"))


@pytest.mark.parametrize("field", ["id", "type", "data", "createdAt", "livemode"])
def test_missing_required_field_raises(field):
payload = _api_response()
del payload[field]

with pytest.raises(InvalidRequestError):
WebhookEvent.from_dict(payload)


def test_non_dict_payload_raises():
with pytest.raises(InvalidRequestError):
WebhookEvent.from_dict("not-a-payload")


def test_unknown_fields_are_preserved():
event = WebhookEvent.from_dict(_api_response(apiVersion="2026-07-01"))
assert event.apiVersion == "2026-07-01"


def test_to_dict_round_trips_by_alias():
event = WebhookEvent.from_dict(_api_response())
dumped = event.to_dict()

assert dumped["id"] == "evt_123"
assert dumped["createdAt"] == datetime(2026, 7, 20, 12, 0, tzinfo=timezone.utc)
assert dumped["data"] == PAYMENT_DATA
assert dumped["livemode"] is True


def test_repr_shows_event_id():
event = WebhookEvent.from_dict(_api_response())
assert repr(event) == "<WebhookEvent id='evt_123'>"


def test_event_type_constants_compare_to_wire_strings():
assert WebhookEventType.PAYMENT_COMPLETED == "payment.completed"
assert WebhookEventType.INVOICE_PAID == "invoice.paid"
assert WebhookEventType.SWAP_SLIPPAGE_EXCEEDED == "swap.slippage_exceeded"


def test_event_type_usable_in_conditionals():
event = WebhookEvent.from_dict(_api_response())
assert event.type == WebhookEventType.PAYMENT_COMPLETED
assert event.type != WebhookEventType.PAYMENT_EXPIRED


def test_unknown_event_type_still_parses():
event = WebhookEvent.from_dict(_api_response(type="payment.refunded"))
assert event.type == "payment.refunded"


def test_exported_from_package_root():
assert shade.WebhookEvent is WebhookEvent
assert shade.WebhookEventType is WebhookEventType
Loading