andaction 1.0.0

No project description provided

and-action

A lightweight Python workflow engine built around persistent state machines. Business logic lives in stage methods on a Process class; the framework handles persistence, scheduling, and state transitions.

Overview

Each process is a class with stage methods. Stages return outcomes that tell the engine what to do next: advance to another stage, sleep until a time, wait for an external event, or end. Process state is persisted to a database between stages, so processes survive restarts and can run across distributed workers.

Installation

pip install and-action

Quick Start

1. Define your process

from datetime import timedelta
from andaction import Process, WaitOn, Event, Timeout

class GreetingProcess(Process):

    def on_start(self):
        print(f"Hello, {self.pinfo.vars['name']}!")
        return self.sleep("on_followup", timedelta(seconds=5))

    def on_followup(self):
        print("Hope you are well.")
        return self.end()

2. Set up the app

import asyncio
from sqlalchemy.ext.asyncio import create_async_engine
from andaction import App
from andaction.db import ProcessRepo

engine = create_async_engine("sqlite+aiosqlite:///myapp.db")
repo = ProcessRepo(engine)

app = App(repo)
app.reg(GreetingProcess)

3. Start a process and run

async def main():
    await repo.initialize()
    await app.start("greeting", vars={"name": "Alice"})
    await app.run()  # starts the scheduler loop

asyncio.run(main())

4. Custom factory (when a process needs dependencies)

import httpx

http = httpx.AsyncClient()

@app.process_factory
def greeting(pinfo) -> GreetingProcess:
    return GreetingProcess(pinfo, http)

Outcomes

Outcome	Method	Behavior
NextStage	self.next("stage_name")	Advance to another stage immediately
Sleep	self.sleep("stage_name", timedelta)	Pause until a timer fires, then resume
WaitOn	return WaitOn(Event(...), Timeout(...))	Wait for an external event or timeout
End	self.end()	Terminate the process

Persistence

ProcessRepo is a SQLAlchemy-backed implementation of the ProcessStorage interface. Pass any custom implementation to App to use a different backend.

class MyStorage(ProcessStorage):
    async def create(self, pinfo): ...
    async def update(self, pinfo): ...
    async def acquire_ready_procs(self, limit=10): ...
    async def acquire_event_proc(self, process_id, event_type): ...
    async def get_by_correlation_id(self, correlation_id): ...