Publish - Upstash Documentation

Info

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

By default, the delay between retries is calculated using an exponential backoff algorithm. You can customize this using the retryDelay parameter. Check out the retries page to learn more about custom retry delay values.

Publish HTML content instead of JSON#

from qstash import QStashclient = 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 QStashclient = 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 QStashclient = QStash("<QSTASH-TOKEN>")client.message.publish_json(    url="https://my-api...",    body={        "hello": "world",    },    timeout="30s",)

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

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

Publish a method with a callback URL#

Configure the number of retries#

Publish HTML content instead of JSON#

Publish a message with content-based-deduplication#

Publish a message with timeout#

Publish a message with content-based-deduplication #