- Inspired by the event-hook(callback) in javascript.
- Zero third-party dependencies, only requires
python>=3.10
- Run lifespan functions, without mental burden of nested
async/await
- Just define event hook functions with
async or not, then run flow with await or not
- Lifespan functions can access the
Context variable during the whole lifespan
- Execute the event hook functions sequentially, or customize the main-logic function
- Enjoy typing check/hint support
1. High Recommend 👍: Define async hook functions, run flow with await flow.run()
from hookio import Flow
# Define Hook type and Ctx type
class MyHook:
hook_one:Callable
hook_two:Callable
@dataclass()
class MyCtx:
state:str=''
state_list:list=field(default_factory=list)
# Define some hooks
async def hook_one(ctx:MyCtx):
print(f"start Hook One: ctx.state is {ctx.state}")
ctx.state = "Completed"
ctx.state_list.append({'hook1':ctx.state})
print(f"leave Hook One: {ctx.state_list}")
async def hook_two(ctx:MyCtx):
print(f"start Hook Two: ctx.state is {ctx.state}")
ctx.state = "error"
ctx.state_list.append({'hook2':ctx.state})
print(f"leave Hook Two: {ctx.state_list}")
# Define main_logic
async def main_function(hooks:MyHook,ctx:MyCtx):
print(f"Executing Main Function:{ctx.state}")
# access ctx in main logic
ctx.state='main_start'
await hooks.hook_one(ctx)
await hooks.hook_two(ctx)
print("Main Function Completed")
# Create a Flow instance
flow = Flow(
hooks=[hook_one, hook_two],
ctx=MyCtx(state='init', state_list=[])
)
# Run with await flow.run()
await flow.run()
print(f'flow finished and before cleanup: {flow.ctx}')
flow.cleanup()
print(f'after cleanup: {flow.ctx}')
# start Hook One: ctx.state is init
# leave Hook One: [{'hook1': 'Completed'}]
# start Hook Two: ctx.state is Completed
# leave Hook Two: [{'hook1': 'Completed'}, {'hook2': 'error'}]
# flow finished and before cleanup: MyCtx(state='error', state_list=[{'hook1': 'Completed'}, {'hook2': 'error'}])
# after cleanup: {}
2. Define hook functions, run flow with flow.safe_run()
from hookio import Flow
# Define Hook type and Ctx type
class MyHook:
hook_one:Callable
hook_two:Callable
@dataclass()
class MyCtx:
state:str=''
state_list:list=field(default_factory=list)
# Define some hooks
await def hook_one(ctx:MyCtx):
print(f"start Hook One: ctx.state is {ctx.state}")
ctx.state = "Completed"
ctx.state_list.append({'hook1':ctx.state})
print(f"leave Hook One: {ctx.state_list}")
def hook_two(ctx:MyCtx):
print(f"start Hook Two: ctx.state is {ctx.state}")
ctx.state = "error"
ctx.state_list.append({'hook2':ctx.state})
print(f"leave Hook Two: {ctx.state_list}")
# Create a Flow instance
flow = Flow(
hooks=[hook_one, hook_two],
ctx=MyCtx(state='init', state_list=[])
)
# Run with flow.safe_run()
flow.safe_run()
print(f'flow finished and before cleanup: {flow.ctx}')
flow.cleanup()
print(f'after cleanup: {flow.ctx}')
# start Hook One: ctx.state is init
# leave Hook One: [{'hook1': 'Completed'}]
# start Hook Two: ctx.state is Completed
# leave Hook Two: [{'hook1': 'Completed'}, {'hook2': 'error'}]
# flow finished and before cleanup: MyCtx(state='error', state_list=[{'hook1': 'Completed'}, {'hook2': 'error'}])
# after cleanup: {}
3. Advance Usage 😈: mixing async/sync hook functions, run your own custom main_flow_logic function
from hookio import Flow
# Define Hook type and Ctx type
class MyHook:
hook_one:Callable
hook_two:Callable
@dataclass()
class MyCtx:
state:str=''
state_list:list=field(default_factory=list)
# Define some hooks
async def hook_one(ctx:MyCtx):
print(f"start Hook One: ctx.state is {ctx.state}")
ctx.state = "Completed"
ctx.state_list.append({'hook1':ctx.state})
print(f"leave Hook One: {ctx.state_list}")
async def hook_two(ctx:MyCtx):
print(f"start Hook Two: ctx.state is {ctx.state}")
ctx.state = "error"
ctx.state_list.append({'hook2':ctx.state})
print(f"leave Hook Two: {ctx.state_list}")
# Define main_logic
async def main_function(hooks:MyHook,ctx:MyCtx):
print(f"Executing Main Function:{ctx.state}")
# access ctx in main logic
ctx.state='main_start'
await hooks.hook_one(ctx)
hooks.hook_two(ctx)
print("Main Function Completed")
# Create a Flow instance
flow = Flow(
hooks=[hook_one, hook_two],
logic=main_function,
ctx=MyCtx(state='init', state_list=[])
)
# Run with await flow.run()
await flow.run()
print(f'flow finished and before cleanup: {flow.ctx}')
flow.cleanup()
print(f'after cleanup: {flow.ctx}')
# Executing Main Function:init
# start Hook One: ctx.state is main_start
# leave Hook One: [{'hook1': 'Completed'}]
# start Hook Two: ctx.state is Completed
# leave Hook Two: [{'hook1': 'Completed'}, {'hook2': 'error'}]
# Main Function Completed
# flow finished and before cleanup: MyCtx(state='error', state_list=[{'hook1': 'Completed'}, {'hook2': 'error'}])
# after cleanup: {}
Typing check/hint for IDE 😊
class Hook:
hook_one:Callable
hook_two:Callable
...
@dataclass()
class MyCtx:
state:str=''
state_list:list=field(default_factory=list)
...