upstash-workflow


Nameupstash-workflow JSON
Version 0.1.0 PyPI version JSON
download
home_pagehttps://github.com/upstash/workflow-py
SummaryPython SDK for Upstash Workflow
upload_time2025-01-29 15:06:44
maintainerUpstash
docs_urlNone
authorUpstash
requires_python<4.0,>=3.8
licenseMIT
keywords workflow upstash workflow durable serverless functions
VCS
bugtrack_url
requirements No requirements were recorded.
Travis-CI No Travis.
coveralls test coverage No coveralls.
            # 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"
}
        
Elapsed time: 1.57854s