Using PyAirbyte as a data orchestrator for hosted Airbyte connections

# Add virtual environment support for running in Google Colab
!apt-get install -qq python3.10-venv

# Install PyAirbyte
%pip install --quiet airbyte

import requests
from airbyte import cloud
from datetime import datetime
from google.colab import userdata

# Initialize the Airbyte Cloud workspace
workspace = cloud.CloudWorkspace(
    workspace_id=userdata.get("AIRBYTE_WORKSPACE_ID"),
    api_key=userdata.get("AIRBYTE_API_KEY"),
)

# Slack webhook URL - replace with your actual Slack webhook URL
slack_webhook_url = "https://hooks.slack.com/services/XXX/YYY/ZZZ"

# Initialize job data for visualization
job_data = []

def send_slack_message(message, webhook_url):
    """Sends a message to a Slack channel specified by the webhook URL."""
    payload = {'text': message}
    response = requests.post(webhook_url, json=payload)
    return response.status_code

def trigger_sync_and_check(connection_id):
    """Triggers a sync job for a specified connection ID and checks the job status."""
    start_time = datetime.now()
    connection = workspace.get_connection(connection_id=connection_id)
    sync_result = connection.run_sync()
    end_time = datetime.now()
    duration = (end_time - start_time).total_seconds()
    status = str(sync_result.get_job_status())

    # Store job data for visualization
    job_data.append({
        'connection_id': connection_id,
        'start_time': start_time,
        'end_time': end_time,
        'duration': duration,
        'status': status
    })

    return status, sync_result, start_time, duration

def format_sync_details(sync_result, start_time, duration):
    """Format sync details for Slack message, including start time and duration."""
    details = (
        f"Job ID: {sync_result.job_id}\n"
        f"Start Time: {start_time.strftime('%Y-%m-%d %H:%M:%S')}\n"
        f"Duration: {duration:.2f} seconds\n"
        f"Records Synced: {sync_result.records_synced}\n"
        f"Bytes Synced: {sync_result.bytes_synced}\n"
        f"Job URL: {sync_result.job_url}"
    )
    return details

def run_syncs():
    """Handle sequential sync jobs where the second job depends on the success of the first."""
    # Connection IDs for the two jobs. Replace with your connection IDs
    first_connection_id = "df2898ce-b21f-43c0-8c82-bd8c6dd24931"
    second_connection_id = "86075252-9004-41fc-b5e3-f61f20388ff5"

    first_status, first_sync_result, first_start, first_duration = trigger_sync_and_check(first_connection_id)
    if "SUCCEEDED" in first_status:
        message = f"First sync succeeded (Connection ID: {first_connection_id}).\nDetails:\n{format_sync_details(first_sync_result, first_start, first_duration)}\nProceeding to second sync."
        send_slack_message(message, slack_webhook_url)
        print(message)

        second_status, second_sync_result, second_start, second_duration = trigger_sync_and_check(second_connection_id)
        if "SUCCEEDED" in second_status:
            message = f"Second sync succeeded (Connection ID: {second_connection_id}).\nDetails:\n{format_sync_details(second_sync_result, second_start, second_duration)}"
        else:
            message = f"Second sync failed (Connection ID: {second_connection_id}). Status: {second_status}\nDetails:\n{format_sync_details(second_sync_result, second_start, second_duration)}"
    else:
        message = f"First sync failed (Connection ID: {first_connection_id}). Status: {first_status}\nDetails:\n{format_sync_details(first_sync_result, first_start, first_duration)}\nSecond sync not initiated."

    send_slack_message(message, slack_webhook_url)
    print(message)

run_syncs()

import matplotlib.pyplot as plt
import matplotlib.patches as patches
from datetime import datetime, timedelta

def get_color(status):
    """Returns color based on the job status."""
    if "SUCCEEDED" in status:
        return 'green'  # Green for success
    elif "FAILED" in status:
        return 'yellow'  # Yellow for failure
    else:
        return 'gray'  # Gray for any other status

def draw_job_graph(job_data):
    """Draws a graph of job executions where rectangles represent the duration and color indicates status."""
    fig, ax = plt.subplots(figsize=(14, 3))

    # Start position and time tracking
    start_time = min(job['start_time'] for job in job_data)  # Find the earliest start time
    end_time = max(job['start_time'] + timedelta(seconds=job['duration']) for job in job_data)  # Find the latest end time
    total_duration = (end_time - start_time).total_seconds()

    for job in job_data:
        start_offset = (job['start_time'] - start_time).total_seconds()
        width = job['duration']  # Duration in seconds
        color = get_color(job['status'])  # Get color based on job status
        rect = patches.Rectangle((start_offset, 0), width, 1, linewidth=1, edgecolor='black', facecolor=color)
        ax.add_patch(rect)

        # Adding text inside the rectangle
        ax.text(start_offset + width / 2, 0.5, f"{job['connection_id']} ({job['duration']}s)",
                verticalalignment='center', horizontalalignment='center', color='black', fontweight='bold')

    # Set the limits and labels of the x-axis
    ax.set_xlim(0, total_duration)
    ax.set_ylim(-0.5, 1.5)
    ax.set_xlabel("Time (seconds from start)")
    ax.set_title('Job Execution Timeline')
    plt.xticks(range(0, int(total_duration) + 1, 20))  # Set x-ticks to appear every 20 seconds
    plt.yticks([])
    plt.grid(True, which='both', axis='x', linestyle='--', linewidth=0.5)
    plt.show()

draw_job_graph(job_data)