from prefect_airbyte.server import AirbyteServer

# running airbyte locally at http://localhost:8000 with default auth
local_airbyte_server = AirbyteServer()

# running airbyte remotely at http://<someIP>:<somePort> as user `Marvin`
remote_airbyte_server = AirbyteServer(
    username="Marvin",
    password="DontPanic42",
    server_host="42.42.42.42",
    server_port="4242"
)

local_airbyte_server.save("my-local-airbyte-server")

remote_airbyte_server.save("my-remote-airbyte-server")

from prefect import flow
from prefect_airbyte.server import AirbyteServer
from prefect_airbyte.connections import AirbyteConnection
from prefect_airbyte.flows import run_connection_sync

server = AirbyteServer(server_host="localhost", server_port=8000)

connection = AirbyteConnection(
    airbyte_server=server,
    connection_id="e1b2078f-882a-4f50-9942-cfe34b2d825b",
    status_updates=True,
)

@flow
def airbyte_syncs():
    # do some setup

    sync_result = run_connection_sync(
        airbyte_connection=connection,
    )

    # do some other things, like trigger DBT based on number of records synced
    print(f'Number of Records Synced: {sync_result.records_synced}')

import gzip

from prefect import flow, task
from prefect_airbyte.configuration import export_configuration
from prefect_airbyte.server import AirbyteServer

@task
def zip_and_write_somewhere(
      airbyte_config: bytearray,
      somewhere: str,
):
    with gzip.open(somewhere, 'wb') as f:
        f.write(airbyte_config)

@flow
def example_export_configuration_flow(filepath: str):

    # Run other tasks and subflows here

    airbyte_config = export_configuration(
        airbyte_server=AirbyteServer.load("my-airbyte-server-block")
    )

    zip_and_write_somewhere(
        somewhere=filepath,
        airbyte_config=airbyte_config
    )

if __name__ == "__main__":
    example_export_configuration_flow('*://**/my_destination.gz')

from prefect import flow
from prefect_airbyte.connections import AirbyteConnection
from prefect_airbyte.flows import run_connection_sync

custom_run_connection_sync = run_connection_sync.with_options(
    name="Custom Airbyte Sync Flow",
    retries=2,
    retry_delay_seconds=10,
)
 
 @flow
 def some_airbyte_flow():
    custom_run_connection_sync(
        airbyte_connection=AirbyteConnection.load("my-airbyte-connection-block")
    )
 
 some_airbyte_flow()