# 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.
[QStash Documentation](https://upstash.com/docs/qstash)
### Install
```shell
pip install qstash
```
### Usage
You can get your QStash token from the [Upstash Console](https://console.upstash.com/qstash).
#### Publish a JSON message
```python
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](https://upstash.com/docs/qstash/features/schedules)
```python
from qstash import QStash
client = QStash("<QSTASH_TOKEN>")
schedule_id = client.schedule.create(
destination="https://example.com",
cron="*/5 * * * *",
)
print(schedule_id)
```
#### [Receiving messages](https://upstash.com/docs/qstash/howto/receiving)
```python
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
```python
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
```python
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
```python
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
```python
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
```python
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
1. Clone the repository
2. Install [Poetry](https://python-poetry.org/docs/#installation)
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`
6. Format with `poetry run ruff format .`
Raw data
{
"_id": null,
"home_page": "https://github.com/upstash/qstash-py",
"name": "qstash",
"maintainer": "Upstash",
"docs_url": null,
"requires_python": "<4.0,>=3.8",
"maintainer_email": "support@upstash.com",
"keywords": "QStash, Upstash QStash, Serverless Queue",
"author": "Upstash",
"author_email": "support@upstash.com",
"download_url": "https://files.pythonhosted.org/packages/e6/2c/4deb192c222da9f79f48221e43351641f45dc0aec66b79f9daf3c7f1d55a/qstash-2.0.2.tar.gz",
"platform": null,
"description": "# Upstash Python QStash SDK\n\n> [!NOTE] \n> **This project is in GA Stage.**\n>\n> The Upstash Professional Support fully covers this project. It receives regular updates, and bug fixes.\n> The Upstash team is committed to maintaining and improving its functionality.\n\n**QStash** is an HTTP based messaging and scheduling solution for serverless and edge runtimes.\n\n[QStash Documentation](https://upstash.com/docs/qstash)\n\n### Install\n\n```shell\npip install qstash\n```\n\n### Usage\n\nYou can get your QStash token from the [Upstash Console](https://console.upstash.com/qstash).\n\n#### Publish a JSON message\n\n```python\nfrom qstash import QStash\n\nclient = QStash(\"<QSTASH_TOKEN>\")\n\nres = client.message.publish_json(\n url=\"https://example.com\",\n body={\"hello\": \"world\"},\n headers={\n \"test-header\": \"test-value\",\n },\n)\n\nprint(res.message_id)\n```\n\n#### [Create a scheduled message](https://upstash.com/docs/qstash/features/schedules)\n\n```python\nfrom qstash import QStash\n\nclient = QStash(\"<QSTASH_TOKEN>\")\n\nschedule_id = client.schedule.create(\n destination=\"https://example.com\",\n cron=\"*/5 * * * *\",\n)\n\nprint(schedule_id)\n```\n\n#### [Receiving messages](https://upstash.com/docs/qstash/howto/receiving)\n\n```python\nfrom qstash import Receiver\n\n# Keys available from the QStash console\nreceiver = Receiver(\n current_signing_key=\"CURRENT_SIGNING_KEY\",\n next_signing_key=\"NEXT_SIGNING_KEY\",\n)\n\n# ... in your request handler\n\nsignature, body = req.headers[\"Upstash-Signature\"], req.body\n\nreceiver.verify(\n body=body,\n signature=signature,\n url=\"https://example.com\", # Optional\n)\n```\n\n#### Create Chat Completions\n\n```python\nfrom qstash import QStash\nfrom qstash.chat import upstash\n\nclient = QStash(\"<QSTASH_TOKEN>\")\n\nres = client.chat.create(\n model=\"meta-llama/Meta-Llama-3-8B-Instruct\",\n provider=upstash(),\n messages=[\n {\n \"role\": \"user\",\n \"content\": \"What is the capital of Turkey?\",\n }\n ],\n)\n\nprint(res.choices[0].message.content)\n```\n\n#### Create Chat Completions Using Custom Providers\n\n```python\nfrom qstash import QStash\nfrom qstash.chat import openai\n\nclient = QStash(\"<QSTASH_TOKEN>\")\n\nres = client.chat.create(\n model=\"gpt-3.5-turbo\",\n provider=openai(\"<OPENAI_API_KEY>\"),\n messages=[\n {\n \"role\": \"user\",\n \"content\": \"What is the capital of Turkey?\",\n }\n ],\n)\n\nprint(res.choices[0].message.content)\n```\n\n#### Publish a JSON message to LLM\n\n```python\nfrom qstash import QStash\nfrom qstash.chat import upstash\n\nclient = QStash(\"<QSTASH_TOKEN>\")\n\nres = client.message.publish_json(\n api={\"name\": \"llm\", \"provider\": upstash()},\n body={\n \"model\": \"meta-llama/Meta-Llama-3-8B-Instruct\",\n \"messages\": [\n {\n \"role\": \"user\",\n \"content\": \"What is the capital of Turkey?\",\n }\n ],\n },\n callback=\"https://example-cb.com\",\n)\n\nprint(res.message_id)\n```\n\n#### Publish a JSON message to LLM Using Custom Providers\n\n```python\nfrom qstash import QStash\nfrom qstash.chat import openai\n\nclient = QStash(\"<QSTASH_TOKEN>\")\n\nres = client.message.publish_json(\n api={\"name\": \"llm\", \"provider\": openai(\"<OPENAI_API_KEY>\")},\n body={\n \"model\": \"gpt-3.5-turbo\",\n \"messages\": [\n {\n \"role\": \"user\",\n \"content\": \"What is the capital of Turkey?\",\n }\n ],\n },\n callback=\"https://example-cb.com\",\n)\n\nprint(res.message_id)\n```\n\n#### Additional configuration\n\n```python\nfrom qstash import QStash\n\n# Create a client with a custom retry configuration. This is\n# for sending messages to QStash, not for sending messages to\n# your endpoints.\n# The default configuration is:\n# {\n# \"retries\": 5,\n# \"backoff\": lambda retry_count: math.exp(retry_count) * 50,\n# }\nclient = QStash(\n token=\"<QSTASH_TOKEN>\",\n retry={\n \"retries\": 1,\n \"backoff\": lambda retry_count: (2 ** retry_count) * 20,\n },\n)\n\n# Publish to URL\nclient.message.publish_json(\n url=\"https://example.com\",\n body={\"key\": \"value\"},\n # Retry sending message to API 3 times\n # https://upstash.com/docs/qstash/features/retry\n retries=3,\n # Schedule message to be sent 4 seconds from now\n delay=\"4s\",\n # When message is sent, send a request to this URL\n # https://upstash.com/docs/qstash/features/callbacks\n callback=\"https://example.com/callback\",\n # When message fails to send, send a request to this URL\n failure_callback=\"https://example.com/failure_callback\",\n # Headers to forward to the endpoint\n headers={\n \"test-header\": \"test-value\",\n },\n # Enable content-based deduplication\n # https://upstash.com/docs/qstash/features/deduplication#content-based-deduplication\n content_based_deduplication=True,\n)\n```\n\nAdditional methods are available for managing url groups, schedules, and messages. See the examples folder for more.\n\n### Development\n\n1. Clone the repository\n2. Install [Poetry](https://python-poetry.org/docs/#installation)\n3. Install dependencies with `poetry install`\n4. Create a .env file with `cp .env.example .env` and fill in the `QSTASH_TOKEN`\n5. Run tests with `poetry run pytest`\n6. Format with `poetry run ruff format .`\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "Python SDK for Upstash QStash",
"version": "2.0.2",
"project_urls": {
"Homepage": "https://github.com/upstash/qstash-py",
"Repository": "https://github.com/upstash/qstash-py"
},
"split_keywords": [
"qstash",
" upstash qstash",
" serverless queue"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "f0cc9becdd74de8775f4066c02c9695a60cbd4b760a1ceabeec58a7ed92c8f70",
"md5": "a474d25b5b44ff644a237975701186c7",
"sha256": "ae71d4c071b8170cc3829ac32ad38288811b8c06e0e70cab312e36e28796e366"
},
"downloads": -1,
"filename": "qstash-2.0.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "a474d25b5b44ff644a237975701186c7",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<4.0,>=3.8",
"size": 41621,
"upload_time": "2024-12-24T16:30:19",
"upload_time_iso_8601": "2024-12-24T16:30:19.433225Z",
"url": "https://files.pythonhosted.org/packages/f0/cc/9becdd74de8775f4066c02c9695a60cbd4b760a1ceabeec58a7ed92c8f70/qstash-2.0.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "e62c4deb192c222da9f79f48221e43351641f45dc0aec66b79f9daf3c7f1d55a",
"md5": "340da8b72c733b1ca26e48d3bc1d529c",
"sha256": "af0ae1021e433447b010db9e9faeecc8b65743a56f5c5645acb39e22d4e5cf87"
},
"downloads": -1,
"filename": "qstash-2.0.2.tar.gz",
"has_sig": false,
"md5_digest": "340da8b72c733b1ca26e48d3bc1d529c",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<4.0,>=3.8",
"size": 31552,
"upload_time": "2024-12-24T16:30:21",
"upload_time_iso_8601": "2024-12-24T16:30:21.752847Z",
"url": "https://files.pythonhosted.org/packages/e6/2c/4deb192c222da9f79f48221e43351641f45dc0aec66b79f9daf3c7f1d55a/qstash-2.0.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-12-24 16:30:21",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "upstash",
"github_project": "qstash-py",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "qstash"
}
JSON
