Reducers: Routing and Identity
Reducers are an SDK feature that "reduce" event input to a single value for use in a Flow.
Overview
The Sym SDK uses special Reducer functions to manage request routing, as well as special cases for user identity matching.
Reducers run at specific points in the Flow defined by the Sym backend, and will be triggered with every run of a given Flow.
Reducer | Required | Function |
---|---|---|
get_approvers | yes | Route requests. |
get_identity_lookup | no | Return a different email to be used by Sym when discovering a user's identity in a third party system. |
get_identity | no | Match and persist user identities in cases where a requester's Slack email address does not match a third party system. |
get_approvers
get_approvers
The only required reducer. Accepts a sym.sdk.event.Event
representing a Sym Request Event, and returns a RequestDestination or RequestDestinationFallback object that defines where that request should go.
This reducer is the core of Sym's routing logic. For more detail, see How Do I Choose Where My Requests Go.
from sym.sdk.annotations import reducer
from sym.sdk.integrations import pagerduty, okta, slack
@reducer
def get_approvers(event):
"""
This reducer returns an object representing where the Sym Request should be sent.
Typically, this method will return utilize one of the Slack SDK methods, such as
`slack.channel` or `slack.fallback` to generate a RequestDestination object.
Args:
event: A sym.sdk.event.Event object containing information about the Request Event
that is currently being handled.
Returns:
A sym.sdk.request_destination.RequestDestination or
sym.sdk.request_destination.RequestDestinationFallback object
indicating where the request should be sent for approval.
"""
if pagerduty.is_on_call(event.user, schedule_name="prod_on_call_schedule"):
# This is a self-approval in a DM
return slack.user(event.user, allow_self=True)
if event.payload.fields["urgency"] == "High":
# This is a self-approval in a channel
return slack.channel("#break-glass", allow_self=True)
on_call_mgrs = okta.users_in_group(group_id="00g12345689")
# This would create a group DM for on-call managers
return slack.group(on_call_mgrs)
get_identity_lookup
get_identity_lookup
This reducer is invoked before get_identity
and before standard identity auto-discovery performed by Sym. If your Sym user emails (i.e. Slack emails) do not match the third party service's emails, you can utilize this reducer to optionally transform your Sym user emails before performing identity discovery. If this reducer returns None
, then the Sym user's original email will be used for auto-discovery.
For example, if your Sym user emails are in the format [email protected]
, but your user emails in the third party service are [email protected]
, you can implement a get_identity_lookup
reducer as follows:
from sym.sdk.annotations import reducer
@reducer
def get_identity_lookup(event, service_type, external_id, user):
"""
The get_identity_lookup reducer runs before identity discovery, and can optionally return a different
email to use when looking up identities in external services.
Args:
event: A sym.sdk.event.Event object containing information about the Event
that is currently being handled.
service_type: A string indicating the service type that this Identity is for. (e.g. 'okta', 'aptible', 'aws_iam', etc.)
external_id: A unique string identifier for this service. This values matches the
external_id value set in the corresponding sym_integration in your Terraform
configuration.
user: The sym.sdk.user.User for whom Identity shall be Discovered
Returns:
An email to be used for Identity Discovery.
If None is returned, then the original Sym email will be used.
"""
# In this example, our Aptible user emails follow a different convention than our
# standard Sym user emails
if service_type == "aptible":
# The original user.email is `[email protected]`.
# Transform this email into `[email protected]`.
username_parts = user.email.split("@")[0].split(".")
first_initial = username_parts[0][0]
last_name = username_parts[1]
# Aptible users have emails with the format `[email protected]`.
# This email will be used by Sym to auto-discover the user's Aptible Identity.
return f"{first_initial}.{last_name}@test.symops.io"
# For all other services, perform normal auto-discovery with the Sym user's original email.
return None
get_identity
get_identity
Accepts a sym.sdk.event.Event
, a service type, the service's external ID, and a sym.sdk.user.User
object, and returns either a string or None
. If it returns a value, that value will be used as the user's identity in the third party service. If it returns None
, the normal identity auto-discovery is still performed.
Normally, Sym attempts to auto-discover users' identities in third-party services you integrate with based on the user's email address. If this isn't possible for whatever reason, this reducer can be used to construct the user's identity in that service instead.
from sym.sdk.annotations import reducer
from sym.sdk.integrations import okta
@reducer
def get_identity(event, service_type, external_id, user):
"""
For a given combination of service and user, returns an identifier for
that user in the service; or, return None to perform automated identity
discovery in the service.
Args:
event: A sym.sdk.event.Event object containing information about Event
that is currently being handled.
service_type: A string indicating the service type that this Identity is for.
external_id: A unique string identifier for this service. This values matches the
external_id value set in the corresponding sym_integration in your Terraform
configuration.
user: The sym.sdk.user.User object to whom the Identity belongs.
Returns:
The string value of the User's identity for the Service identified by the given
service_type and external_id. This value will vary based on the Service; for example,
AWS IAM Identities are the AWS IAM User's ARN, while Okta Identities are the unique
Okta User ID.
If None is returned, then standard identity discovery will be performed.
"""
if service_type == "aws_iam":
# For AWS IAM, construct the user's ARN based on the external ID (for
# AWS IAM, this is the AWS account ID) and the username portion of the
# user's email
return f"arn:aws:iam::{external_id}:user/{user.email.split('@')[0]}"
elif service_type == "okta":
# For Okta, we need to hard-code the IDs for some users because their
# emails in Slack don't match their Okta emails.
email_to_okta_uid = {
"[email protected]": "00u12345678",
"[email protected]": "00u9abcdefg",
"[email protected]": "00uhijklmno"
}
# For all other users, return None to indicate that we want to perform
# auto-discovery of the user's identity as normal
return email_to_okta_uid.get(user.email, None)
return None # For all other services, perform normal auto-discovery
Updated 9 days ago