airbyte-cloud-prefect-to-orchestra

Overview

This skill converts Prefect prefect_airbyte tasks targeting Airbyte Cloud (api.airbyte.com) into Orchestra pipeline tasks using integration: AIRBYTE_CLOUD and integration_job: AIRBYTE_CLOUD_JOB. It covers sync and reset job types.

Parameter Mapping

Prefect construct	Orchestra field	Notes
AirbyteConnection.load("...")	connection:	Orchestra Airbyte Cloud connection slug
connection_id (UUID)	parameters.connection_id	Copy verbatim — UUID string
trigger_sync_run_and_wait_for_completion()	parameters.job_type: sync	Orchestra always waits; no separate polling needed
reset_cache=True	parameters.job_type: reset	Full reload of the connection
Prefect block host api.airbyte.com	integration: AIRBYTE_CLOUD	Any other host → use airbyte-server-prefect-to-orchestra

Orchestra YAML Structure

integration: AIRBYTE_CLOUD
integration_job: AIRBYTE_CLOUD_JOB
name: sync-task
connection: airbyte_cloud_prod_12345
parameters:
  connection_id: 'a1b2c3d4-e5f6-7890-abcd-ef1234567890'
  job_type: sync
depends_on: []
condition: null
tags: []

Conversion Steps

• Confirm the Prefect block's airbyte_server_host is api.airbyte.com — if not, use airbyte-server-prefect-to-orchestra
• Set integration: AIRBYTE_CLOUD
• Set integration_job: AIRBYTE_CLOUD_JOB
• Copy the connection_id UUID verbatim into parameters.connection_id
• Map reset_cache=True → parameters.job_type: reset; otherwise parameters.job_type: sync
• Set connection: to the Orchestra Airbyte Cloud connection slug (ask operator if unknown)
• Set depends_on: from upstream task dependencies

Before / After Example

Prefect (before)

from prefect import flow, task
from prefect_airbyte.connections import AirbyteConnection
from prefect_airbyte.flows import trigger_sync_run_and_wait_for_completion

@flow
def sync_airbyte_cloud():
    airbyte_conn = AirbyteConnection.load("my-airbyte-cloud-block")
    trigger_sync_run_and_wait_for_completion(
        airbyte_connection=airbyte_conn,
        connection_id="a1b2c3d4-e5f6-7890-abcd-ef1234567890",
    )