Python SDK for Upstash QStash
Project description
Upstash Python QStash SDK
[!NOTE]
This project is in GA Stage.The Upstash Professional Support fully covers this project. It receives regular updates, and bug fixes. The Upstash team is committed to maintaining and improving its functionality.
QStash is an HTTP based messaging and scheduling solution for serverless and edge runtimes.
Install
pip install qstash
Usage
You can get your QStash token from the Upstash Console.
Publish a JSON message
from qstash import QStash
client = QStash("<QSTASH_TOKEN>")
res = client.message.publish_json(
url="https://example.com",
body={"hello": "world"},
headers={
"test-header": "test-value",
},
)
print(res.message_id)
Create a scheduled message
from qstash import QStash
client = QStash("<QSTASH_TOKEN>")
schedule_id = client.schedule.create(
destination="https://example.com",
cron="*/5 * * * *",
)
print(schedule_id)
Receiving messages
from qstash import Receiver
# Keys available from the QStash console
receiver = Receiver(
current_signing_key="CURRENT_SIGNING_KEY",
next_signing_key="NEXT_SIGNING_KEY",
)
# ... in your request handler
signature, body = req.headers["Upstash-Signature"], req.body
receiver.verify(
body=body,
signature=signature,
url="https://example.com", # Optional
)
Create Chat Completions
from qstash import QStash
from qstash.chat import upstash
client = QStash("<QSTASH_TOKEN>")
res = client.chat.create(
model="meta-llama/Meta-Llama-3-8B-Instruct",
provider=upstash(),
messages=[
{
"role": "user",
"content": "What is the capital of Turkey?",
}
],
)
print(res.choices[0].message.content)
Create Chat Completions Using Custom Providers
from qstash import QStash
from qstash.chat import openai
client = QStash("<QSTASH_TOKEN>")
res = client.chat.create(
model="gpt-3.5-turbo",
provider=openai("<OPENAI_API_KEY>"),
messages=[
{
"role": "user",
"content": "What is the capital of Turkey?",
}
],
)
print(res.choices[0].message.content)
Publish a JSON message to LLM
from qstash import QStash
from qstash.chat import upstash
client = QStash("<QSTASH_TOKEN>")
res = client.message.publish_json(
api={"name": "llm", "provider": upstash()},
body={
"model": "meta-llama/Meta-Llama-3-8B-Instruct",
"messages": [
{
"role": "user",
"content": "What is the capital of Turkey?",
}
],
},
callback="https://example-cb.com",
)
print(res.message_id)
Publish a JSON message to LLM Using Custom Providers
from qstash import QStash
from qstash.chat import openai
client = QStash("<QSTASH_TOKEN>")
res = client.message.publish_json(
api={"name": "llm", "provider": openai("<OPENAI_API_KEY>")},
body={
"model": "gpt-3.5-turbo",
"messages": [
{
"role": "user",
"content": "What is the capital of Turkey?",
}
],
},
callback="https://example-cb.com",
)
print(res.message_id)
Additional configuration
from qstash import QStash
# 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:
# {
# "retries": 5,
# "backoff": lambda retry_count: math.exp(retry_count) * 50,
# }
client = QStash(
token="<QSTASH_TOKEN>",
retry={
"retries": 1,
"backoff": lambda retry_count: (2 ** retry_count) * 20,
},
)
# Publish to URL
client.message.publish_json(
url="https://example.com",
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="4s",
# When message is sent, send a request to this URL
# https://upstash.com/docs/qstash/features/callbacks
callback="https://example.com/callback",
# When message fails to send, send a request to this URL
failure_callback="https://example.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 url groups, 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
- Format with
poetry run ruff format .
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-2.0.0.tar.gz
(31.4 kB
view details)
Built Distribution
qstash-2.0.0-py3-none-any.whl
(41.4 kB
view details)
File details
Details for the file qstash-2.0.0.tar.gz
.
File metadata
- Download URL: qstash-2.0.0.tar.gz
- Upload date:
- Size: 31.4 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via: poetry/1.8.3 CPython/3.8.18 Linux/6.5.0-1023-azure
File hashes
Algorithm | Hash digest | |
---|---|---|
SHA256 | 507942977a4880104228aa3cec6dc9beba6e0dd3c0c7f69361666195bbba1891 |
|
MD5 | af02fa337739420bd04090901f3b17a2 |
|
BLAKE2b-256 | ace1da9bb47752d4bdf1ef8b6d28bef043e2d5ef89290083a947947ffad06eb0 |
Provenance
File details
Details for the file qstash-2.0.0-py3-none-any.whl
.
File metadata
- Download URL: qstash-2.0.0-py3-none-any.whl
- Upload date:
- Size: 41.4 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via: poetry/1.8.3 CPython/3.8.18 Linux/6.5.0-1023-azure
File hashes
Algorithm | Hash digest | |
---|---|---|
SHA256 | b0eaca6c97264a1f7018f88a3ab9278c01b9215e9b1c4262b874f5aa01a329ce |
|
MD5 | e75c8422f9577f39586ce6559a7ead09 |
|
BLAKE2b-256 | 32922346ddb28bdc865682484abb5a6c2cc2985762f7d2d8cb4708345dac4e97 |