chatflock.sequencial_process package

Submodules

chatflock.sequencial_process.sequential_process module

class chatflock.sequencial_process.sequential_process.SequentialProcess(steps, initial_state, save_state)

Bases: Generic[T]

Parameters:

• steps (Sequence[Step[T]])

• initial_state (T)

• save_state (Callable[[T], None])

run()

Return type:

T

save_state(state)

Parameters:

state (T)

Return type:

None

class chatflock.sequencial_process.sequential_process.Step(name, func, on_step_start=None, on_step_completed=None, on_step_failed=None)

Bases: Generic[T]

Parameters:

• name (str)

• func (Callable[[T, Any], Generator[T, None, None] | None])

• on_step_start (Callable[[T], None] | None)

• on_step_completed (Callable[[T], None] | None)

• on_step_failed (Callable[[T], None] | None)

run(state, **kwargs)

Parameters:

• state (T)

• kwargs (Any)

Return type:

Generator[T, None, None]

Module contents

class chatflock.sequencial_process.SequentialProcess(steps, initial_state, save_state)

Bases: Generic[T]

Parameters:

• steps (Sequence[Step[T]])

• initial_state (T)

• save_state (Callable[[T], None])

run()

Return type:

T

save_state(state)

Parameters:

state (T)

Return type:

None

class chatflock.sequencial_process.Step(name, func, on_step_start=None, on_step_completed=None, on_step_failed=None)

Bases: Generic[T]

Parameters:

• name (str)

• func (Callable[[T, Any], Generator[T, None, None] | None])

• on_step_start (Callable[[T], None] | None)

• on_step_completed (Callable[[T], None] | None)

• on_step_failed (Callable[[T], None] | None)

run(state, **kwargs)

Parameters:

• state (T)

• kwargs (Any)

Return type:

Generator[T, None, None]