import boto3 import json from pprint import pprint from datetime import datetime, timedelta, timezone from dotenv import load_dotenv import os import uuid import snow load_dotenv() aws_session = boto3.Session( aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID"), aws_secret_access_key=os.environ.get("AWS_SECRET_ACCESS_KEY"), aws_session_token=os.environ.get("AWS_SESSION_TOKEN"), ) aws_dev_session = boto3.Session( aws_access_key_id=os.environ.get("DEV_AWS_ACCESS_KEY_ID"), aws_secret_access_key=os.environ.get("DEV_AWS_SECRET_ACCESS_KEY"), aws_session_token=os.environ.get("DEV_AWS_SESSION_TOKEN"), ) swf_client = aws_session.client("swf") s3_client = aws_dev_session.client("s3") domain: str = "prod_swf_feed_ingestion" def get_execution_events(execution_info, events): activities = {} for event in events: if event["eventType"] == "ActivityTaskScheduled": attributes = event["activityTaskScheduledEventAttributes"] activities[event["eventId"]] = { "activity_type": attributes["activityType"]["name"], "input": attributes.get("input"), "reason": None, "result": None, "start_time": None, "end_time": None, "run_id": execution_info["execution"].get("runId"), "scheduled_event_id": event["eventId"], "workflow_id": execution_info["execution"].get("workflowId"), "workflow_type": execution_info["workflowType"].get("name"), "execution_status": execution_info.get("executionStatus"), "close_status": execution_info.get("closeStatus"), "task_status": "SCHEDULED", } elif event["eventType"] == "ActivityTaskStarted": attributes = event["activityTaskStartedEventAttributes"] scheduled_event_id = attributes["scheduledEventId"] activity = activities.get(scheduled_event_id) if activity is not None: timestamp = event["eventTimestamp"].astimezone(timezone.utc).isoformat() activity["start_time"] = timestamp activity["task_status"] = "STARTED" elif event["eventType"] == "ActivityTaskFailed": attributes = event["activityTaskFailedEventAttributes"] scheduled_event_id = attributes["scheduledEventId"] activity = activities.get(scheduled_event_id) if activity is not None: timestamp = event["eventTimestamp"].astimezone(timezone.utc).isoformat() activity["end_time"] = timestamp activity["reason"] = attributes.get("reason") activity["result"] = attributes.get("result") activity["task_status"] = "FAILED" elif event["eventType"] == "ActivityTaskTimedOut": attributes = event["activityTaskTimedOutEventAttributes"] scheduled_event_id = attributes["scheduledEventId"] activity = activities.get(scheduled_event_id) if activity is not None: timestamp = event["eventTimestamp"].astimezone(timezone.utc).isoformat() activity["end_time"] = timestamp activity["reason"] = attributes.get("reason") activity["result"] = attributes.get("result") activity["task_status"] = "TIMED_OUT" elif event["eventType"] == "ActivityTaskCanceled": attributes = event["activityTaskCanceledEventAttributes"] scheduled_event_id = attributes["scheduledEventId"] activity = activities.get(scheduled_event_id) if activity is not None: timestamp = event["eventTimestamp"].astimezone(timezone.utc).isoformat() activity["end_time"] = timestamp activity["reason"] = attributes.get("reason") activity["result"] = attributes.get("result") activity["task_status"] = "CANCELED" elif event["eventType"] == "ActivityTaskCompleted": attributes = event["activityTaskCompletedEventAttributes"] scheduled_event_id = attributes["scheduledEventId"] activity = activities.get(scheduled_event_id) if activity is not None: timestamp = event["eventTimestamp"].astimezone(timezone.utc).isoformat() activity["end_time"] = timestamp activity["reason"] = attributes.get("reason") activity["result"] = attributes.get("result") activity["task_status"] = "COMPLETED" activity_list = [] for a in activities.values(): if a["start_time"] is None: continue if a.get("result") is not None: a["result"] = json.loads(a["result"]) if a.get("input") is not None: a["input"] = json.loads(a["input"]) activity_list.append(a) return activity_list def list_execution_events( execution_status: str, start_time: datetime, end_time: datetime ): paginator = None response_iterator = None if execution_status == "closed": paginator = swf_client.get_paginator("list_closed_workflow_executions") response_iterator = paginator.paginate( domain=domain, closeTimeFilter={ "oldestDate": start_time, "latestDate": end_time, }, ) elif execution_status == "open": paginator = swf_client.get_paginator("list_open_workflow_executions") response_iterator = paginator.paginate( domain=domain, startTimeFilter={ "oldestDate": start_time, "latestDate": end_time, }, ) execution_info_list = [] for response in response_iterator: for execution_info in response["executionInfos"]: # execution_info_list.append(execution_info) workflow_type = execution_info["workflowType"]["name"] if workflow_type.startswith("apple_music"): execution_info_list.append(execution_info) if workflow_type.startswith("spotify_feed"): execution_info_list.append(execution_info) # if workflow_type.startswith("spotify_charts"): # execution_info_list.append(execution_info) events = [] for execution_info in execution_info_list: workflow_id = execution_info["execution"]["workflowId"] print(workflow_id) run_id = execution_info["execution"]["runId"] paginator = swf_client.get_paginator("get_workflow_execution_history") response_iterator = paginator.paginate( domain=domain, execution={"workflowId": workflow_id, "runId": run_id} ) response_events = [] for response in response_iterator: response_events += response["events"] events += get_execution_events(execution_info, response_events) return events def store_open(events): filename = f"{str(uuid.uuid4())}.json" body = json.dumps(events) s3_client.put_object( Body=body, Bucket="dev-cucumbers", Key=f"FeedIngestionStatusAudit/swf/{filename}", ContentType="application/json", ) copy_into_query = f""" COPY INTO dev_engineering.sduberg.temp_staging_raw_audit_swf_open FROM 's3://dev-cucumbers/FeedIngestionStatusAudit/swf/{filename}' FILE_FORMAT = ( TYPE='JSON' DATE_FORMAT='YYYY-MM-DD' TIMESTAMP_FORMAT=AUTO COMPRESSION=AUTO STRIP_OUTER_ARRAY=TRUE ) FORCE=TRUE MATCH_BY_COLUMN_NAME = CASE_INSENSITIVE CREDENTIALS=( AWS_KEY_ID='{os.environ.get("DEV_AWS_ACCESS_KEY_ID")}' AWS_SECRET_KEY='{os.environ.get("DEV_AWS_SECRET_ACCESS_KEY")}' AWS_TOKEN='{os.environ.get("DEV_AWS_SESSION_TOKEN")}' ); """ merge_into_query = """ MERGE INTO dev_engineering.sduberg.staging_raw_audit_swf as target USING dev_engineering.sduberg.temp_staging_raw_audit_swf_open as source ON target.run_id = source.run_id AND target.scheduled_event_id = source.scheduled_event_id WHEN MATCHED THEN UPDATE SET target.activity_type = source.activity_type, target.input = source.input, target.reason = source.reason, target.result = source.result, target.start_time = source.start_time, target.end_time = source.end_time, target.run_id = source.run_id, target.scheduled_event_id = source.scheduled_event_id, target.workflow_id = source.workflow_id, target.workflow_type = source.workflow_type, target.execution_status = source.execution_status, target.close_status = source.close_status, target.task_status = source.task_status WHEN NOT MATCHED THEN INSERT ( activity_type, input, reason, result, start_time, end_time, run_id, scheduled_event_id, workflow_id, workflow_type, execution_status, close_status, task_status ) VALUES ( source.activity_type, source.input, source.reason, source.result, source.start_time, source.end_time, source.run_id, source.scheduled_event_id, source.workflow_id, source.workflow_type, source.execution_status, source.close_status, source.task_status ) """ truncate_temp_query = """ TRUNCATE TABLE dev_engineering.sduberg.temp_staging_raw_audit_swf_open """ snow.conn.cursor().execute(truncate_temp_query) snow.conn.cursor().execute(copy_into_query) snow.conn.cursor().execute(merge_into_query) snow.conn.cursor().execute(truncate_temp_query) def store_closed(events): filename = f"{str(uuid.uuid4())}.json" body = json.dumps(events) s3_client.put_object( Body=body, Bucket="dev-cucumbers", Key=f"FeedIngestionStatusAudit/swf/{filename}", ContentType="application/json", ) copy_into_query = f""" COPY INTO dev_engineering.sduberg.temp_staging_raw_audit_swf_closed FROM 's3://dev-cucumbers/FeedIngestionStatusAudit/swf/{filename}' FILE_FORMAT = ( TYPE='JSON' DATE_FORMAT='YYYY-MM-DD' TIMESTAMP_FORMAT=AUTO COMPRESSION=AUTO STRIP_OUTER_ARRAY=TRUE ) FORCE=TRUE MATCH_BY_COLUMN_NAME = CASE_INSENSITIVE CREDENTIALS=( AWS_KEY_ID='{os.environ.get("DEV_AWS_ACCESS_KEY_ID")}' AWS_SECRET_KEY='{os.environ.get("DEV_AWS_SECRET_ACCESS_KEY")}' AWS_TOKEN='{os.environ.get("DEV_AWS_SESSION_TOKEN")}' ); """ merge_into_query = """ MERGE INTO dev_engineering.sduberg.staging_raw_audit_swf as target USING dev_engineering.sduberg.temp_staging_raw_audit_swf_closed as source ON target.run_id = source.run_id AND target.scheduled_event_id = source.scheduled_event_id WHEN MATCHED THEN UPDATE SET target.activity_type = source.activity_type, target.input = source.input, target.reason = source.reason, target.result = source.result, target.start_time = source.start_time, target.end_time = source.end_time, target.run_id = source.run_id, target.scheduled_event_id = source.scheduled_event_id, target.workflow_id = source.workflow_id, target.workflow_type = source.workflow_type, target.execution_status = source.execution_status, target.close_status = source.close_status, target.task_status = source.task_status WHEN NOT MATCHED THEN INSERT ( activity_type, input, reason, result, start_time, end_time, run_id, scheduled_event_id, workflow_id, workflow_type, execution_status, close_status, task_status ) VALUES ( source.activity_type, source.input, source.reason, source.result, source.start_time, source.end_time, source.run_id, source.scheduled_event_id, source.workflow_id, source.workflow_type, source.execution_status, source.close_status, source.task_status ) """ truncate_temp_query = """ TRUNCATE TABLE dev_engineering.sduberg.temp_staging_raw_audit_swf_closed """ snow.conn.cursor().execute(truncate_temp_query) snow.conn.cursor().execute(copy_into_query) snow.conn.cursor().execute(merge_into_query) snow.conn.cursor().execute(truncate_temp_query) def get_last_end_time(): query = "select max(end_time) from dev_engineering.sduberg.staging_raw_audit_swf" result = snow.conn.cursor().execute(query).fetchone() timestamp = result[0] if timestamp is None: return datetime.now(timezone.utc) - timedelta(hours=1) return timestamp if __name__ == "__main__": bucket = "dev-cucumbers" end_time = datetime.now(timezone.utc) start_time = get_last_end_time() # start_time = datetime.fromisoformat('2025-07-16T00:00:00+00:00') # end_time = datetime.fromisoformat('2025-06-18T00:00:00+00:00') pprint({"start_time": start_time.isoformat(), "end_time": end_time.isoformat()}) open_events = list_execution_events( execution_status="open", start_time=start_time, end_time=end_time, ) closed_events = list_execution_events( execution_status="closed", start_time=start_time, end_time=end_time, ) store_open(open_events) store_closed(closed_events)