convoy-python

This is the official Convoy Python SDK. It contains methods for easily interacting with Convoy's API. Below are examples to get you started. See our API Reference for more.

Installation

Install convoy-python with

pip install convoy-python

Setup Client

Import the convoy module and set it up with your instance URL, API key, and project ID. Both the API key and project ID are available from your Project Settings page.

from convoy import Convoy

convoy = Convoy({
    "api_key": "your_api_key",
    "uri": "https://us.getconvoy.cloud/api/v1",
    "project_id": "your_project_id",
})

Your instance URL depends on where your project lives:

• Convoy Cloud (US): https://us.getconvoy.cloud/api/v1
• Convoy Cloud (EU): https://eu.getconvoy.cloud/api/v1
• Self-hosted: https://your-instance/api/v1

Usage

Each method takes a query dict and returns a (response, status) tuple.

Create an Endpoint

An endpoint represents a target URL to receive events.

endpoint_data = {
    "name": "default-endpoint",
    "url": "https://example.com/webhooks/convoy",
    "description": "Default Endpoint",
    "secret": "endpoint-secret",
}

(response, status) = convoy.endpoint.create({}, endpoint_data)
endpoint_id = response["data"]["uid"]

Create a Subscription

Subscriptions route events from a source to an endpoint.

subscription_data = {
    "name": "event-sub",
    "endpoint_id": endpoint_id,
}

(response, status) = convoy.subscription.create({}, subscription_data)

Send an Event

To send an event, you'll need the uid of the endpoint we created earlier.

event_data = {
    "endpoint_id": endpoint_id,
    "event_type": "payment.success",
    "data": {
        "status": "Completed",
        "description": "Transaction Successful",
    },
}

(response, status) = convoy.event.create({}, event_data)

To fan an event out to all endpoints with the same owner_id, or broadcast to every endpoint in the project:

(response, status) = convoy.event.fanout({}, {"owner_id": "owner-1", "event_type": "payment.success", "data": {}})
(response, status) = convoy.event.broadcast({}, {"event_type": "payment.success", "data": {}})

Verify Webhook Signatures

Verify with the raw request body, before parsing it. On failure the helper returns an error message (a truthy string), so compare against True explicitly.

from convoy.utils.webhook import Webhook

webhook = Webhook(secret="endpoint-secret")

payload = request.body.decode("utf-8")
signature = request.headers.get("X-Convoy-Signature", "")

if webhook.verify_signature(payload, signature) is not True:
    # reject the request
    ...

Testing

pytest test/test.py

Contributing

Please see CONTRIBUTING for details.

Credits

• Frain

License

The MIT License (MIT). Please see License File for more information.