You can run the async code by importing AsyncQStash from qstash and awaiting the methods.

​
Publish to a URL with a 3 second delay and headers/body

from qstash import QStash

client = QStash("<QSTASH-TOKEN>")
res = client.message.publish_json(
    url="https://my-api...",
    body={
        "hello": "world",
    },
    headers={
        "test-header": "test-value",
    },
    delay="3s",
)

print(res.message_id)

​
Publish to a URL group with a 3 second delay and headers/body

You can make a URL group on the QStash console or using the URL group API

from qstash import QStash

client = QStash("<QSTASH-TOKEN>")
res = client.message.publish_json(
    url_group="my-url-group",
    body={
        "hello": "world",
    },
    headers={
        "test-header": "test-value",
    },
    delay="3s",
)

# When publishing to a URL group, the response is an array of messages for each URL in the group
print(res[0].message_id)

​
Publish a method with a callback URL

Callbacks are useful for long running functions. Here, QStash will return the response of the publish request to the callback URL.

We also change the method to GET in this use case so QStash will make a GET request to the url. The default is POST.

from qstash import QStash

client = QStash("<QSTASH-TOKEN>")
client.message.publish_json(
    url="https://my-api...",
    body={
        "hello": "world",
    },
    callback="https://my-callback...",
    failure_callback="https://my-failure-callback...",
    method="GET",
)

​
Configure the number of retries

The max number of retries is based on your QStash plan

from qstash import QStash

client = QStash("<QSTASH-TOKEN>")
client.message.publish_json(
    url="https://my-api...",
    body={
        "hello": "world",
    },
    retries=1,
)

​
Publish HTML content instead of JSON

from qstash import QStash

client = QStash("<QSTASH-TOKEN>")
client.message.publish(
    url="https://my-api...",
    body="<html><body><h1>Hello World</h1></body></html>",
    content_type="text/html",
)

​
Publish a message with content-based-deduplication

from qstash import QStash

client = QStash("<QSTASH-TOKEN>")
client.message.publish_json(
    url="https://my-api...",
    body={
        "hello": "world",
    },
    content_based_deduplication=True,
)

​
Publish a message with timeout

Timeout value to use when calling a url (See Upstash-Timeout in Publish Message page)

from qstash import QStash

client = QStash("<QSTASH-TOKEN>")
client.message.publish_json(
    url="https://my-api...",
    body={
        "hello": "world",
    },
    timeout="30s",
)

Was this page helpful?

OverviewSchedules