# Upstash Python QStash SDK
**QStash** is an HTTP based messaging and scheduling solution for serverless and edge runtimes.
[QStash Documentation](https://upstash.com/docs/qstash)
### Usage
You can get your QStash token from the [Upstash Console](https://console.upstash.com/qstash).
#### Publish a JSON message
```python
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](https://upstash.com/docs/qstash/features/schedules)
```python
from upstash_qstash import Client
client = Client("QSTASH_TOKEN")
schedules = client.schedules()
schedules.create({
"destination": "https://my-api...",
"cron": "*/5 * * * *",
})
```
#### [Receiving messages](https://upstash.com/docs/qstash/howto/receiving)
```python
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
```python
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
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` and examples with `python3 examples/<example>.py`
6. Format with `poetry run black .`
Raw data
{
"_id": null,
"home_page": "https://github.com/upstash/qstash-python",
"name": "qstash-python",
"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/b5/5e/8dc505f215432ee75a44b1c73bbc6d6f817ecd46a7a0f6180ff665609aee/qstash_python-1.1.0.tar.gz",
"platform": null,
"description": "# Upstash Python QStash SDK\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### Usage\n\nYou can get your QStash token from the [Upstash Console](https://console.upstash.com/qstash).\n\n#### Publish a JSON message\n```python\nfrom upstash_qstash import Client\n\nclient = Client(\"QSTASH_TOKEN\")\nclient.publish_json(\n {\n \"url\": \"https://my-api...\",\n \"body\": {\n \"hello\": \"world\"\n },\n \"headers\": {\n \"test-header\": \"test-value\",\n },\n }\n)\n```\n\n#### [Create a scheduled message](https://upstash.com/docs/qstash/features/schedules)\n```python\nfrom upstash_qstash import Client\n\nclient = Client(\"QSTASH_TOKEN\")\nschedules = client.schedules()\nschedules.create({\n \"destination\": \"https://my-api...\",\n \"cron\": \"*/5 * * * *\",\n})\n```\n\n#### [Receiving messages](https://upstash.com/docs/qstash/howto/receiving)\n```python\nfrom upstash_qstash import Client\n\n# Keys available from the QStash console\nreceiver = Receiver(\n {\n \"current_signing_key\": \"CURRENT_SIGNING_KEY\",\n \"next_signing_key\": \"NEXT_SIGNING_KEY\",\n }\n)\n\nverified = receiver.verify(\n {\n \"signature\": req.headers[\"Upstash-Signature\"],\n \"body\": req.body,\n \"url\": \"https://my-api...\", # Optional\n }\n)\n```\n\n#### Additional configuration\n```python\nfrom upstash_qstash import Client\n\nclient = Client(\"QSTASH_TOKEN\")\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 # \"attempts\": 6,\n # \"backoff\": lambda retry_count: math.exp(retry_count) * 50,\n # }\n client = Client(\"QSTASH_TOKEN\", {\n \"attempts\": 2,\n \"backoff\": lambda retry_count: (2 ** retry_count) * 20,\n })\n \n # Create Topic\n topics = client.topics()\n topics.upsert_or_add_endpoints(\"my-topic\", [\n {\n \"name\": \"endpoint1\",\n \"url\": \"https://example.com\"\n },\n {\n \"name\": \"endpoint2\",\n \"url\": \"https://somewhere-else.com\"\n }\n ])\n\n # Publish to Topic\n client.publish_json({\n \"topic\": \"my-topic\",\n \"body\": {\n \"key\": \"value\"\n },\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\": 4, \n # When message is sent, send a request to this URL\n # https://upstash.com/docs/qstash/features/callbacks\n \"callback\": \"https://my-api.com/callback\",\n # When message fails to send, send a request to this URL\n \"failure_callback\": \"https://my-api.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 topics, schedules, and messages. See the examples folder for more.\n\n### Development\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` and examples with `python3 examples/<example>.py`\n6. Format with `poetry run black .`\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "Python client for Upstash QStash",
"version": "1.1.0",
"project_urls": {
"Homepage": "https://github.com/upstash/qstash-python",
"Repository": "https://github.com/upstash/qstash-python"
},
"split_keywords": [
"qstash",
" upstash qstash",
" serverless queue"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "4315eb29e85f77943180164389b8a40ce96bc53155c78a24d548a68a6b22a92f",
"md5": "75413b0a6f1c124fcade0e795f8e3013",
"sha256": "34fac282edd85786a8d10f140983c469ea6aa4c3eb8d4ec740252511cb7acf0d"
},
"downloads": -1,
"filename": "qstash_python-1.1.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "75413b0a6f1c124fcade0e795f8e3013",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<4.0,>=3.8",
"size": 25171,
"upload_time": "2024-04-25T21:45:50",
"upload_time_iso_8601": "2024-04-25T21:45:50.600263Z",
"url": "https://files.pythonhosted.org/packages/43/15/eb29e85f77943180164389b8a40ce96bc53155c78a24d548a68a6b22a92f/qstash_python-1.1.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "b55e8dc505f215432ee75a44b1c73bbc6d6f817ecd46a7a0f6180ff665609aee",
"md5": "fdf5a05a49681a738c7878fab4f659f6",
"sha256": "5c976e8b3c963da6b0b2566e894edbccf0d39881e8ab4d9eaf50b54751f98d07"
},
"downloads": -1,
"filename": "qstash_python-1.1.0.tar.gz",
"has_sig": false,
"md5_digest": "fdf5a05a49681a738c7878fab4f659f6",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<4.0,>=3.8",
"size": 16014,
"upload_time": "2024-04-25T21:45:52",
"upload_time_iso_8601": "2024-04-25T21:45:52.129215Z",
"url": "https://files.pythonhosted.org/packages/b5/5e/8dc505f215432ee75a44b1c73bbc6d6f817ecd46a7a0f6180ff665609aee/qstash_python-1.1.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-04-25 21:45:52",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "upstash",
"github_project": "qstash-python",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "qstash-python"
}
JSON
