Temporal documentation decorator plugin for python
Project description
py-temporal-doc
Sample json structure
{
"callers": {
"Periodic": {
"data": ["Workflow1", "Workflow2"],
"docstring": "comment...",
"path": "project.directory.file.py"
},
"PostMethod": {
"workflows": ["Workflow3", "Workflow4"],
"docstring": "comment...",
"path": "project.directory.file.py"
}
},
"workflows": {
"Workflow1": {
"data": [
"ChildWorkflow1",
"Activity1",
"ChildWorkflow2",
"Activity2"
],
"docstring": "comment...",
"path": "project.directory.file.py"
},
"Workflow2": {
"sequence": [
"Activity2",
"Activity3"
],
"docstring": "comment...",
"path": "project.directory.file.py"
},
"Workflow3": {
"sequence": [
"Activity3",
"Activity4"
],
"docstring": "comment...",
"path": "project.directory.file.py"
},
"Workflow4": {
"sequence": [
"Activity4"
],
"docstring": "comment...",
"path": "project.directory.file.py"
},
"ChildWorkflow1": {
"sequence": [
"Activity1"
],
"docstring": "comment...",
"path": "project.directory.file.py"
},
"ChildWorkflow2": {
"sequence": [
"Activity2"
],
"docstring": "comment...",
"path": "project.directory.file.py"
}
},
"activities": {
"Activity1": {
"docstring": "comment...",
"path": "project.directory.file.py"
},
"Activity2": {
"docstring": "comment...",
"path": "project.directory.file.py"
},
"Activity3": {
"docstring": "comment...",
"path": "project.directory.file.py"
},
"Activity4": {
"docstring": "comment...",
"path": "project.directory.file.py"
}
}
}
Decorator Usage
@activity.defn
async def activity1() -> None:
pass
@activity.defn
async def activity2() -> None:
pass
@activity.defn
async def child_activity1() -> None:
pass
@activity.defn
async def child_activity2() -> None:
pass
@document.workflow_sequence(
child_activity1
)
@workflow.defn
class ChildWorkflow1:
@workflow.run
async def run(self) -> None:
pass
@document.workflow_sequence(
child_activity2
)
@workflow.defn
class ChildWorkflow2:
@workflow.run
async def run(self) -> None:
pass
@document.workflow_sequence( # Decorator to indicate the sequence of activities and child workflows inside a workflow
activity1,
ChildWorkflow1,
activity2,
ChildWorkflow2
)
@workflow.defn
class Workflow1:
@workflow.run
async def run(self) -> None:
pass
@document.workflow_sequence(
ChildWorkflow1,
ChildWorkflow2
)
@workflow.defn
class Workflow2:
@workflow.run
async def run(self) -> None:
pass
@document.workflow_caller( # Decorator to indicate the starting point on when the workflows are called
Workflow1,
)
def call_worklow1():
client = await temporal_client(namespace=TEMPORAL_NAMESPACE)
await temporal.execute_workflow(
Workflow1.run,
...
)
@document.workflow_caller(
Workflow2
)
def call_worklow2():
client = await temporal_client(namespace=TEMPORAL_NAMESPACE)
await temporal.execute_workflow(
Workflow2.run,
...
)
How to run
python main.py generate-json <path>
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
py_temporal_doc-0.0.2.tar.gz
(11.1 kB
view hashes)
Built Distribution
Close
Hashes for py_temporal_doc-0.0.2-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 0152432bc7c77c5a23999baf6be2292fb8465ab81487b3a16de2a9b5150a3047 |
|
MD5 | e9e847e08c71bddb8b36ad26b8229d92 |
|
BLAKE2b-256 | c8af503b194c242ed77a776f3e3297727d4a959169604cd933d03781b4d3b548 |