Hooks: Flow Control and Automation
Hooks are a powerful SDK feature that enable full workflow customization, including side-effects, intercepts, and redirects.
Overview
Hooks are a core SDK annotation that enable your flows to take arbitrarily complex behaviors. You can think of them as the mortar that binds all of your connected systems into a comprehensive workflow.
For example:
- Special routing and auto-approve or -deny logic
- Ensuring that only certain actors can approve or deny certain types of requests
- Reflecting requests in external systems
Implementation
Hooks are all defined inside of a Sym Approval Flow's impl.py
file, fire at specific points in the workflow, and can deliver a broad range of user-defined behaviors.
Hooks prefixed with on_
will fire after a relevant event has been triggered, but before it is processed. For example, an on_request
Hook will trigger after a request form is submitted; but before the get_approvers
Reducer is processed and a channel message is sent.
Hooks prefixed with after_
will fire after a relevant event has finished processing.
Recursion is tightly gated
In theory, a Hook can return its own calling event (e.g.
return ApprovalTemplate.request()
inside ofon_request
). To prevent unintended timeouts due to user behavior, the Sym SDK will process the first recursive call only.In other words, any recursive returns in Hooks should be written with the expectation that they will be processed exactly once, after which either a different outcome should be processed, or the Flow will terminate.
Determining if an event
came from a Hook
event
came from a HookYou may want to split your logic based on the source of an event
. For example, if you want to gate approvals to a specific list of people, but always allow SDK automations to pass through. In such cases, you'll want to inspect event.channel.type
, as in the following example.
@hook
def on_approve(event):
"""
Only let members of the approver safelist approve requests
"""
# Don't check approval permissions if the event came from another hook
if event.channel.type == "sdk":
return
if not has_approve_access(event):
return ApprovalTemplate.ignore(
message="You are not authorized to approve this request."
)
For more information about event
attributes, please see Working With Flow Fields and Data.
List of Hooks
Hook | Fires when | Usage example |
---|---|---|
on_prompt | The access request form is about to be shown | Restricting access to certain flows |
after_prompt | The access request form has just been shown | Tracking usage in external systems |
on_request | The access request form is submitted | Auto-approving requests which do not need outside approval |
after_request | The access request message has just been sent | Tracking requests in external systems |
on_approve | The "approve" button on an access request message is clicked | Restricting who may approve access to specific resources |
after_approve | The access request message has just been updated to show an approval | Tracking approvals in external systems |
on_deny | The "deny" button on an access request message is clicked | Restricting who many deny access to specific resources |
after_deny | The access request message has just been updated to show a denial | Tracking denials in external systems |
on_escalate | A user is about to be given escalated permissions | Tracking access grants in external systems |
after_escalate | A user has just been given escalated permissions | Tracking access grants in external systems |
on_deescalate | A user is about to have their escalated permissions removed | Tracking access grant expirations in external systems |
after_deescalate | A user has just had their escalated permissions removed | Tracking access grant expirations in external systems |
Examples
on_prompt
on_prompt
@hook
def on_prompt(event):
if event.flow.srn.slug == "super_secret_flow" and event.user.email != "[email protected]":
return ApprovalTemplate.ignore(message="You're not allowed to access the super secret Flow!")
after_prompt
after_prompt
@hook
def after_prompt(event):
# Send a dad joke to everyone who wants to make an access request
response = requests.get("icanhazdadjoke.com")
print(response["joke"])
on_request
on_request
@hook
def on_request(event):
# Auto-approve urgent requests for access by the person on call
if event.payload.fields["urgency"] == "Urgent" and pagerduty.is_on_call(event.user):
return ApprovalTemplate.approve()
after_request
after_request
@hook
def after_request(event):
# Send a text to everyone on call when a request is made
for user in pagerduty.users_on_call(escalation_policy_name="123"):
requests.post(
url="http://my-website.com/sms",
body={"message": f"{event.user.email} just made an access request!"}
)
on_approve
on_approve
@hook
def on_approve(event):
if not okta.is_user_in_group(event.user, group_id=event.flow.vars["okta_managers_group"]):
return ApprovalTemplate.ignore(message="Only managers may approve access requests.")
after_approve
after_approve
@hook
def after_approve(event):
requester = event.get_actor("request")
# Turn the requester's lamp green when they're approved
requests.post(
url="http://lamps.com",
body={
"color": "green",
"lamp_owner": requester.email
}
)
on_deny
on_deny
@hook
def on_deny(event):
if okta.is_user_in_group(event.user, group_id=event.flow.vars["okta_interns_group"]):
return ApprovalTemplate.ignore(message="Interns cannot deny access requests.")
after_deny
after_deny
@hook
def after_deny(event):
requester = event.get_actor("request")
# Turn the requester's lamp red when they're denied
requests.post(
url="http://lamps.com",
body={
"color": "red",
"lamp_owner": requester.email
}
)
on_escalate
on_escalate
@hook
def on_escalate(event):
requester = event.get_actor("request")
# Comment on the ticket tracking this access.
requests.post(
url=f"http://jira.com/comment-ticket",
body={
"ticket_id": "123",
"message": f"{requester.email} has been granted access to {event.flow.srn.slug}."
}
)
after_escalate
after_escalate
@hook
def after_escalate(event):
requester = event.get_actor("request")
# Send a text to the requester when their access is granted
requests.post(
url="http://my-website.com/sms",
body={
"message": f"Your access to {event.flow.srn.slug} has been granted!",
"user_email": requester.email
}
)
on_deescalate
on_deescalate
@hook
def on_deescalate(event):
# Comment on the ticket tracking this access.
requests.post(
url=f"http://jira.com/comment-ticket",
body={
"ticket_id": "123",
"message": f"{event.user.email}'s {event.flow.srn.slug} access has ended."
}
)
after_deescalate
after_deescalate
@hook
def after_escalate(event):
# Send a text to the requester when their access has expired
requests.post(
url="http://my-website.com/sms",
body={
"message": f"Your access to {event.flow.srn.slug} has expired!",
"user_email": event.user.email
}
)
Updated 10 days ago