<gh_stars>1-10
# <EMAIL> - 10/18/21
# aws lambda-based, event-driven flow trigger
# using example event: s3:ObjectCreated:*

import logging
import os
from typing import Any, Dict

import boto3

import prefect
from prefect.run_configs.kubernetes import KubernetesRun

logging.basicConfig()
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# Get secret(s) from AWS Systems Manager Parameter Store
ssm = boto3.client("ssm")
PREFECT_AUTH_TOKEN = ssm.get_parameter(Name="keyName")
prefect_client = prefect.Client(api_key=PREFECT_AUTH_TOKEN)


# kick-off flow run with prefect.Client instance
def trigger_flow_run(params: Dict[str, Any]) -> dict:
    return prefect_client.create_flow_run(
        parameters=params,
        version_group_id=os.getenv("PREFECT_VERSION_GROUP_ID"),
        run_config=KubernetesRun(),
    )


# parse new S3 object as needed for flow params
def parse(s3_object: object) -> dict:
    # TODO
    return dict()


# parse event JSON (e.g. S3 ObjectCreated -> flow params)
def run(event, context):
    s3 = boto3.resource("s3")
    s3_bucket = event["Records"][0]["s3"]["bucket"]["name"]
    s3_key = event["Records"][0]["s3"]["object"]["key"]
    s3_object = s3.Object(s3_bucket, s3_key)

    params = parse(s3_object)

    try:
        trigger_flow_run(params)
        return {"success": True}

    except prefect.Client.ClientError:
        raise

# <EMAIL> - 10/18/21
# aws lambda-based, event-driven flow trigger
# using example event: s3:ObjectCreated:*

import logging
import os
from typing import Any, Dict

import boto3

import prefect
from prefect.run_configs.kubernetes import KubernetesRun

logging.basicConfig()
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# Get secret(s) from AWS Systems Manager Parameter Store
ssm = boto3.client("ssm")
PREFECT_AUTH_TOKEN = ssm.get_parameter(Name="keyName")
prefect_client = prefect.Client(api_key=PREFECT_AUTH_TOKEN)


# kick-off flow run with prefect.Client instance
def trigger_flow_run(params: Dict[str, Any]) -> dict:
    return prefect_client.create_flow_run(
        parameters=params,
        version_group_id=os.getenv("PREFECT_VERSION_GROUP_ID"),
        run_config=KubernetesRun(),
    )


# parse new S3 object as needed for flow params
def parse(s3_object: object) -> dict:
    # TODO
    return dict()


# parse event JSON (e.g. S3 ObjectCreated -> flow params)
def run(event, context):
    s3 = boto3.resource("s3")
    s3_bucket = event["Records"][0]["s3"]["bucket"]["name"]
    s3_key = event["Records"][0]["s3"]["object"]["key"]
    s3_object = s3.Object(s3_bucket, s3_key)

    params = parse(s3_object)

    try:
        trigger_flow_run(params)
        return {"success": True}

    except prefect.Client.ClientError:
        raise
# <EMAIL> - 10/18/21 # aws lambda-based, event-driven flow trigger # using example event: s3:ObjectCreated:* # Get secret(s) from AWS Systems Manager Parameter Store # kick-off flow run with prefect.Client instance # parse new S3 object as needed for flow params # TODO # parse event JSON (e.g. S3 ObjectCreated -> flow params)