qstash-python 1.1.0

Python client for Upstash QStash

Upstash Python QStash SDK

QStash is an HTTP based messaging and scheduling solution for serverless and edge runtimes.

QStash Documentation

Usage

You can get your QStash token from the Upstash Console.

Publish a JSON message

from upstash_qstash import Client

client = Client("QSTASH_TOKEN")
client.publish_json(
  {
    "url": "https://my-api...",
    "body": {
      "hello": "world"
    },
    "headers": {
        "test-header": "test-value",
    },
  }
)

Create a scheduled message

from upstash_qstash import Client

client = Client("QSTASH_TOKEN")
schedules = client.schedules()
schedules.create({
  "destination": "https://my-api...",
  "cron": "*/5 * * * *",
})

Receiving messages

from upstash_qstash import Client

# Keys available from the QStash console
receiver = Receiver(
  {
    "current_signing_key": "CURRENT_SIGNING_KEY",
    "next_signing_key": "NEXT_SIGNING_KEY",
  }
)

verified = receiver.verify(
  {
    "signature": req.headers["Upstash-Signature"],
    "body": req.body,
    "url": "https://my-api...", # Optional
  }
)

Additional configuration

from upstash_qstash import Client

client = Client("QSTASH_TOKEN")
  # Create a client with a custom retry configuration. This is 
  # for sending messages to QStash, not for sending messages to
  # your endpoints.
  # The default configuration is:
  # {
  #   "attempts": 6,
  #   "backoff": lambda retry_count: math.exp(retry_count) * 50,
  # }
  client = Client("QSTASH_TOKEN", {
    "attempts": 2,
    "backoff": lambda retry_count: (2 ** retry_count) * 20,
  })
  
  # Create Topic
  topics = client.topics()
  topics.upsert_or_add_endpoints("my-topic", [
    {
      "name": "endpoint1",
      "url": "https://example.com"
    },
    {
      "name": "endpoint2",
      "url": "https://somewhere-else.com"
    }
  ])

  # Publish to Topic
  client.publish_json({
    "topic": "my-topic",
    "body": {
      "key": "value"
    },
    # Retry sending message to API 3 times
    # https://upstash.com/docs/qstash/features/retry
    "retries": 3,
    # Schedule message to be sent 4 seconds from now
    "delay": 4, 
    # When message is sent, send a request to this URL
    # https://upstash.com/docs/qstash/features/callbacks
    "callback": "https://my-api.com/callback",
    # When message fails to send, send a request to this URL
    "failure_callback": "https://my-api.com/failure_callback",
    # Headers to forward to the endpoint
    "headers": {
      "test-header": "test-value",
    },
    # Enable content-based deduplication
    # https://upstash.com/docs/qstash/features/deduplication#content-based-deduplication
    "content_based_deduplication": True,
  })

Additional methods are available for managing topics, schedules, and messages. See the examples folder for more.

Development

1. Clone the repository
2. Install Poetry
3. Install dependencies with poetry install
4. Create a .env file with cp .env.example .env and fill in the QSTASH_TOKEN
5. Run tests with poetry run pytest and examples with python3 examples/<example>.py
6. Format with poetry run black .