Python client for Upstash QStash
Project description
Upstash Python QStash SDK
QStash is an HTTP based messaging and scheduling solution for serverless and edge runtimes.
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
- Clone the repository
- Install Poetry
- Install dependencies with
poetry install
- Create a .env file with
cp .env.example .env
and fill in theQSTASH_TOKEN
- Run tests with
poetry run pytest
and examples withpython3 examples/<example>.py
- Format with
poetry run black .
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
qstash_python-1.0.0.tar.gz
(13.2 kB
view hashes)
Built Distribution
Close
Hashes for qstash_python-1.0.0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 503afda80bc6454527d7a99167d6facd577fae61adc87629f139f762e6f360bf |
|
MD5 | ee2e3e25792326a5dcc3defd677d8841 |
|
BLAKE2b-256 | cdb17b7a13f4e9e17cf662306f57eb21360d61183241e7a8f8785b469dae8f3a |