Custom Integrations

Connect to APIs that the Sym SDK does not natively support.

Suggest Edits

Overview

Declare a custom integration to connect to web APIs from the Sym SDK and to define custom user mappings with Sym's User Management system.

VictorOps Example

We'll take the use case of integrating with Splunk On-Call (VictorOps), a service that Sym does not support natively, to demonstrate how custom integrations work. VictorOps provides an API to tell you what users are currently on call. In our example, we'll use VictorOps on-call status to determine whether a request should be fast tracked or sent for human-in-the-loop approval.

📘

Full Example

The full code for this tutorial is in our examples repo.

Declaring a Custom Integration

Define a Secret

First you need to set up a sym_secret to store the VictorOps API key. Read our secrets management guide for general info on setting up secrets with Sym.

# An AWS Secrets Manager Secret to hold your VictorOps API Key. Set the value with:
# aws secretsmanager put-secret-value --secret-id "main/victorops-api-key" --secret-string "YOUR-VICTOROPS-API-KEY"
resource "aws_secretsmanager_secret" "victorops_api_key" {
  name        = "main/victorops-api-key"
  description = "API Key for Sym to call VictorOps APIs"

  # This SymEnv tag is required and MUST match the SymEnv tag in the 
  # aws_iam_policy.secrets_manager_access in your `secrets.tf` file
  tags = {
    SymEnv = local.environment_name
  }
}

# This resources tells Sym how to access your VictorOps API Key.
resource "sym_secret" "victorops_api_key" {
  # The source of your secrets and the permissions needed to access
  # i.e. AWS Secrets Manager, access with IAM Role.
  source_id = sym_secrets.this.id

  # name of the key in AWS Secrets Manager
  path = aws_secretsmanager_secret.victorops_api_key.name
}

Custom integrations are declared by setting the sym_integration type to custom. Use an external_id value that uniquely identifies this integration. In this example, we declare an integration with VictorOps that is identified by API ID:

# A custom integration for VictorOps. Custom integrations let us use secrets in
# the SDK and configure user mappings when necessary
resource "sym_integration" "victorops" {
  type        = "custom"
  name        = "main-victorops-integration"
  external_id = "aaabbcccddee" # VictorOps API ID

  settings = {
    # `type=custom` sym_integrations use a secret_ids_json property
    secret_ids_json = jsonencode([sym_secret.victorops_api_key.id])
  }
}

Using your custom integration in the SDK

Once you've declared a custom integration, you need to add it to your Flow's sym_environment. Then you can refer to it from within your Flow implementation code.

# The sym_environment is a container for sym_flows that share configuration values
# (e.g. shared integrations or error logging)
resource "sym_environment" "this" {
  name            = "main"
  error_logger_id = sym_error_logger.slack.id

  integrations = {
    slack_id = sym_integration.slack.id

    # This lets us access the custom API key from within the SDK
    victorops_id = sym_integration.victorops.id
  }
}

Now that the custom integration is in your environment, you can access it in your implementation as a property of your Flow. We use the API Key and the requests library to hit the VictorOps API:

# Get the custom integration we set up for VictorOps
integration = event.flow.environment.integrations["victorops"]
# Get the api key we passed in to the integration with the secret_ids_json setting
token = integration.settings["secrets"][0].retrieve_value()

headers = {
    "Accept": "application/json",
    # We used the API ID as our custom integration's external ID
    "X-VO-Api-Id": integration.external_id,
    "X-VO-Api-Key": token,
}
response = requests.get(
    f"https://api.victorops.com/api-public/{api_path}",
    headers=headers,
    params=params,
)
body = response.json()

if not response.ok:
    message = body.get("message", "")
    raise RuntimeError(f"API failed with message: {message}")

We'll use the response payload to determine if the requesting user is on call:

def is_requester_on_call(event):
    """Check if the requesting user is currently on call in VictorOps using our custom
    integration"""
    ...
    for team in body.get("teamsOnCall", []):
        for oncall in team["oncallNow"]:
            for user in oncall["users"]:
                if user["onCalluser"]["username"] == username:
                    return True

And finally - we'll fast-track approval for on call users using an on_request hook:

# Hooks let you customize workflow behavior running code before or after each
# step in your workflow.
@hook
def on_request(event):
    """If the requester is on-call, auto-approve their requests"""
    if is_requester_on_call(event):
        original_reason = event.payload.fields["reason"]
        return ApprovalTemplate.approve(
            reason=f"Auto-approved On-call engineer: {original_reason}️"
        )

Defining Custom User Mappings

Once you've declared an integration of type custom, you can define custom user mappings for that integration. You can manage user mappings manually using symflow, or you can manage the user mappings programmatically in the SDK.

Manage user mappings in the SDK

To map your users programmatically in the SDK, you use the persist_user_identity method. Here we check if a custom user identity already exists for our requesting user. If it does not, we use the VictorOps API to find a matching user. If that user exists, we save it:

def get_custom_user(user, integration):
    """
    Get the VictorOps username for the given Sym user. Each Sym user may have a separate identity
    stored for each integrated service.

    If the user already has a VictorOps identity, then return it.
    Otherwise, fetch the identity from VictorOps and persist it.

    Note: You can also use the symflow CLI to custom map user identities if necessary.
    """
    identity = user.identity("custom", integration.external_id)
    if identity:
        return identity.user_id
    
    # Use the VictorOps API to find a matching user
    user_id = find_user_by_email(user.email, integration)
    if user_id:
        # Use the Sym SDK to store the user identity so we don't need to look it up again.
        persist_user_identity(
            email=user.email,
            service="custom",
            service_id=integration.external_id,
            user_id=user_id,
        )

    return user_id

Updated 10 days ago