trigger_airbyte_connections.py

def get_connections_for_workspace(workspace_id: str) -> List[str]:
    response = requests.post(
        f'{API_ROOT}/connections/list',
        json={'workspaceId': workspace_id},
    )
    response.raise_for_status()
    return [
        connection['connectionId']
        for connection in response.json()['connections']
        if connection['status'] == 'active'  # So we can still disable connections in the UI
    ]


def trigger_connection_sync(connection_id: str) -> dict:
    response = requests.post(
        f'{API_ROOT}/connections/sync',
        json={'connectionId': connection_id},
    )
    response.raise_for_status()
    return response.json()