# Substantial
[](https://badge.fury.io/py/substantial)
Brokerless durable execution for Python.
> Substantial is part of the
> [Metatype ecosystem](https://github.com/metatypedev/metatype). Consider
> checking out how this component integrates with the whole ecosystem and browse
> the
> [documentation](https://metatype.dev?utm_source=github&utm_medium=readme&utm_campaign=substantial)
> to see more examples.
## What is durable execution?
Durable execution is a programming model where the state of a function is preserved across failures, restarts, or other (un)voluntary disruptions. This ensures that applications can continue execution from their last stable state without losing context or causing additional side effects. It is particularly well-suited for long-running workflows as it enables the management of complex sequences of steps, handling breaks, retries and recovery gracefully.
Substantial is designed around a replay mechanism that reconstructs the function state by re-executing historical events from stored logs. All the logic is embedded in a protocol and there is no centralized broker, allowing it to work with any backend (local files, cloud storage and databases). It aims to be an alternative for use cases that do not require the scale and complexity of [Temporal](https://github.com/temporalio/temporal) or [Durable Task](https://github.com/Azure/durabletask).
## Getting started
```
# pypi
pip install substantial
poetry add substantial
# remote master
pip install --upgrade git+https://github.com/zifeo/substantial.git
poetry add git+https://github.com/zifeo/substantial.git
# local repo/dev
poetry install
pre-commit install
protoc -I . --python_betterproto_out=. protocol/*
```
### Workflow
```py
from dataclasses import dataclass
from datetime import timedelta
import random
from substantial import workflow, Context
def roulette():
if random.random() < 1 / 6:
raise Exception("shot!")
return 100
def multiply(a, b):
return a * b
@dataclass
class State:
cancelled: bool = False
def update(self):
self.cancelled = True
@workflow()
async def example(c: Context):
res = await c.save(roulette)
await c.sleep(timedelta(seconds=10))
res = await c.save(lambda: multiply(res, 0.5))
n = await c.receive("by")
res = await c.save(lambda: multiply(res, n))
s = State(cancelled=False)
c.handle("cancel", lambda _: s.update())
await c.ensure(lambda: s.cancelled)
return res
```
### Worker
Brokerless means that there is no active broker and all the scheduling is solely organized around files/key values. Substantial currently supports Redis, local files and s3-compatible object storages.
```py
import asyncio
from substantial import Conductor
from substantial.backends.fs import FSBackend
from demo.readme_ws import example
async def main():
backend = FSBackend("./demo/logs/readme")
substantial = Conductor(backend)
substantial.register(example)
# run the agent in the background
agent = substantial.run()
# start the workflow
w = await substantial.start(example)
await asyncio.sleep(3)
print(await w.send("by", 2))
await asyncio.sleep(5)
print(await w.send("cancel"))
output = await w.result()
print("Final output", output) # 100
# stop the agent
agent.cancel()
await agent
if __name__ == "__main__":
asyncio.run(main())
```
## API
_Note: not all features are implemented/completed yet._
### Primitives
`save(f: Callable, compensate_with: Optional[Callable]): Any` - memoize the result of a function to avoid re-execution on replay. Functions shall be idempotent as they may be called more than once in case of failure just before the value is persisted. The function can be compensated by providing its inverse effect and trigger later in the workflow with `revert`.
`handle(event_name: str, cb: Callable): None` - register a callback to be executed when a specific event is received. The callbacks are executed in the order they were received and whenever a primitive being called.
`ensure(f: Callable): True` - wait for the function to evaluate to true and schedule a new run when false.
### Higher-level
`sleep` - schedule a replay after a certain amount of time.
`receive` - wait for the value of an event to be received.
`log` - TODO
`datetime.now` - TODO
`random` - TODO
`uuid4` - TODO
### Advanced
`revert()` - execute the compensations and stop the workflow.
`continue_using(workflow, *args, **kwargs)` - stop the current workflow and pass the context to a new one.
`compact()` - a key can be defined on all the primitives to avoid the infinitely growing log issue. This function will keep only the last occurrence of each of the keys.
Raw data
{
"_id": null,
"home_page": null,
"name": "substantial",
"maintainer": null,
"docs_url": null,
"requires_python": "<4.0,>=3.11",
"maintainer_email": null,
"keywords": "durable, execution, redis, s3, filesystem",
"author": null,
"author_email": null,
"download_url": "https://files.pythonhosted.org/packages/0d/1c/fc6bbdd049a0a1add6295f339f171743c745c2b5c614dbb937dce1d91e6d/substantial-0.1.0.tar.gz",
"platform": null,
"description": "# Substantial\n\n[](https://badge.fury.io/py/substantial)\n\nBrokerless durable execution for Python.\n\n> Substantial is part of the\n> [Metatype ecosystem](https://github.com/metatypedev/metatype). Consider\n> checking out how this component integrates with the whole ecosystem and browse\n> the\n> [documentation](https://metatype.dev?utm_source=github&utm_medium=readme&utm_campaign=substantial)\n> to see more examples.\n\n## What is durable execution?\n\nDurable execution is a programming model where the state of a function is preserved across failures, restarts, or other (un)voluntary disruptions. This ensures that applications can continue execution from their last stable state without losing context or causing additional side effects. It is particularly well-suited for long-running workflows as it enables the management of complex sequences of steps, handling breaks, retries and recovery gracefully.\n\nSubstantial is designed around a replay mechanism that reconstructs the function state by re-executing historical events from stored logs. All the logic is embedded in a protocol and there is no centralized broker, allowing it to work with any backend (local files, cloud storage and databases). It aims to be an alternative for use cases that do not require the scale and complexity of [Temporal](https://github.com/temporalio/temporal) or [Durable Task](https://github.com/Azure/durabletask).\n\n## Getting started\n\n```\n# pypi\npip install substantial\npoetry add substantial\n\n# remote master\npip install --upgrade git+https://github.com/zifeo/substantial.git\npoetry add git+https://github.com/zifeo/substantial.git\n\n# local repo/dev\npoetry install\npre-commit install\nprotoc -I . --python_betterproto_out=. protocol/*\n```\n\n### Workflow\n\n```py\nfrom dataclasses import dataclass\nfrom datetime import timedelta\nimport random\nfrom substantial import workflow, Context\n\n\ndef roulette():\n if random.random() < 1 / 6:\n raise Exception(\"shot!\")\n return 100\n\n\ndef multiply(a, b):\n return a * b\n\n\n@dataclass\nclass State:\n cancelled: bool = False\n\n def update(self):\n self.cancelled = True\n\n\n@workflow()\nasync def example(c: Context):\n res = await c.save(roulette)\n\n await c.sleep(timedelta(seconds=10))\n res = await c.save(lambda: multiply(res, 0.5))\n\n n = await c.receive(\"by\")\n res = await c.save(lambda: multiply(res, n))\n\n s = State(cancelled=False)\n c.handle(\"cancel\", lambda _: s.update())\n\n await c.ensure(lambda: s.cancelled)\n return res\n```\n\n### Worker\n\nBrokerless means that there is no active broker and all the scheduling is solely organized around files/key values. Substantial currently supports Redis, local files and s3-compatible object storages.\n\n```py\nimport asyncio\nfrom substantial import Conductor\nfrom substantial.backends.fs import FSBackend\nfrom demo.readme_ws import example\n\n\nasync def main():\n backend = FSBackend(\"./demo/logs/readme\")\n substantial = Conductor(backend)\n substantial.register(example)\n\n # run the agent in the background\n agent = substantial.run()\n\n # start the workflow\n w = await substantial.start(example)\n\n await asyncio.sleep(3)\n print(await w.send(\"by\", 2))\n\n await asyncio.sleep(5)\n print(await w.send(\"cancel\"))\n\n output = await w.result()\n print(\"Final output\", output) # 100\n\n # stop the agent\n agent.cancel()\n await agent\n\n\nif __name__ == \"__main__\":\n asyncio.run(main())\n```\n\n## API\n\n_Note: not all features are implemented/completed yet._\n\n### Primitives\n\n`save(f: Callable, compensate_with: Optional[Callable]): Any` - memoize the result of a function to avoid re-execution on replay. Functions shall be idempotent as they may be called more than once in case of failure just before the value is persisted. The function can be compensated by providing its inverse effect and trigger later in the workflow with `revert`.\n\n`handle(event_name: str, cb: Callable): None` - register a callback to be executed when a specific event is received. The callbacks are executed in the order they were received and whenever a primitive being called.\n\n`ensure(f: Callable): True` - wait for the function to evaluate to true and schedule a new run when false.\n\n### Higher-level\n\n`sleep` - schedule a replay after a certain amount of time.\n\n`receive` - wait for the value of an event to be received.\n\n`log` - TODO\n\n`datetime.now` - TODO\n\n`random` - TODO\n\n`uuid4` - TODO\n\n### Advanced\n\n`revert()` - execute the compensations and stop the workflow.\n\n`continue_using(workflow, *args, **kwargs)` - stop the current workflow and pass the context to a new one.\n\n`compact()` - a key can be defined on all the primitives to avoid the infinitely growing log issue. This function will keep only the last occurrence of each of the keys.\n",
"bugtrack_url": null,
"license": "MPL-2.0",
"summary": "Brokerless durable execution for Python.",
"version": "0.1.0",
"project_urls": null,
"split_keywords": [
"durable",
" execution",
" redis",
" s3",
" filesystem"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "2a1d2584eed6442421f2ef84ba1eacb26fe2d28cb0d5db37450a55098090d464",
"md5": "fbe2f551ea9b74a05e32641a43f10de8",
"sha256": "91d0264f4a3b7c2b56ae3738c4cdf163a987877add8712cec90101719307233d"
},
"downloads": -1,
"filename": "substantial-0.1.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "fbe2f551ea9b74a05e32641a43f10de8",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<4.0,>=3.11",
"size": 26800,
"upload_time": "2025-01-05T14:14:25",
"upload_time_iso_8601": "2025-01-05T14:14:25.612005Z",
"url": "https://files.pythonhosted.org/packages/2a/1d/2584eed6442421f2ef84ba1eacb26fe2d28cb0d5db37450a55098090d464/substantial-0.1.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "0d1cfc6bbdd049a0a1add6295f339f171743c745c2b5c614dbb937dce1d91e6d",
"md5": "ab20c3a2643524d86ca3653fd58d5704",
"sha256": "9db614bcb6dde29f07b1e45e8a443f0c5f3b0d8ead8a866b61cddba21b1631a5"
},
"downloads": -1,
"filename": "substantial-0.1.0.tar.gz",
"has_sig": false,
"md5_digest": "ab20c3a2643524d86ca3653fd58d5704",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<4.0,>=3.11",
"size": 22362,
"upload_time": "2025-01-05T14:14:26",
"upload_time_iso_8601": "2025-01-05T14:14:26.729522Z",
"url": "https://files.pythonhosted.org/packages/0d/1c/fc6bbdd049a0a1add6295f339f171743c745c2b5c614dbb937dce1d91e6d/substantial-0.1.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-01-05 14:14:26",
"github": false,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"lcname": "substantial"
}
JSON