# Upstash Workflow SDK
**Upstash Workflow** lets you write durable, reliable and performant serverless functions. Get delivery guarantees, automatic retries on failure, scheduling and more without managing any infrastructure.
See [the documentation](https://upstash.com/docs/workflow/getstarted) for more details
## Quick Start
Here, we will briefly showcase how you can get started with Upstash Workflow using FastAPI.
Alternatively, you can check [our quickstarts for different frameworks](https://upstash.com/docs/workflow/quickstarts/platforms), including [FastAPI](https://upstash.com/docs/workflow/quickstarts/fastapi) and [Next.js & FastAPI](https://upstash.com/docs/workflow/quickstarts/nextjs-fastapi).
### Install
First, create a new directory and set up a virtual environment:
```sh
python -m venv venv
source venv/bin/activate
```
Then, install the required packages:
```sh
pip install fastapi uvicorn upstash-workflow
```
### Get QStash token
Go to [Upstash Console](https://console.upstash.com/qstash) and copy the `QSTASH_TOKEN`, set it in the `.env` file.
```sh
export QSTASH_TOKEN=
```
### Define a Workflow Endpoint
To declare workflow endpoints, use the `@serve.post` decorator. Save the following code to `main.py`:
```python
from fastapi import FastAPI
from upstash_workflow.fastapi import Serve
from upstash_workflow import AsyncWorkflowContext
app = FastAPI()
serve = Serve(app)
# mock function
def some_work(input: str) -> str:
return f"processed '{input}'"
# serve endpoint which expects a string payload:
@serve.post("/example")
async def example(context: AsyncWorkflowContext[str]) -> None:
# get request body:
input = context.request_payload
async def _step1() -> str:
output = some_work(input)
print("step 1 input", input, "output", output)
return output
# run the first step:
result: str = await context.run("step1", _step1)
async def _step2() -> None:
output = some_work(result)
print("step 2 input", result, "output", output)
# run the second step:
await context.run("step2", _step2)
```
In the example, you can see that steps are declared through the `context` object.
The kinds of steps which are available are:
* `context.run`: execute a function
* `context.sleep`: sleep for some time
* `context.sleep_until`: sleep until some timestamp
* `context.call`: make a third party call without consuming any runtime
You can [learn more about these methods from our documentation](https://upstash.com/docs/workflow/basics/context).
### Run the Server
Upstash Workflow needs a public URL to orchestrate the workflow. Check out our [Local Development](https://upstash.com/docs/workflow/howto/local-development) guide to learn how to set up a local tunnel.
Create the tunnel and set the `UPSTASH_WORKFLOW_URL` environment variable in the `.env` file with the public URL:
```sh
ngrok http localhost:8000
```
```sh
export UPSTASH_WORKFLOW_URL=
```
Then, set the environment variables:
```sh
source .env
```
Finally, run the server:
```sh
uvicorn main:app --reload
```
FastAPI server will be running at `localhost:8000`.
## Contributing
### 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 environment variables
5. Run tests with `poetry run pytest`
6. Format with `poetry run ruff format .`
7. Check with `poetry run ruff check .`
8. Type check with `poetry run mypy --show-error-codes .`
Raw data
{
"_id": null,
"home_page": "https://github.com/upstash/workflow-py",
"name": "upstash-workflow",
"maintainer": "Upstash",
"docs_url": null,
"requires_python": "<4.0,>=3.8",
"maintainer_email": "support@upstash.com",
"keywords": "Workflow, Upstash Workflow, Durable Serverless Functions",
"author": "Upstash",
"author_email": "support@upstash.com",
"download_url": "https://files.pythonhosted.org/packages/f9/68/c68cd2cbd6b0b04210c7d3ba6e07bee6d2ab27ded72c723563a2c570c8d2/upstash_workflow-0.1.0.tar.gz",
"platform": null,
"description": "# Upstash Workflow SDK\n\n**Upstash Workflow** lets you write durable, reliable and performant serverless functions. Get delivery guarantees, automatic retries on failure, scheduling and more without managing any infrastructure.\n\nSee [the documentation](https://upstash.com/docs/workflow/getstarted) for more details\n\n## Quick Start\n\nHere, we will briefly showcase how you can get started with Upstash Workflow using FastAPI.\n\nAlternatively, you can check [our quickstarts for different frameworks](https://upstash.com/docs/workflow/quickstarts/platforms), including [FastAPI](https://upstash.com/docs/workflow/quickstarts/fastapi) and [Next.js & FastAPI](https://upstash.com/docs/workflow/quickstarts/nextjs-fastapi).\n\n### Install\n\nFirst, create a new directory and set up a virtual environment:\n\n```sh\npython -m venv venv\nsource venv/bin/activate\n```\n\nThen, install the required packages:\n\n```sh\npip install fastapi uvicorn upstash-workflow\n```\n\n### Get QStash token\n\nGo to [Upstash Console](https://console.upstash.com/qstash) and copy the `QSTASH_TOKEN`, set it in the `.env` file.\n\n```sh\nexport QSTASH_TOKEN=\n```\n\n### Define a Workflow Endpoint\n\nTo declare workflow endpoints, use the `@serve.post` decorator. Save the following code to `main.py`:\n\n```python\nfrom fastapi import FastAPI\nfrom upstash_workflow.fastapi import Serve\nfrom upstash_workflow import AsyncWorkflowContext\n\napp = FastAPI()\nserve = Serve(app)\n\n# mock function\ndef some_work(input: str) -> str:\n return f\"processed '{input}'\"\n\n# serve endpoint which expects a string payload:\n@serve.post(\"/example\")\nasync def example(context: AsyncWorkflowContext[str]) -> None:\n # get request body:\n input = context.request_payload\n\n async def _step1() -> str:\n output = some_work(input)\n print(\"step 1 input\", input, \"output\", output)\n return output\n\n # run the first step:\n result: str = await context.run(\"step1\", _step1)\n\n async def _step2() -> None:\n output = some_work(result)\n print(\"step 2 input\", result, \"output\", output)\n\n # run the second step:\n await context.run(\"step2\", _step2)\n```\n\nIn the example, you can see that steps are declared through the `context` object.\n\nThe kinds of steps which are available are:\n\n* `context.run`: execute a function\n* `context.sleep`: sleep for some time\n* `context.sleep_until`: sleep until some timestamp\n* `context.call`: make a third party call without consuming any runtime\n\nYou can [learn more about these methods from our documentation](https://upstash.com/docs/workflow/basics/context).\n\n### Run the Server\n\nUpstash Workflow needs a public URL to orchestrate the workflow. Check out our [Local Development](https://upstash.com/docs/workflow/howto/local-development) guide to learn how to set up a local tunnel.\n\nCreate the tunnel and set the `UPSTASH_WORKFLOW_URL` environment variable in the `.env` file with the public URL:\n\n```sh\nngrok http localhost:8000\n```\n\n```sh\nexport UPSTASH_WORKFLOW_URL=\n```\n\nThen, set the environment variables:\n\n```sh\nsource .env\n```\n\nFinally, run the server:\n\n```sh\nuvicorn main:app --reload\n```\n\nFastAPI server will be running at `localhost:8000`.\n\n## Contributing\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 environment variables\n5. Run tests with `poetry run pytest`\n6. Format with `poetry run ruff format .`\n7. Check with `poetry run ruff check .`\n8. Type check with `poetry run mypy --show-error-codes .`\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "Python SDK for Upstash Workflow",
"version": "0.1.0",
"project_urls": {
"Homepage": "https://github.com/upstash/workflow-py",
"Repository": "https://github.com/upstash/workflow-py"
},
"split_keywords": [
"workflow",
" upstash workflow",
" durable serverless functions"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "b5e5316d63970028c03facd53bced3829d540f751f2d88701c581504047a000c",
"md5": "0294cf36ae58515f58dbd18f1f30c7b9",
"sha256": "df33c0033999c9ae701f902e4cf73bc7b0eaf73d75ee86d03250d91e57e706dd"
},
"downloads": -1,
"filename": "upstash_workflow-0.1.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "0294cf36ae58515f58dbd18f1f30c7b9",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<4.0,>=3.8",
"size": 39228,
"upload_time": "2025-01-29T15:06:41",
"upload_time_iso_8601": "2025-01-29T15:06:41.561555Z",
"url": "https://files.pythonhosted.org/packages/b5/e5/316d63970028c03facd53bced3829d540f751f2d88701c581504047a000c/upstash_workflow-0.1.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "f968c68cd2cbd6b0b04210c7d3ba6e07bee6d2ab27ded72c723563a2c570c8d2",
"md5": "9817339a3f0a4b347bdce2a168cbe2d1",
"sha256": "31d35743c8a520195e6b5c4cf1af7b9de2dfe188d40e0055f2405221b3813dc2"
},
"downloads": -1,
"filename": "upstash_workflow-0.1.0.tar.gz",
"has_sig": false,
"md5_digest": "9817339a3f0a4b347bdce2a168cbe2d1",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<4.0,>=3.8",
"size": 27344,
"upload_time": "2025-01-29T15:06:44",
"upload_time_iso_8601": "2025-01-29T15:06:44.076462Z",
"url": "https://files.pythonhosted.org/packages/f9/68/c68cd2cbd6b0b04210c7d3ba6e07bee6d2ab27ded72c723563a2c570c8d2/upstash_workflow-0.1.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-01-29 15:06:44",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "upstash",
"github_project": "workflow-py",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "upstash-workflow"
}
JSON
