MANUAL

Connections List

Job History

AUTH_SERVER_API_CONNECTION

<http://localhost:8080/>

AUTH_SERVER_API_CONNECTION

python from airflow.models import Variable  # Set the variable if it doesn't exist if not Variable.get("AUTH_SERVER_API_CONNECTION", default_var=None):     Variable.set("AUTH_SERVER_API_CONNECTION", "your_connection_value")

AUTH_SERVER_API_CONNECTION

AUTH_SERVER_API_CONNECTION

plaintext <http://your-airbyte-instance:8000/api/v1>

plaintext <http://your-airbyte-instance:8000/api/v1?token=your_api_token>

<http://localhost:8080/>

AUTH_SERVER_API_CONNECTION

python from airflow.models import Variable  # Set the variable if it doesn't exist if not Variable.get("AUTH_SERVER_API_CONNECTION", default_var=None):     Variable.set("AUTH_SERVER_API_CONNECTION", "<http://your-airbyte-instance:8000/api/v1>")

AUTH_SERVER_API_CONNECTION

<http://your-airbyte-instance:8000/api/v1>

your-airbyte-instance

<https://api.airbyte.com>

<https://api.airbyte.com>

<https://api.airbyte.com>

plaintext <http://localhost:8000/api/v1>

plaintext http://<your-server-ip>:8000/api/v1

<http://localhost:8080/>

AUTH_SERVER_API_CONNECTION

<http://localhost:8000/api/v1>

python from airflow.models import Variable  # Set the variable if it doesn't exist if not Variable.get("AUTH_SERVER_API_CONNECTION", default_var=None):     Variable.set("AUTH_SERVER_API_CONNECTION", "<http://localhost:8000/api/v1>")

AUTH_SERVER_API_CONNECTION

json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)

jq

python import json import logging  # Example function that parses JSON def parse_json(json_string):     try:         data = json.loads(json_string)         return data     except json.JSONDecodeError as e:         logging.error(f"Failed to decode JSON: {e}")         logging.error(f"JSON input: {json_string}")         raise  # Example usage in your DAG json_string = Variable.get("some_json_variable", default_var="{}") parsed_data = parse_json(json_string)

Variable.get

AIRBYTE_SOURCE_ID

AIRBYTE_DESTINATION_ID

AIRBYTE_CONNECTION_ID

AIRBYTE_SOURCE_NAME

AIRBYTE_WORKSPACE_ID

Variable.get

Variable.get

AIRBYTE_SOURCE_ID

AIRBYTE_DESTINATION_ID

AIRBYTE_CONNECTION_ID

AIRBYTE_SOURCE_NAME

AIRBYTE_WORKSPACE_ID

Variable.get

.env

.env

.env

plaintext
     AIRBYTE_SOURCE_ID=your_source_id
     AIRBYTE_DESTINATION_ID=your_destination_id
     AIRBYTE_CONNECTION_ID=your_connection_id
     AIRBYTE_SOURCE_NAME=your_source_name
     AIRBYTE_WORKSPACE_ID=your_workspace_id

.env

python-dotenv

.env

pip install python-dotenv

.env

.env

.env

.env

.env

plaintext      AIRBYTE_SOURCE_ID=your_source_id      AIRBYTE_DESTINATION_ID=your_destination_id      AIRBYTE_CONNECTION_ID=your_connection_id      AIRBYTE_SOURCE_NAME=your_source_name      AIRBYTE_WORKSPACE_ID=your_workspace_id

.env

python-dotenv

.env

pip install python-dotenv

.env

.env

Exception: Failed to list sources: b'{"code":401,"message":"Jwt is missing"}\n'

AirbyteApiClient

Exception: Failed to list sources: b'{"code":401,"message":"Jwt is missing"}\n'

AirbyteApiClient

<http://localhost:8000/api/v1>

AIRBYTE_JWT_TOKEN

<http://localhost:8000/api/v1>

AIRBYTE_JWT_TOKEN

<http://localhost:8000/api/v1>

<https://api.airbyte.com>

<http://localhost:8000/api/v1>

<https://api.airbyte.com>

python from access_afya_elt.common_package.api_clients.airbyte_manager import AirbyteApiClient from airflow import DAG from airflow.operators.python_operator import PythonOperator from dotenv import load_dotenv import os import logging  # Load environment variables from .env file load_dotenv()  dag = DAG(     dag_id="airbyte_integration_dag",     description="A DAG to interact with Airbyte API",     schedule_interval=None, )  # Initialize AirbyteApiClient with the local base URL airbyte_client = AirbyteApiClient(base_url="<http://localhost:8000/api/v1>")  def update_source(source_id) -> None:     <http://logging.info|logging.info>(f"Updating source with ID: {source_id}")     airbyte_client.update_source(source_id)  update_source_task = PythonOperator(     task_id="update_source_task",     python_callable=update_source,     op_kwargs={"source_id": os.getenv("AIRBYTE_SOURCE_ID", "default_source_id")},     dag=dag, )  def create_connection(source_id, destination_id) -> None:     <http://logging.info|logging.info>(f"Creating connection with source ID: {source_id} and destination ID: {destination_id}")     airbyte_client.create_connection(source_id, destination_id)  create_connection_task = PythonOperator(     task_id="create_connection_task",     python_callable=create_connection,     op_kwargs={         "source_id": os.getenv("AIRBYTE_SOURCE_ID"),         "destination_id": os.getenv("AIRBYTE_DESTINATION_ID", "default_destination_id"),     },     dag=dag, )  def delete_connection(connection_id) -> None:     <http://logging.info|logging.info>(f"Deleting connection with ID: {connection_id}")     airbyte_client.delete_connection(connection_id)  delete_connection_task = PythonOperator(     task_id="delete_connection_task",     python_callable=delete_connection,     op_kwargs={"connection_id": os.getenv("AIRBYTE_CONNECTION_ID", "default_connection_id")},     dag=dag, )  def list_sources() -> str:     sources = airbyte_client.list_sources()     <http://logging.info|logging.info>(f"List of sources: {sources}")     return str(sources)  list_sources_task = PythonOperator(     task_id="list_sources_task",     python_callable=list_sources,     dag=dag, )  def create_source(name, workspace_id, configuration) -> None:     <http://logging.info|logging.info>(f"Creating source with name: {name}, workspace ID: {workspace_id}, and configuration: {configuration}")     airbyte_client.create_source(name, workspace_id, configuration)  create_source_task = PythonOperator(     task_id="create_source_task",     python_callable=create_source,     op_kwargs={         "name": os.getenv("AIRBYTE_SOURCE_NAME", "default_source_name"),         "workspace_id": os.getenv("AIRBYTE_WORKSPACE_ID", "default_workspace_id"),         "configuration": {},     },     dag=dag, )

<http://localhost:8000>

requests.exceptions.ConnectionError: HTTPConnectionPool(host='localhost', port=8000): Max retries exceeded with url: /api/v1/sources?includeDeleted=False&limit=20&offset=0 (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x7943547dbb50>: Failed to establish a new connection: [Errno 111] Connection refused'))

localhost:8000

<http://localhost:8000>

<http://localhost:8000>

requests.exceptions.ConnectionError: HTTPConnectionPool(host='localhost', port=8000): Max retries exceeded with url: /api/v1/sources?includeDeleted=False&limit=20&offset=0 (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x7943547dbb50>: Failed to establish a new connection: [Errno 111] Connection refused'))

localhost:8000

<http://localhost:8000>

<http://localhost:8000>

localhost

--network="host"

bash
docker run --network="host" your_airflow_container

localhost

hostname -I

ifconfig

ipconfig getifaddr en0

localhost

localhost

--network="host"

bash docker run --network="host" your_airflow_container

localhost

hostname -I

ifconfig

ipconfig getifaddr en0

localhost

Exception: Failed to list sources: b'Object not found.'

<http://localhost:8000/api/v1>

<http://localhost:8000>

Exception: Failed to list sources: b'Object not found.'

<http://localhost:8000/api/v1>

<http://localhost:8000>

AIRBYTE_JWT_TOKEN

AirbyteApiClient

python
import os
from access_afya_elt.common_package.api_clients.airbyte_manager import AirbyteApiClient
from airflow import DAG
from airflow.operators.python_operator import PythonOperator

dag = DAG(
    dag_id="airbyte_integration_dag",
    description="A DAG to interact with Airbyte API",
    schedule_interval=None,
)

# Initialize AirbyteApiClient with the correct base URL and JWT token
base_url = "<http://192.168.100.26:8000/api/v1>"
jwt_token = os.getenv("AIRBYTE_JWT_TOKEN")
airbyte_client = AirbyteApiClient(base_url=base_url, jwt_token=jwt_token)

source_id = os.getenv("AIRBYTE_SOURCE_ID")
destination_id = os.getenv("AIRBYTE_DESTINATION_ID")
connection_id = os.getenv("AIRBYTE_CONNECTION_ID")
name = os.getenv("AIRBYTE_SOURCE_NAME")
workspace_id = os.getenv("AIRBYTE_WORKSPACE_ID")

def update_source(source_id) -> None:
    airbyte_client.update_source(source_id)

update_source_task = PythonOperator(
    task_id="update_source_task",
    python_callable=update_source,
    op_kwargs={"source_id": source_id},
    dag=dag,
)

def create_connection(source_id, destination_id) -> None:
    airbyte_client.create_connection(source_id, destination_id)

create_connection_task = PythonOperator(
    task_id="create_connection_task",
    python_callable=create_connection,
    op_kwargs={"source_id": source_id, "destination_id": destination_id},
    dag=dag,
)

def delete_connection(connection_id) -> None:
    airbyte_client.delete_connection(connection_id)

delete_connection_task = PythonOperator(
    task_id="delete_connection_task",
    python_callable=delete_connection,
    op_kwargs={"connection_id": connection_id},
    dag=dag,
)

def list_sources() -> str:
    return str(airbyte_client.list_sources())

list_sources_task = PythonOperator(
    task_id="list_sources_task",
    python_callable=list_sources,
    dag=dag,
)

AIRBYTE_JWT_TOKEN

AIRBYTE_SOURCE_ID

AIRBYTE_DESTINATION_ID

AIRBYTE_CONNECTION_ID

AIRBYTE_SOURCE_NAME

AIRBYTE_WORKSPACE_ID

AIRBYTE_JWT_TOKEN

AirbyteApiClient

python import os from access_afya_elt.common_package.api_clients.airbyte_manager import AirbyteApiClient from airflow import DAG from airflow.operators.python_operator import PythonOperator  dag = DAG(     dag_id="airbyte_integration_dag",     description="A DAG to interact with Airbyte API",     schedule_interval=None, )  # Initialize AirbyteApiClient with the correct base URL and JWT token base_url = "<http://192.168.100.26:8000/api/v1>" jwt_token = os.getenv("AIRBYTE_JWT_TOKEN") airbyte_client = AirbyteApiClient(base_url=base_url, jwt_token=jwt_token)  source_id = os.getenv("AIRBYTE_SOURCE_ID") destination_id = os.getenv("AIRBYTE_DESTINATION_ID") connection_id = os.getenv("AIRBYTE_CONNECTION_ID") name = os.getenv("AIRBYTE_SOURCE_NAME") workspace_id = os.getenv("AIRBYTE_WORKSPACE_ID")  def update_source(source_id) -> None:     airbyte_client.update_source(source_id)  update_source_task = PythonOperator(     task_id="update_source_task",     python_callable=update_source,     op_kwargs={"source_id": source_id},     dag=dag, )  def create_connection(source_id, destination_id) -> None:     airbyte_client.create_connection(source_id, destination_id)  create_connection_task = PythonOperator(     task_id="create_connection_task",     python_callable=create_connection,     op_kwargs={"source_id": source_id, "destination_id": destination_id},     dag=dag, )  def delete_connection(connection_id) -> None:     airbyte_client.delete_connection(connection_id)  delete_connection_task = PythonOperator(     task_id="delete_connection_task",     python_callable=delete_connection,     op_kwargs={"connection_id": connection_id},     dag=dag, )  def list_sources() -> str:     return str(airbyte_client.list_sources())  list_sources_task = PythonOperator(     task_id="list_sources_task",     python_callable=list_sources,     dag=dag, )

AIRBYTE_JWT_TOKEN

AIRBYTE_SOURCE_ID

AIRBYTE_DESTINATION_ID

AIRBYTE_CONNECTION_ID

AIRBYTE_SOURCE_NAME

AIRBYTE_WORKSPACE_ID

AirbyteApiClient

airbyte_manager

AirbyteApiClient

airbyte_manager

AirbyteApiClient

jwt_token

__init__

Authorization

AirbyteApiClient

AirbyteApiClient

airbyte_manager

AirbyteApiClient

jwt_token

__init__

Authorization

AirbyteApiClient

python import json import requests from sil_wrapper_utils import CRUDMixin  class AirbyteApiClient(CRUDMixin):     """Airbyte API Client."""      def __init__(self, base_url: str = "<http://192.168.100.26:8000/api/v1>", jwt_token: str = None) -> None:         """Initialize the Airbyte API Client."""         self.base_url = base_url         self.jwt_token = jwt_token         self.headers = {             "Content-Type": "application/json",             "Authorization": f"Bearer {self.jwt_token}" if self.jwt_token else None         }      def update_source(self, source_id, **kwargs) -> None:         """Update a source in the Airbyte API."""         url = f"{self.base_url}/sources/{source_id}"         payload = {             "sourceId": source_id,         }         payload.update(kwargs)         response = <http://requests.post|requests.post>(url, json=payload, headers=self.headers)         if response.status_code == 200:             print(f"Source {source_id} updated successfully")         else:             raise Exception(f"Failed to update source {source_id}: {response.content}")          return None      def create_connection(self, source_id, destination_id, **kwargs) -> None:         """Create a new connection in the Airbyte API."""         url = f"{self.base_url}/connections"         payload = {             "sourceId": source_id,             "destinationId": destination_id,         }         payload.update(kwargs)         response = <http://requests.post|requests.post>(url, json=payload, headers=self.headers)         if response.status_code == 200:             print("Connection created successfully")         else:             raise Exception(f"Failed to create connection: {response.content}")          return None      def delete_connection(self, connection_id) -> None:         """Delete a connection in the Airbyte API."""         url = f"{self.base_url}/connections/{connection_id}"         response = requests.delete(url, headers=self.headers)         if response.status_code == 200:             print(f"Connection {connection_id} deleted successfully")         else:             raise Exception(f"Failed to delete connection {connection_id}: {response.content}")          return None      def list_sources(self, limit: int = 20, offset: int = 0) -> str:         """List sources in the Airbyte API."""         url = f"{self.base_url}/sources"         params = {"includeDeleted": False, "limit": limit, "offset": offset}         response = requests.get(url, params=params, headers=self.headers)         if response.status_code == 200:             return str(response.content)         else:             raise Exception(f"Failed to list sources: {response.content}")

AIRBYTE_JWT_TOKEN

GET

plaintext      <http://192.168.100.26:8000/api/v1/sources>

Authorization

Key: Authorization      Value: Bearer YOUR_JWT_TOKEN

200 OK

<http://192.168.100.26:8000/api/v1/sources>

Authorization: Bearer YOUR_JWT_TOKEN

bash curl -X GET "<http://192.168.100.26:8000/api/v1/sources>" \      -H "Authorization: Bearer YOUR_JWT_TOKEN" \      -H "Content-Type: application/json"

YOUR_JWT_TOKEN

<http://192.168.100.26:8000/api/v1/sources/list>

POST

list_sources

AirbyteApiClient

AirbyteApiClient

<http://192.168.100.26:8000/api/v1/sources/list>

POST

list_sources

AirbyteApiClient

AirbyteApiClient

python import json import requests from sil_wrapper_utils import CRUDMixin  class AirbyteApiClient(CRUDMixin):     """Airbyte API Client."""      def __init__(self, base_url: str = "<http://192.168.100.26:8000/api/v1>") -> None:         """Initialize the Airbyte API Client."""         self.base_url = base_url         self.headers = {             "Content-Type": "application/json"         }      def update_source(self, source_id, **kwargs) -> None:         """Update a source in the Airbyte API."""         url = f"{self.base_url}/sources/{source_id}"         payload = {             "sourceId": source_id,         }         payload.update(kwargs)         response = <http://requests.post|requests.post>(url, json=payload, headers=self.headers)         if response.status_code == 200:             print(f"Source {source_id} updated successfully")         else:             raise Exception(f"Failed to update source {source_id}: {response.content}")          return None      def create_connection(self, source_id, destination_id, **kwargs) -> None:         """Create a new connection in the Airbyte API."""         url = f"{self.base_url}/connections"         payload = {             "sourceId": source_id,             "destinationId": destination_id,         }         payload.update(kwargs)         response = <http://requests.post|requests.post>(url, json=payload, headers=self.headers)         if response.status_code == 200:             print("Connection created successfully")         else:             raise Exception(f"Failed to create connection: {response.content}")          return None      def delete_connection(self, connection_id) -> None:         """Delete a connection in the Airbyte API."""         url = f"{self.base_url}/connections/{connection_id}"         response = requests.delete(url, headers=self.headers)         if response.status_code == 200:             print(f"Connection {connection_id} deleted successfully")         else:             raise Exception(f"Failed to delete connection {connection_id}: {response.content}")          return None      def list_sources(self, limit: int = 20, offset: int = 0) -> str:         """List sources in the Airbyte API."""         url = f"{self.base_url}/sources/list"         payload = {"includeDeleted": False, "limit": limit, "offset": offset}         response = <http://requests.post|requests.post>(url, json=payload, headers=self.headers)         if response.status_code == 200:             return str(response.content)         else:             raise Exception(f"Failed to list sources: {response.content}")

AirbyteApiClient

"sources":[]

workspaceId

list_sources

list_sources

workspaceId

AirbyteApiClient

workspaceId

list_sources

"sources":[]

workspaceId

list_sources

list_sources

workspaceId

AirbyteApiClient

workspaceId

list_sources

Exception: Failed to update source e7066349-ea5b-4ec5-99ee-1a936c462c62: b'Object not found.'

source_id

update_source

update_source

update_source

update_source

AirbyteApiClient

update_source

Exception: Failed to update source e7066349-ea5b-4ec5-99ee-1a936c462c62: b'Object not found.'

source_id

update_source

update_source

update_source

update_source

AirbyteApiClient

update_source

update_source

update_source_task

AirbyteApiClient

python
import os
from access_afya_elt.common_package.api_clients.airbyte_manager import AirbyteApiClient
from airflow import DAG
from airflow.operators.python_operator import PythonOperator

dag = DAG(
    dag_id="airbyte_integration_dag",
    description="A DAG to interact with Airbyte API",
    schedule_interval=None,
)

# Initialize AirbyteApiClient with the correct base URL
base_url = "<http://192.168.100.26:8000/api/v1>"
airbyte_client = AirbyteApiClient(base_url=base_url)

source_id = os.getenv("AIRBYTE_SOURCE_ID")
destination_id = os.getenv("AIRBYTE_DESTINATION_ID")
connection_id = os.getenv("AIRBYTE_CONNECTION_ID")
name = os.getenv("AIRBYTE_SOURCE_NAME")
workspace_id = os.getenv("AIRBYTE_WORKSPACE_ID")

# Define the new connection configuration
new_connection_configuration = {
    "base": "EUR",
    "access_key": "123456789qwertyuiopasdfghjklzxcvbnm",
    "start_date": "2021-01-01",
    "ignore_weekends": True,
}

def update_source(source_id, connection_configuration) -> None:
    airbyte_client.update_source(source_id, **connection_configuration)

update_source_task = PythonOperator(
    task_id="update_source_task",
    python_callable=update_source,
    op_kwargs={"source_id": source_id,
               "connection_configuration": new_connection_configuration},
    dag=dag,
)

def create_connection(source_id, destination_id) -> None:
    airbyte_client.create_connection(source_id, destination_id)

create_connection_task = PythonOperator(
    task_id="create_connection_task",
    python_callable=create_connection,
    op_kwargs={"source_id": source_id, "destination_id": destination_id},
    dag=dag,
)

def delete_connection(connection_id) -> None:
    airbyte_client.delete_connection(connection_id)

delete_connection_task = PythonOperator(
    task_id="delete_connection_task",
    python_callable=delete_connection,
    op_kwargs={"connection_id": connection_id},
    dag=dag,
)

def list_sources() -> str:
    return str(airbyte_client.list_sources(workspace_id=workspace_id))

list_sources_task = PythonOperator(
    task_id="list_sources_task",
    python_callable=list_sources,
    dag=dag,
)

update_source

update_source

connection_configuration

update_source

AirbyteApiClient

update_source_task

new_connection_configuration

update_source

op_kwargs

workspace_id

list_sources

list_sources

workspace_id

update_source

update_source_task

AirbyteApiClient

python import os from access_afya_elt.common_package.api_clients.airbyte_manager import AirbyteApiClient from airflow import DAG from airflow.operators.python_operator import PythonOperator  dag = DAG(     dag_id="airbyte_integration_dag",     description="A DAG to interact with Airbyte API",     schedule_interval=None, )  # Initialize AirbyteApiClient with the correct base URL base_url = "<http://192.168.100.26:8000/api/v1>" airbyte_client = AirbyteApiClient(base_url=base_url)  source_id = os.getenv("AIRBYTE_SOURCE_ID") destination_id = os.getenv("AIRBYTE_DESTINATION_ID") connection_id = os.getenv("AIRBYTE_CONNECTION_ID") name = os.getenv("AIRBYTE_SOURCE_NAME") workspace_id = os.getenv("AIRBYTE_WORKSPACE_ID")  # Define the new connection configuration new_connection_configuration = {     "base": "EUR",     "access_key": "123456789qwertyuiopasdfghjklzxcvbnm",     "start_date": "2021-01-01",     "ignore_weekends": True, }  def update_source(source_id, connection_configuration) -> None:     airbyte_client.update_source(source_id, **connection_configuration)  update_source_task = PythonOperator(     task_id="update_source_task",     python_callable=update_source,     op_kwargs={"source_id": source_id,                "connection_configuration": new_connection_configuration},     dag=dag, )  def create_connection(source_id, destination_id) -> None:     airbyte_client.create_connection(source_id, destination_id)  create_connection_task = PythonOperator(     task_id="create_connection_task",     python_callable=create_connection,     op_kwargs={"source_id": source_id, "destination_id": destination_id},     dag=dag, )  def delete_connection(connection_id) -> None:     airbyte_client.delete_connection(connection_id)  delete_connection_task = PythonOperator(     task_id="delete_connection_task",     python_callable=delete_connection,     op_kwargs={"connection_id": connection_id},     dag=dag, )  def list_sources() -> str:     return str(airbyte_client.list_sources(workspace_id=workspace_id))  list_sources_task = PythonOperator(     task_id="list_sources_task",     python_callable=list_sources,     dag=dag, )

update_source

update_source

connection_configuration

update_source

AirbyteApiClient

update_source_task

new_connection_configuration

update_source

op_kwargs

workspace_id

list_sources

list_sources

workspace_id

source_id

workspace_id

source_id

workspace_id

update_source

bash
curl -X POST "<http://192.168.100.26:8000/api/v1/sources/update>" \
     -H "Content-Type: application/json" \
     -d '{
           "sourceId": "e7066349-ea5b-4ec5-99ee-1a936c462c62",
           "workspaceId": "your_workspace_id",
           "configuration": {
             "base": "EUR",
             "access_key": "123456789qwertyuiopasdfghjklzxcvbnm",
             "start_date": "2021-01-01",
             "ignore_weekends": true
           }
         }'

POST

plaintext
     <http://192.168.100.26:8000/api/v1/sources/update>

Content-Type

Key: Content-Type
     Value: application/json

json
     {
       "sourceId": "e7066349-ea5b-4ec5-99ee-1a936c462c62",
       "workspaceId": "your_workspace_id",
       "configuration": {
         "base": "EUR",
         "access_key": "123456789qwertyuiopasdfghjklzxcvbnm",
         "start_date": "2021-01-01",
         "ignore_weekends": true
       }
     }

sourceId

workspaceId

source_id

workspace_id

source_id

workspace_id

update_source

bash curl -X POST "<http://192.168.100.26:8000/api/v1/sources/update>" \      -H "Content-Type: application/json" \      -d '{            "sourceId": "e7066349-ea5b-4ec5-99ee-1a936c462c62",            "workspaceId": "your_workspace_id",            "configuration": {              "base": "EUR",              "access_key": "123456789qwertyuiopasdfghjklzxcvbnm",              "start_date": "2021-01-01",              "ignore_weekends": true            }          }'

POST

plaintext      <http://192.168.100.26:8000/api/v1/sources/update>

Content-Type

Key: Content-Type      Value: application/json

json      {        "sourceId": "e7066349-ea5b-4ec5-99ee-1a936c462c62",        "workspaceId": "your_workspace_id",        "configuration": {          "base": "EUR",          "access_key": "123456789qwertyuiopasdfghjklzxcvbnm",          "start_date": "2021-01-01",          "ignore_weekends": true        }      }

sourceId

workspaceId

update_source

update_source

plaintext <http://192.168.100.26:8000/api/v1/sources/update>

POST

plaintext      <http://192.168.100.26:8000/api/v1/sources/update>

Content-Type

Key: Content-Type      Value: application/json

"your_workspace_id"

"e7066349-ea5b-4ec5-99ee-1a936c462c62"

json      {        "sourceId": "e7066349-ea5b-4ec5-99ee-1a936c462c62",        "workspaceId": "your_workspace_id",        "configuration": {          "base": "EUR",          "access_key": "123456789qwertyuiopasdfghjklzxcvbnm",          "start_date": "2021-01-01",          "ignore_weekends": true        }      }

bash curl -X POST "<http://192.168.100.26:8000/api/v1/sources/update>" \      -H "Content-Type: application/json" \      -d '{            "sourceId": "e7066349-ea5b-4ec5-99ee-1a936c462c62",            "workspaceId": "your_workspace_id",            "configuration": {              "base": "EUR",              "access_key": "123456789qwertyuiopasdfghjklzxcvbnm",              "start_date": "2021-01-01",              "ignore_weekends": true            }          }'

update_source

EHOSTUNREACH

192.168.100.26:8000

<http://192.168.100.26:8000>

8000

192.168.100.26

8000

bash      ping 192.168.100.26

192.168.100.26:8000

localhost

plaintext      <http://localhost:8000/api/v1/sources/update>

POST

plaintext      <http://192.168.100.26:8000/api/v1/sources/update>

Content-Type

Key: Content-Type      Value: application/json

"your_workspace_id"

"e7066349-ea5b-4ec5-99ee-1a936c462c62"

json      {        "sourceId": "e7066349-ea5b-4ec5-99ee-1a936c462c62",        "workspaceId": "your_workspace_id",        "configuration": {          "base": "EUR",          "access_key": "123456789qwertyuiopasdfghjklzxcvbnm",          "start_date": "2021-01-01",          "ignore_weekends": true        }      }

EHOSTUNREACH

update_source

Destination Host Unreachable

192.168.100.26

192.168.100.26

192.168.100.26

8000

localhost

plaintext      <http://localhost:8000/api/v1/sources/update>

8000

POST

plaintext      <http://192.168.100.26:8000/api/v1/sources/update>

Content-Type

Key: Content-Type      Value: application/json

"your_workspace_id"

"e7066349-ea5b-4ec5-99ee-1a936c462c62"

json      {        "sourceId": "e7066349-ea5b-4ec5-99ee-1a936c462c62",        "workspaceId": "your_workspace_id",        "configuration": {          "base": "EUR",          "access_key": "123456789qwertyuiopasdfghjklzxcvbnm",          "start_date": "2021-01-01",          "ignore_weekends": true        }      }

Destination Host Unreachable

update_source

Internal Server Error: Parameter specified as non-null is null: method io.airbyte.config.secrets.JsonSecretsProcessor.copySecrets, parameter dst

dst

copySecrets

JsonSecretsProcessor

updateSource

update_source

update_source

AirbyteApiClient

json
{
  "sourceId": "e7066349-ea5b-4ec5-99ee-1a936c462c62",
  "workspaceId": "your_workspace_id",
  "configuration": {
    "base": "EUR",
    "access_key": "123456789qwertyuiopasdfghjklzxcvbnm",
    "start_date": "2021-01-01",
    "ignore_weekends": true
  }
}

AirbyteApiClient

update_source

AirbyteApiClient

python
import json
import requests
from sil_wrapper_utils import CRUDMixin

class AirbyteApiClient(CRUDMixin):
    """Airbyte API Client."""

    def __init__(self, base_url: str = "<http://localhost:8000/api/v1>", jwt_token: str = None) -> None:
        """Initialize the Airbyte API Client."""
        self.base_url = base_url
        self.jwt_token = jwt_token
        self.headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.jwt_token}" if self.jwt_token else None
        }

    def update_source(self, source_id, **kwargs) -> None:
        """Update a source in the Airbyte API."""
        url = f"{self.base_url}/sources/update"
        payload = {
            "sourceId": source_id,
            "workspaceId": kwargs.get("workspaceId"),
            "configuration": kwargs.get("configuration")
        }
        response = <http://requests.post|requests.post>(url, json=payload, headers=self.headers)
        if response.status_code == 200:
            print(f"Source {source_id} updated successfully")
        else:
            raise Exception(f"Failed to update source {source_id}: {response.content}")

        return None

AirbyteApiClient

Internal Server Error: Parameter specified as non-null is null: method io.airbyte.config.secrets.JsonSecretsProcessor.copySecrets, parameter dst

dst

copySecrets

JsonSecretsProcessor

updateSource

update_source

update_source

AirbyteApiClient

json {   "sourceId": "e7066349-ea5b-4ec5-99ee-1a936c462c62",   "workspaceId": "your_workspace_id",   "configuration": {     "base": "EUR",     "access_key": "123456789qwertyuiopasdfghjklzxcvbnm",     "start_date": "2021-01-01",     "ignore_weekends": true   } }

AirbyteApiClient

update_source

AirbyteApiClient

python import json import requests from sil_wrapper_utils import CRUDMixin  class AirbyteApiClient(CRUDMixin):     """Airbyte API Client."""      def __init__(self, base_url: str = "<http://localhost:8000/api/v1>", jwt_token: str = None) -> None:         """Initialize the Airbyte API Client."""         self.base_url = base_url         self.jwt_token = jwt_token         self.headers = {             "Content-Type": "application/json",             "Authorization": f"Bearer {self.jwt_token}" if self.jwt_token else None         }      def update_source(self, source_id, **kwargs) -> None:         """Update a source in the Airbyte API."""         url = f"{self.base_url}/sources/update"         payload = {             "sourceId": source_id,             "workspaceId": kwargs.get("workspaceId"),             "configuration": kwargs.get("configuration")         }         response = <http://requests.post|requests.post>(url, json=payload, headers=self.headers)         if response.status_code == 200:             print(f"Source {source_id} updated successfully")         else:             raise Exception(f"Failed to update source {source_id}: {response.content}")          return None

AirbyteApiClient

dst

copySecrets

JsonSecretsProcessor

updateSource

update_source

update_source

AirbyteApiClient

json
{
  "sourceId": "e7066349-ea5b-4ec5-99ee-1a936c462c62",
  "workspaceId": "your_workspace_id",
  "configuration": {
    "base": "EUR",
    "access_key": "123456789qwertyuiopasdfghjklzxcvbnm",
    "start_date": "2021-01-01",
    "ignore_weekends": true
  }
}

AirbyteApiClient

update_source

AirbyteApiClient

python
import json
import requests
from sil_wrapper_utils import CRUDMixin

class AirbyteApiClient(CRUDMixin):
    """Airbyte API Client."""

    def __init__(self, base_url: str = "<http://localhost:8000/api/v1>", jwt_token: str = None) -> None:
        """Initialize the Airbyte API Client."""
        self.base_url = base_url
        self.jwt_token = jwt_token
        self.headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.jwt_token}" if self.jwt_token else None
        }

    def update_source(self, source_id, workspace_id, configuration) -> None:
        """Update a source in the Airbyte API."""
        url = f"{self.base_url}/sources/update"
        payload = {
            "sourceId": source_id,
            "workspaceId": workspace_id,
            "configuration": configuration
        }
        response = <http://requests.post|requests.post>(url, json=payload, headers=self.headers)
        if response.status_code == 200:
            print(f"Source {source_id} updated successfully")
        else:
            raise Exception(f"Failed to update source {source_id}: {response.content}")

        return None

AirbyteApiClient

dst

copySecrets

JsonSecretsProcessor

updateSource

update_source

update_source

AirbyteApiClient

json {   "sourceId": "e7066349-ea5b-4ec5-99ee-1a936c462c62",   "workspaceId": "your_workspace_id",   "configuration": {     "base": "EUR",     "access_key": "123456789qwertyuiopasdfghjklzxcvbnm",     "start_date": "2021-01-01",     "ignore_weekends": true   } }

AirbyteApiClient

update_source

AirbyteApiClient

python import json import requests from sil_wrapper_utils import CRUDMixin  class AirbyteApiClient(CRUDMixin):     """Airbyte API Client."""      def __init__(self, base_url: str = "<http://localhost:8000/api/v1>", jwt_token: str = None) -> None:         """Initialize the Airbyte API Client."""         self.base_url = base_url         self.jwt_token = jwt_token         self.headers = {             "Content-Type": "application/json",             "Authorization": f"Bearer {self.jwt_token}" if self.jwt_token else None         }      def update_source(self, source_id, workspace_id, configuration) -> None:         """Update a source in the Airbyte API."""         url = f"{self.base_url}/sources/update"         payload = {             "sourceId": source_id,             "workspaceId": workspace_id,             "configuration": configuration         }         response = <http://requests.post|requests.post>(url, json=payload, headers=self.headers)         if response.status_code == 200:             print(f"Source {source_id} updated successfully")         else:             raise Exception(f"Failed to update source {source_id}: {response.content}")          return None

AirbyteApiClient

plaintext      <http://localhost:8000/api/v1/sources/update>

POST

Content-Type

application/json

json      {        "sourceId": "e7066349-ea5b-4ec5-99ee-1a936c462c62",        "workspaceId": "your_workspace_id",        "configuration": {          "base": "EUR",          "access_key": "123456789qwertyuiopasdfghjklzxcvbnm",          "start_date": "2021-01-01",          "ignore_weekends": true        }      }

POST

plaintext      <http://localhost:8000/api/v1/sources/update>

Content-Type

Key: Content-Type      Value: application/json

"your_workspace_id"

"e7066349-ea5b-4ec5-99ee-1a936c462c62"

json      {        "sourceId": "e7066349-ea5b-4ec5-99ee-1a936c462c62",        "workspaceId": "your_workspace_id",        "configuration": {          "base": "EUR",          "access_key": "123456789qwertyuiopasdfghjklzxcvbnm",          "start_date": "2021-01-01",          "ignore_weekends": true        }      }

Internal Server Error

update_source

sourceId

plaintext      <http://localhost:8000/api/v1/sources/update>

sourceId

sourceId

POST

plaintext      <http://localhost:8000/api/v1/sources/update>

Content-Type

Key: Content-Type      Value: application/json

"your_workspace_id"

"e7066349-ea5b-4ec5-99ee-1a936c462c62"

json      {        "sourceId": "e7066349-ea5b-4ec5-99ee-1a936c462c62",        "workspaceId": "64cf55a5-7a9c-4a15-8928-e3d2eddfaeb7",        "name": "Exchange_Rates_Api",        "configuration": {          "sourceType": "exchange-rates",          "access_key": "O30wXHn6VKKqUYCOa9Wt4i9JkhjPHgBs",          "start_date": "2020-05-01",          "ignore_weekends": true,          "base": "EUR"        }      }

sourceId

POST

plaintext      <http://localhost:8000/api/v1/sources/list>

Content-Type

Key: Content-Type      Value: application/json

update_source

update_source

plaintext
<http://localhost:8000/api/v1/sources/update>

update_source

python
import json
import requests

class AirbyteApiClient:
    """Airbyte API Client."""

    def __init__(self, base_url: str = "<http://localhost:8000/api/v1>", jwt_token: str = None) -> None:
        """Initialize the Airbyte API Client."""
        self.base_url = base_url
        self.jwt_token = jwt_token
        self.headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.jwt_token}" if self.jwt_token else None
        }

    def update_source(self, source_id, workspace_id, configuration) -> None:
        """Update a source in the Airbyte API."""
        url = f"{self.base_url}/sources/update"
        payload = {
            "sourceId": source_id,
            "workspaceId": workspace_id,
            "configuration": configuration
        }
        headers = {'Content-Type': 'application/json'}
        # Log the request details
        print(f"Sending POST request to URL: {url}")
        print(f"Payload: {json.dumps(payload, indent=2)}")
        print(f"Headers: {headers}")

        response = <http://requests.post|requests.post>(url, json=payload, headers=headers)

        # Log the response details
        print(f"Response Status Code: {response.status_code}")
        print(f"Response Content: {response.content}")

        if response.status_code == 200:
            print(f"Source {source_id} updated successfully")
        else:
            raise Exception(f"Failed to update source {source_id}: {response.content}")

        return None

POST

plaintext
     <http://localhost:8000/api/v1/sources/update>

Content-Type

Key: Content-Type
     Value: application/json

update_source

update_source

plaintext <http://localhost:8000/api/v1/sources/update>

update_source

python import json import requests  class AirbyteApiClient:     """Airbyte API Client."""      def __init__(self, base_url: str = "<http://localhost:8000/api/v1>", jwt_token: str = None) -> None:         """Initialize the Airbyte API Client."""         self.base_url = base_url         self.jwt_token = jwt_token         self.headers = {             "Content-Type": "application/json",             "Authorization": f"Bearer {self.jwt_token}" if self.jwt_token else None         }      def update_source(self, source_id, workspace_id, configuration) -> None:         """Update a source in the Airbyte API."""         url = f"{self.base_url}/sources/update"         payload = {             "sourceId": source_id,             "workspaceId": workspace_id,             "configuration": configuration         }         headers = {'Content-Type': 'application/json'}         # Log the request details         print(f"Sending POST request to URL: {url}")         print(f"Payload: {json.dumps(payload, indent=2)}")         print(f"Headers: {headers}")          response = <http://requests.post|requests.post>(url, json=payload, headers=headers)          # Log the response details         print(f"Response Status Code: {response.status_code}")         print(f"Response Content: {response.content}")          if response.status_code == 200:             print(f"Source {source_id} updated successfully")         else:             raise Exception(f"Failed to update source {source_id}: {response.content}")          return None

POST

plaintext      <http://localhost:8000/api/v1/sources/update>

Content-Type

Key: Content-Type      Value: application/json

"your_workspace_id"

"e7066349-ea5b-4ec5-99ee-1a936c462c62"

update_source

update_source

update_source

AirbyteApiClient

update_source

update_source

update_source

python
import json
import requests

class AirbyteApiClient:
    """Airbyte API Client."""

    def __init__(self, base_url: str = "<http://localhost:8000/api/v1>", jwt_token: str = None) -> None:
        """Initialize the Airbyte API Client."""
        self.base_url = base_url
        self.jwt_token = jwt_token
        self.headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.jwt_token}" if self.jwt_token else None
        }

    def update_source(self, source_id, workspace_id, configuration) -> None:
        """Update a source in the Airbyte API."""
        url = f"{self.base_url}/sources/update"
        payload = {
            "sourceId": source_id,
            "workspaceId": workspace_id,
            "configuration": configuration
        }
        headers = {'Content-Type': 'application/json'}
        # Log the request details
        print(f"Sending POST request to URL: {url}")
        print(f"Payload: {json.dumps(payload, indent=2)}")
        print(f"Headers: {headers}")

        response = <http://requests.post|requests.post>(url, json=payload, headers=headers)

        # Log the response details
        print(f"Response Status Code: {response.status_code}")
        print(f"Response Content: {response.content}")

        if response.status_code == 200:
            print(f"Source {source_id} updated successfully")
        else:
            raise Exception(f"Failed to update source {source_id}: {response.content}")

        return None

update_source

update_source

update_source

update_source

AirbyteApiClient

update_source

update_source

update_source

python import json import requests  class AirbyteApiClient:     """Airbyte API Client."""      def __init__(self, base_url: str = "<http://localhost:8000/api/v1>", jwt_token: str = None) -> None:         """Initialize the Airbyte API Client."""         self.base_url = base_url         self.jwt_token = jwt_token         self.headers = {             "Content-Type": "application/json",             "Authorization": f"Bearer {self.jwt_token}" if self.jwt_token else None         }      def update_source(self, source_id, workspace_id, configuration) -> None:         """Update a source in the Airbyte API."""         url = f"{self.base_url}/sources/update"         payload = {             "sourceId": source_id,             "workspaceId": workspace_id,             "configuration": configuration         }         headers = {'Content-Type': 'application/json'}         # Log the request details         print(f"Sending POST request to URL: {url}")         print(f"Payload: {json.dumps(payload, indent=2)}")         print(f"Headers: {headers}")          response = <http://requests.post|requests.post>(url, json=payload, headers=headers)          # Log the response details         print(f"Response Status Code: {response.status_code}")         print(f"Response Content: {response.content}")          if response.status_code == 200:             print(f"Source {source_id} updated successfully")         else:             raise Exception(f"Failed to update source {source_id}: {response.content}")          return None

update_source

plaintext <http://localhost:8000/api/v1/sources/{source_id}>

source_id

PATCH

{source_id}

plaintext      <http://localhost:8000/api/v1/sources/e7066349-ea5b-4ec5-99ee-1a936c462c62>

Content-Type

Key: Content-Type      Value: application/json

"your_workspace_id"

json      {        "workspaceId": "64cf55a5-7a9c-4a15-8928-e3d2eddfaeb7",        "name": "Exchange_Rates_Api",        "configuration": {          "sourceType": "exchange-rates",          "access_key": "O30wXHn6VKKqUYCOa9Wt4i9JkhjPHgBs",          "start_date": "2020-05-01",          "ignore_weekends": true,          "base": "EUR"        }      }

update_source

update_source

definitionId

definitionId

definitionId

plaintext      <http://localhost:8000/api/v1/source_definitions/list>

POST

<http://localhost:8000/api/v1/source_definitions/list>

Key: Content-Type      Value: application/json

bash    curl -X POST "<http://localhost:8000/api/v1/source_definitions/list>" \         -H "Content-Type: application/json" \         -d '{}'

definitionId

json      {        "sourceDefinitions": [          {            "sourceDefinitionId": "ef69ef6e-aa7f-4af1-a01d-ef775033524e",            "name": "Postgres",            "dockerRepository": "airbyte/source-postgres",            "dockerImageTag": "0.3.2",            "documentationUrl": "<https://docs.airbyte.io/integrations/sources/postgres>"          },          {            "sourceDefinitionId": "b3c5e883-1b1c-4a1b-8c8b-1a2b3c4d5e6f",            "name": "MySQL",            "dockerRepository": "airbyte/source-mysql",            "dockerImageTag": "0.2.1",            "documentationUrl": "<https://docs.airbyte.io/integrations/sources/mysql>"          }          // More source definitions...        ]      }

definitionId

definitionId

plaintext      <http://localhost:8000/api/v1/sources/create>

POST

<http://localhost:8000/api/v1/sources/create>

Key: Content-Type      Value: application/json

json      {        "name": "My New Source",        "workspaceId": "your_workspace_id",        "sourceDefinitionId": "ef69ef6e-aa7f-4af1-a01d-ef775033524e",        "connectionConfiguration": {          "host": "localhost",          "port": 5432,          "database": "my_database",          "username": "my_username",          "password": "my_password"        }      }

Failed to create source. Status Code: 404

b'Object not found.'

definitionId

plaintext      <http://localhost:8000/api/v1/sources/create>

definitionId

definitionId

definitionId

plaintext      <http://localhost:8000/api/v1/source_definitions/list>

POST

<http://localhost:8000/api/v1/source_definitions/list>

Key: Content-Type      Value: application/json

bash    curl -X POST "<http://localhost:8000/api/v1/source_definitions/list>" \         -H "Content-Type: application/json" \         -d '{}'

definitionId

json      {        "sourceDefinitions": [          {            "sourceDefinitionId": "ef69ef6e-aa7f-4af1-a01d-ef775033524e",            "name": "Postgres",            "dockerRepository": "airbyte/source-postgres",            "dockerImageTag": "0.3.2",            "documentationUrl": "<https://docs.airbyte.io/integrations/sources/postgres>"          },          {            "sourceDefinitionId": "b3c5e883-1b1c-4a1b-8c8b-1a2b3c4d5e6f",            "name": "MySQL",            "dockerRepository": "airbyte/source-mysql",            "dockerImageTag": "0.2.1",            "documentationUrl": "<https://docs.airbyte.io/integrations/sources/mysql>"          }          // More source definitions...        ]      }

definitionId

plaintext      <http://localhost:8000/api/v1/sources/create>

POST

<http://localhost:8000/api/v1/sources/create>

Key: Content-Type      Value: application/json

Internal Server Error: null

java.lang.NullPointerException

definitionId

definitionId

definitionId

plaintext      <http://localhost:8000/api/v1/source_definitions/list>

POST

<http://localhost:8000/api/v1/source_definitions/list>

Key: Content-Type      Value: application/json

bash    curl -X POST "<http://localhost:8000/api/v1/source_definitions/list>" \         -H "Content-Type: application/json" \         -d '{}'

definitionId

json      {        "sourceDefinitions": [          {            "sourceDefinitionId": "ef69ef6e-aa7f-4af1-a01d-ef775033524e",            "name": "Postgres",            "dockerRepository": "airbyte/source-postgres",            "dockerImageTag": "0.3.2",            "documentationUrl": "<https://docs.airbyte.io/integrations/sources/postgres>"          },          {            "sourceDefinitionId": "b3c5e883-1b1c-4a1b-8c8b-1a2b3c4d5e6f",            "name": "MySQL",            "dockerRepository": "airbyte/source-mysql",            "dockerImageTag": "0.2.1",            "documentationUrl": "<https://docs.airbyte.io/integrations/sources/mysql>"          }          // More source definitions...        ]      }

definitionId

plaintext      <http://localhost:8000/api/v1/sources/create>

POST

<http://localhost:8000/api/v1/sources/create>

Key: Content-Type      Value: application/json

plaintext      <http://localhost:8000/api/v1/sources/create>

definitionId

sourceDefinitionId

json {     "workspaceId": "64cf55a5-7a9c-4a15-8928-e3d2eddfaeb7",     "name": "Exchange Rates Api",     "sourceDefinitionId": "e2b40e36-aa0e-4bed-b41b-bcea6fa348b1",     "connectionConfiguration": {         "start_date": "2020-05-01",         "access_key": "O30wXHn6VKKqUYCOa9Wt4i9JkhjPHgBs",         "base": "EUR",         "sourceType": "exchange-rates"     } }

POST

plaintext      <http://localhost:8000/api/v1/sources/create>

Content-Type

Key: Content-Type      Value: application/json

json      {          "workspaceId": "64cf55a5-7a9c-4a15-8928-e3d2eddfaeb7",          "name": "Exchange Rates Api",          "sourceDefinitionId": "e2b40e36-aa0e-4bed-b41b-bcea6fa348b1",          "connectionConfiguration": {              "start_date": "2020-05-01",              "access_key": "O30wXHn6VKKqUYCOa9Wt4i9JkhjPHgBs",              "base": "EUR",              "sourceType": "exchange-rates"          }      }

Could not find configuration for STANDARD_DESTINATION_DEFINITION: 64cf55a5-7a9c-4a15-8928-e3d2eddfaeb7

destinationDefinitionId

destinationDefinitionId

destinationDefinitionId

destinationDefinitionId

destinationDefinitionId

plaintext      <http://localhost:8000/api/v1/destination_definitions/list>

POST

<http://localhost:8000/api/v1/destination_definitions/list>

Key: Content-Type      Value: application/json

bash    curl -X POST "<http://localhost:8000/api/v1/destination_definitions/list>" \         -H "Content-Type: application/json" \         -d '{}'

destinationDefinitionId

json      {        "destinationDefinitions": [          {            "destinationDefinitionId": "8be1cf83-fde1-477f-a4ad-318d23c9f3c6",            "name": "Postgres",            "dockerRepository": "airbyte/destination-postgres",            "dockerImageTag": "0.3.2",            "documentationUrl": "<https://docs.airbyte.io/integrations/destinations/postgres>"          },          {            "destinationDefinitionId": "25c5221d-dce2-4163-ade9-739ef790f503",            "name": "MySQL",            "dockerRepository": "airbyte/destination-mysql",            "dockerImageTag": "0.2.1",            "documentationUrl": "<https://docs.airbyte.io/integrations/destinations/mysql>"          }          // More destination definitions...        ]      }

destinationDefinitionId

plaintext      <http://localhost:8000/api/v1/destinations/create>

POST

<http://localhost:8000/api/v1/destinations/create>

Key: Content-Type      Value: application/json

python
# Step 1: Create Source
try:
    connectionConfiguration = client.get_source_configuration()
    source_id = client.create_source(
        client.name, client.workspaceId, client.sourceDefinitionId, connectionConfiguration
    )
    print(f"Created source with ID: {source_id}")

    # Step 2: Create Destination
    connectionConfiguration = client.get_destination_configuration()
    destination_id = client.create_destination(
        client.destination_name, client.workspaceId, client.destinationDefinitionId, connectionConfiguration
    )
    print(f"Created destination with ID: {destination_id}")

    # Step 3: Create Connection
    connection_id = client.create_connection(source_id, destination_id)
    print(f"Created connection with ID: {connection_id}")

except Exception as e:
    print(f"Error: {e}")

plaintext
     <http://localhost:8000/api/v1/connections/create>

sourceId

destinationId

name

schedule

status

json
{
  "name": "My Connection",
  "sourceId": "source_id",
  "destinationId": "destination_id",
  "syncCatalog": {
    "streams": [
      {
        "stream": {
          "name": "my_stream",
          "namespace": "public"
        },
        "config": {
          "syncMode": "full_refresh",
          "destinationSyncMode": "overwrite"
        }
      }
    ]
  },
  "schedule": {
    "units": 1,
    "timeUnit": "hours"
  },
  "status": "active"
}

python # Step 1: Create Source try:     connectionConfiguration = client.get_source_configuration()     source_id = client.create_source(         client.name, client.workspaceId, client.sourceDefinitionId, connectionConfiguration     )     print(f"Created source with ID: {source_id}")      # Step 2: Create Destination     connectionConfiguration = client.get_destination_configuration()     destination_id = client.create_destination(         client.destination_name, client.workspaceId, client.destinationDefinitionId, connectionConfiguration     )     print(f"Created destination with ID: {destination_id}")      # Step 3: Create Connection     connection_id = client.create_connection(source_id, destination_id)     print(f"Created connection with ID: {connection_id}")  except Exception as e:     print(f"Error: {e}")

plaintext      <http://localhost:8000/api/v1/connections/create>

sourceId

destinationId

name

schedule

status

json {   "name": "My Connection",   "sourceId": "source_id",   "destinationId": "destination_id",   "syncCatalog": {     "streams": [       {         "stream": {           "name": "my_stream",           "namespace": "public"         },         "config": {           "syncMode": "full_refresh",           "destinationSyncMode": "overwrite"         }       }     ]   },   "schedule": {     "units": 1,     "timeUnit": "hours"   },   "status": "active" }

Could not find configuration for SOURCE_CONNECTION: 8ae47576-ecb7-430c-9f52-a86b40a33f33

sourceId

destinationId

sourceId

destinationId

python
# Step 1: Create Source
try:
    connectionConfiguration = client.get_source_configuration()
    source_id = client.create_source(
        client.name, client.workspaceId, client.sourceDefinitionId, connectionConfiguration
    )
    print(f"Created source with ID: {source_id}")

    # Step 2: Create Destination
    connectionConfiguration = client.get_destination_configuration()
    destination_id = client.create_destination(
        client.destination_name, client.workspaceId, client.destinationDefinitionId, connectionConfiguration
    )
    print(f"Created destination with ID: {destination_id}")

    # Step 3: Create Connection
    connection_id = client.create_connection(source_id, destination_id, client.workspaceId)
    print(f"Created connection with ID: {connection_id}")

except Exception as e:
    print(f"Error: {e}")

plaintext
     <http://localhost:8000/api/v1/connections/create>

sourceId

destinationId

name

schedule

status

json
{
  "name": "My Connection",
  "sourceId": "source_id",
  "destinationId": "destination_id",
  "syncCatalog": {
    "streams": [
      {
        "stream": {
          "name": "my_stream",
          "namespace": "public"
        },
        "config": {
          "syncMode": "full_refresh",
          "destinationSyncMode": "overwrite"
        }
      }
    ]
  },
  "schedule": {
    "units": 1,
    "timeUnit": "hours"
  },
  "status": "active"
}

Could not find configuration for SOURCE_CONNECTION: 8ae47576-ecb7-430c-9f52-a86b40a33f33

sourceId

destinationId

sourceId

destinationId

python # Step 1: Create Source try:     connectionConfiguration = client.get_source_configuration()     source_id = client.create_source(         client.name, client.workspaceId, client.sourceDefinitionId, connectionConfiguration     )     print(f"Created source with ID: {source_id}")      # Step 2: Create Destination     connectionConfiguration = client.get_destination_configuration()     destination_id = client.create_destination(         client.destination_name, client.workspaceId, client.destinationDefinitionId, connectionConfiguration     )     print(f"Created destination with ID: {destination_id}")      # Step 3: Create Connection     connection_id = client.create_connection(source_id, destination_id, client.workspaceId)     print(f"Created connection with ID: {connection_id}")  except Exception as e:     print(f"Error: {e}")

plaintext      <http://localhost:8000/api/v1/connections/create>

sourceId

destinationId

name

schedule

status

json {   "name": "My Connection",   "sourceId": "source_id",   "destinationId": "destination_id",   "syncCatalog": {     "streams": [       {         "stream": {           "name": "my_stream",           "namespace": "public"         },         "config": {           "syncMode": "full_refresh",           "destinationSyncMode": "overwrite"         }       }     ]   },   "schedule": {     "units": 1,     "timeUnit": "hours"   },   "status": "active" }

python
import time

# Step 1: Create Source
try:
    connectionConfiguration = client.get_source_configuration()
    source_id = client.create_source(
        client.name, client.workspaceId, client.sourceDefinitionId, connectionConfiguration
    )
    print(f"Created source with ID: {source_id}")

    # Add a delay to allow the source to be fully registered
    time.sleep(10)  # Delay for 10 seconds

    # Step 2: Create Destination
    connectionConfiguration = client.get_destination_configuration()
    destination_id = client.create_destination(
        client.destination_name, client.workspaceId, client.destinationDefinitionId, connectionConfiguration
    )
    print(f"Created destination with ID: {destination_id}")

    # Add a delay to allow the destination to be fully registered
    time.sleep(10)  # Delay for 10 seconds

    # Step 3: Create Connection
    connection_id = client.create_connection(source_id, destination_id, client.workspaceId)
    print(f"Created connection with ID: {connection_id}")

except Exception as e:
    print(f"Error: {e}")

time.sleep(10)

plaintext
     <http://localhost:8000/api/v1/connections/create>

sourceId

destinationId

name

schedule

status

json
{
  "name": "My Connection",
  "sourceId": "source_id",
  "destinationId": "destination_id",
  "syncCatalog": {
    "streams": [
      {
        "stream": {
          "name": "my_stream",
          "namespace": "public"
        },
        "config": {
          "syncMode": "full_refresh",
          "destinationSyncMode": "overwrite"
        }
      }
    ]
  },
  "schedule": {
    "units": 1,
    "timeUnit": "hours"
  },
  "status": "active"
}

python import time  # Step 1: Create Source try:     connectionConfiguration = client.get_source_configuration()     source_id = client.create_source(         client.name, client.workspaceId, client.sourceDefinitionId, connectionConfiguration     )     print(f"Created source with ID: {source_id}")      # Add a delay to allow the source to be fully registered     time.sleep(10)  # Delay for 10 seconds      # Step 2: Create Destination     connectionConfiguration = client.get_destination_configuration()     destination_id = client.create_destination(         client.destination_name, client.workspaceId, client.destinationDefinitionId, connectionConfiguration     )     print(f"Created destination with ID: {destination_id}")      # Add a delay to allow the destination to be fully registered     time.sleep(10)  # Delay for 10 seconds      # Step 3: Create Connection     connection_id = client.create_connection(source_id, destination_id, client.workspaceId)     print(f"Created connection with ID: {connection_id}")  except Exception as e:     print(f"Error: {e}")

time.sleep(10)

plaintext      <http://localhost:8000/api/v1/connections/create>

sourceId

destinationId

name

schedule

status

json {   "name": "My Connection",   "sourceId": "source_id",   "destinationId": "destination_id",   "syncCatalog": {     "streams": [       {         "stream": {           "name": "my_stream",           "namespace": "public"         },         "config": {           "syncMode": "full_refresh",           "destinationSyncMode": "overwrite"         }       }     ]   },   "schedule": {     "units": 1,     "timeUnit": "hours"   },   "status": "active" }

json {   "name": "Dr_Chrono --> Local_JSON",   "workspaceId": "64cf55a5-7a9c-4a15-8928-e3d2eddfaeb7",   "destinationId": "362673f3-d948-40b3-b48f-b8896ee8ca4e",   "sourceId": "14fa1639-c5f6-4181-8faf-505b54086069",   "namespaceDefinition": "source",   "prefix": "",   "operationIds": [],   "syncCatalog": {     "streams": [       {         "stream": {           "name": "patients",           "jsonSchema": {             "type": "object",             "$schema": "<http://json-schema.org/schema#>"           }         },         "config": {           "syncMode": "full_refresh",           "destinationSyncMode": "overwrite"         }       }     ]   },   "scheduleType": "manual",   "status": "active",   "sourceCatalogId": "ad0fff60-aa0d-467d-a0dd-61b702c27557",   "geography": "auto",   "breakingChange": false,   "notifySchemaChanges": true,   "notifySchemaChangesByEmail": false,   "nonBreakingChangesPreference": "propagate_columns",   "created_at": 1718869706,   "backfillPreference": "disabled" }

scheduleType

status

sourceCatalogId

syncCatalog

config

streams

POST

plaintext      <http://localhost:8000/api/v1/connections/create>

Content-Type

Key: Content-Type      Value: application/json

Internal Server Error: getSelected(...) must not be null

syncCatalog

selected

syncCatalog

syncCatalog

syncCatalog

selected

selected

syncCatalog

selected

json {   "name": "Dr_Chrono --> Local_JSON",   "workspaceId": "64cf55a5-7a9c-4a15-8928-e3d2eddfaeb7",   "destinationId": "362673f3-d948-40b3-b48f-b8896ee8ca4e",   "sourceId": "14fa1639-c5f6-4181-8faf-505b54086069",   "namespaceDefinition": "source",   "prefix": "",   "operationIds": [],   "syncCatalog": {     "streams": [       {         "stream": {           "name": "patients",           "jsonSchema": {             "type": "object",             "$schema": "<http://json-schema.org/schema#>"           }         },         "config": {           "syncMode": "full_refresh",           "destinationSyncMode": "overwrite",           "selected": true         }       }     ]   },   "scheduleType": "manual",   "status": "active",   "sourceCatalogId": "ad0fff60-aa0d-467d-a0dd-61b702c27557",   "geography": "auto",   "breakingChange": false,   "notifySchemaChanges": true,   "notifySchemaChangesByEmail": false,   "nonBreakingChangesPreference": "propagate_columns",   "created_at": 1718869706,   "backfillPreference": "disabled" }

selected

selected

config

syncCatalog

POST

plaintext      <http://localhost:8000/api/v1/connections/create>

Content-Type

Key: Content-Type      Value: application/json

400 Bad Request

json {   "name": "Dr_Chrono --> Local_JSON",   "workspaceId": "64cf55a5-7a9c-4a15-8928-e3d2eddfaeb7",   "destinationId": "362673f3-d948-40b3-b48f-b8896ee8ca4e",   "sourceId": "14fa1639-c5f6-4181-8faf-505b54086069",   "namespaceDefinition": "source",   "prefix": "",   "operationIds": [],   "syncCatalog": {     "streams": [       {         "stream": {           "name": "patients",           "jsonSchema": {             "type": "object",             "$schema": "<http://json-schema.org/schema#>"           }         },         "config": {           "syncMode": "full_refresh",           "destinationSyncMode": "overwrite",           "selected": true         }       }     ]   },   "scheduleType": "manual",   "status": "active",   "sourceCatalogId": "ad0fff60-aa0d-467d-a0dd-61b702c27557",   "geography": "auto",   "breakingChange": false,   "notifySchemaChanges": true,   "notifySchemaChangesByEmail": false,   "nonBreakingChangesPreference": "propagate_columns",   "created_at": 1718869706,   "backfillPreference": "disabled" }

selected

selected

config

syncCatalog

POST

plaintext      <http://localhost:8000/api/v1/connections/create>

Content-Type

Key: Content-Type      Value: application/json

scheduleType

json {   "name": "Dr_Chrono --> Local_JSON",   "workspaceId": "64cf55a5-7a9c-4a15-8928-e3d2eddfaeb7",   "destinationId": "362673f3-d948-40b3-b48f-b8896ee8ca4e",   "sourceId": "14fa1639-c5f6-4181-8faf-505b54086069",   "namespaceDefinition": "source",   "prefix": "",   "operationIds": [],   "syncCatalog": {     "streams": [       {         "stream": {           "name": "patients",           "jsonSchema": {             "type": "object",             "$schema": "<http://json-schema.org/schema#>"           }         },         "config": {           "syncMode": "full_refresh",           "destinationSyncMode": "overwrite",           "selected": true         }       }     ]   },   "status": "active",   "sourceCatalogId": "ad0fff60-aa0d-467d-a0dd-61b702c27557",   "geography": "auto",   "breakingChange": false,   "notifySchemaChanges": true,   "notifySchemaChangesByEmail": false,   "nonBreakingChangesPreference": "propagate_columns",   "created_at": 1718869706,   "backfillPreference": "disabled" }

source

destination

active

json {   "name": "Dr_Chrono --> Local_JSON",   "workspaceId": "64cf55a5-7a9c-4a15-8928-e3d2eddfaeb7",   "sourceId": "14fa1639-c5f6-4181-8faf-505b54086069",   "destinationId": "362673f3-d948-40b3-b48f-b8896ee8ca4e",   "namespaceDefinition": "source",   "prefix": "",   "operationIds": [],   "syncCatalog": {     "streams": [       {         "stream": {           "name": "patients",           "jsonSchema": {             "type": "object",             "$schema": "<http://json-schema.org/schema#>"           }         },         "config": {           "syncMode": "full_refresh",           "destinationSyncMode": "overwrite",           "selected": true         }       }     ]   },   "status": "active" }

POST

plaintext      <http://localhost:8000/api/v1/connections/create>

Content-Type

Key: Content-Type      Value: application/json

source

destination

active

json
{
  "name": "Dr_Chrono --> Local_JSON",
  "workspaceId": "64cf55a5-7a9c-4a15-8928-e3d2eddfaeb7",
  "sourceId": "14fa1639-c5f6-4181-8faf-505b54086069",
  "destinationId": "362673f3-d948-40b3-b48f-b8896ee8ca4e",
  "namespaceDefinition": "source",
  "prefix": "",
  "operationIds": [],
  "syncCatalog": {
    "streams": [
      {
        "stream": {
          "name": "patients",
          "jsonSchema": {
            "type": "object",
            "$schema": "<http://json-schema.org/schema#>"
          }
        },
        "config": {
          "syncMode": "full_refresh",
          "destinationSyncMode": "overwrite",
          "selected": true
        }
      }
    ]
  },
  "status": "active"
}

source

destination

active

json {   "name": "Dr_Chrono --> Local_JSON",   "workspaceId": "64cf55a5-7a9c-4a15-8928-e3d2eddfaeb7",   "sourceId": "14fa1639-c5f6-4181-8faf-505b54086069",   "destinationId": "362673f3-d948-40b3-b48f-b8896ee8ca4e",   "namespaceDefinition": "source",   "prefix": "",   "operationIds": [],   "syncCatalog": {     "streams": [       {         "stream": {           "name": "patients",           "jsonSchema": {             "type": "object",             "$schema": "<http://json-schema.org/schema#>"           }         },         "config": {           "syncMode": "full_refresh",           "destinationSyncMode": "overwrite",           "selected": true         }       }     ]   },   "status": "active" }

Internal Server Error: Cannot invoke "com.fasterxml.jackson.databind.JsonNode.fieldNames()" because the return value of "com.fasterxml.jackson.databind.JsonNode.get(String)" is null

syncCatalog

jsonSchema

jsonSchema

jsonSchema

jsonSchema

syncCatalog

syncCatalog

jsonSchema

syncCatalog

json {   "name": "Dr_Chrono --> Local_JSON",   "workspaceId": "64cf55a5-7a9c-4a15-8928-e3d2eddfaeb7",   "sourceId": "14fa1639-c5f6-4181-8faf-505b54086069",   "destinationId": "362673f3-d948-40b3-b48f-b8896ee8ca4e",   "namespaceDefinition": "source",   "prefix": "",   "operationIds": [],   "syncCatalog": {     "streams": [       {         "stream": {           "name": "patients",           "jsonSchema": {             "type": "object",             "$schema": "<http://json-schema.org/schema#>",             "properties": {               "id": { "type": "string" },               "name": { "type": "string" },               "age": { "type": "integer" }             }           }         },         "config": {           "syncMode": "full_refresh",           "destinationSyncMode": "overwrite",           "selected": true         }       }     ]   },   "status": "active" }

properties

jsonSchema

properties

jsonSchema

id

name

age

POST

plaintext      <http://localhost:8000/api/v1/connections/create>

Content-Type

Key: Content-Type      Value: application/json

syncCatalog

namespaceDefinition

source

destination

manual

active

json
{
  "name": "Dr_Chrono --> Local_JSON",
  "workspaceId": "64cf55a5-7a9c-4a15-8928-e3d2eddfaeb7",
  "destinationId": "362673f3-d948-40b3-b48f-b8896ee8ca4e",
  "sourceId": "14fa1639-c5f6-4181-8faf-505b54086069",
  "namespaceDefinition": "source",
  "syncCatalog": {
    "streams": [
      {
        "stream": {
          "name": "patients",
          "jsonSchema": {
            "type": "object",
            "properties": {
              "id": { "type": "number" },
              "city": { "type": "string" },
              "race": { "type": "string" },
              "email": { "type": "string" },
              "state": { "type": "string" },
              "gender": { "type": "string" },
              "address": { "type": "string" },
              "zip_code": { "type": "string" },
              "last_name": { "type": "string" },
              "first_name": { "type": "string" },
              "date_of_birth": { "type": "string" }
            }
          },
          "supportedSyncModes": ["full_refresh"]
        },
        "config": {
          "syncMode": "full_refresh",
          "destinationSyncMode": "overwrite",
          "selected": true
        }
      }
    ]
  },
  "scheduleType": "manual",
  "status": "active"
}

syncCatalog

namespaceDefinition

source

destination

manual

active

json {   "name": "Dr_Chrono --> Local_JSON",   "workspaceId": "64cf55a5-7a9c-4a15-8928-e3d2eddfaeb7",   "destinationId": "362673f3-d948-40b3-b48f-b8896ee8ca4e",   "sourceId": "14fa1639-c5f6-4181-8faf-505b54086069",   "namespaceDefinition": "source",   "syncCatalog": {     "streams": [       {         "stream": {           "name": "patients",           "jsonSchema": {             "type": "object",             "properties": {               "id": { "type": "number" },               "city": { "type": "string" },               "race": { "type": "string" },               "email": { "type": "string" },               "state": { "type": "string" },               "gender": { "type": "string" },               "address": { "type": "string" },               "zip_code": { "type": "string" },               "last_name": { "type": "string" },               "first_name": { "type": "string" },               "date_of_birth": { "type": "string" }             }           },           "supportedSyncModes": ["full_refresh"]         },         "config": {           "syncMode": "full_refresh",           "destinationSyncMode": "overwrite",           "selected": true         }       }     ]   },   "scheduleType": "manual",   "status": "active" }

sourceId

workspaceId

plaintext <http://localhost:8000/api/v1/sources/update>

json {   "sourceId": "e7066349-ea5b-4ec5-99ee-1a936c462c62",   "workspaceId": "64cf55a5-7a9c-4a15-8928-e3d2eddfaeb7",   "name": "Updated Source Name",   "connectionConfiguration": {     "base": "USD",     "access_key": "new_access_key",     "start_date": "2021-01-01",     "ignore_weekends": true   } }

POST

plaintext      <http://localhost:8000/api/v1/sources/update>

Content-Type

Key: Content-Type      Value: application/json

bash curl -X POST "<http://localhost:8000/api/v1/sources/update>" \      -H "Content-Type: application/json" \      -d '{            "sourceId": "e7066349-ea5b-4ec5-99ee-1a936c462c62",            "workspaceId": "64cf55a5-7a9c-4a15-8928-e3d2eddfaeb7",            "name": "Updated Source Name",            "connectionConfiguration": {              "base": "USD",              "access_key": "new_access_key",              "start_date": "2021-01-01",              "ignore_weekends": true            }          }'

connectionId

plaintext <http://localhost:8000/api/v1/connections/sync>

json {{   "connectionId": "your_connection_id" }}

POST

plaintext      <http://localhost:8000/api/v1/connections/sync>

Content-Type

Key: Content-Type      Value: application/json

"your_connection_id"

connectionId

bash curl -X POST "<http://localhost:8000/api/v1/connections/sync>" \      -H "Content-Type: application/json" \      -d '{{            "connectionId": "your_connection_id"          }}'

python from airflow import DAG from airflow.operators.python_operator import PythonOperator from datetime import datetime import requests import json  def start_airbyte_sync(connection_id):     url = "<http://localhost:8000/api/v1/connections/sync>"     payload = {{         "connectionId": connection_id     }}     headers = {{         "Content-Type": "application/json"     }}     response = <http://requests.post|requests.post>(url, json=payload, headers=headers)     if response.status_code == 200:         print(f"Sync started successfully for connection ID: {{connection_id}}")     else:         raise Exception(f"Failed to start sync: {{response.content}}")  default_args = {{     'owner': 'airflow',     'start_date': datetime(2023, 1, 1),     'retries': 1, }}  dag = DAG(     'airbyte_sync_dag',     default_args=default_args,     description='A simple Airbyte sync DAG',     schedule_interval='@daily', )  start_sync_task = PythonOperator(     task_id='start_airbyte_sync',     python_callable=start_airbyte_sync,     op_kwargs={{'connection_id': 'your_connection_id'}},     dag=dag, )

'your_connection_id'

connectionId

plaintext DELETE /connections/{connectionId}

DELETE

{connectionId}

plaintext      <http://localhost:8000/api/v1/connections/{connectionId}>

Content-Type

Key: Content-Type      Value: application/json

bash curl -X DELETE "<http://localhost:8000/api/v1/connections/{connectionId}>" \      -H "Content-Type: application/json"

{connectionId}

sourceId

plaintext DELETE /sources/{{sourceId}}

DELETE

{{sourceId}}

plaintext      <http://localhost:8000/api/v1/sources/{{sourceId}}>

Content-Type

Key: Content-Type      Value: application/json

bash curl -X DELETE "<http://localhost:8000/api/v1/sources/{{sourceId}}>" \      -H "Content-Type: application/json"

{{sourceId}}

connectionId

plaintext DELETE /connections/{{connectionId}}

DELETE

{{connectionId}}

plaintext      <http://localhost:8000/api/v1/connections/{{connectionId}}>

Content-Type

Key: Content-Type      Value: application/json

bash curl -X DELETE "<http://localhost:8000/api/v1/connections/{{connectionId}}>" \      -H "Content-Type: application/json"

{{connectionId}}

204 No Content

200 OK

204 No Content

204 No Content

200 OK

python import requests from requests.auth import HTTPBasicAuth  class AirbyteApiClient:     def __init__(self, base_url, username, password):         self.base_url = base_url         self.username = username         self.password = password      def delete_connection(self, connection_id) -> None:         """Delete a connection in the Airbyte API."""         url = f"{self.base_url}/connections/delete"         payload = {             "connectionId": connection_id,         }         auth = HTTPBasicAuth(self.username, self.password)         response = <http://requests.post|requests.post>(url, json=payload, auth=auth)         if response.status_code == 204:             print(f"Connection {connection_id} deleted successfully")         else:             raise Exception(                 f"Failed to delete connection {connection_id}: {response.content}"             )          return None

204 No Content

204

python # Initialize the Airbyte API client client = AirbyteApiClient(base_url="<http://localhost:8000/api/v1>", username="your_username", password="your_password")  # Delete a connection try:     client.delete_connection("d7b9c70d-9132-49c1-b5b6-28a31447f792") except Exception as e:     print(f"Error: {e}")

204 No Content

AirbyteApiClient

connection_id

connection_id

sync_data

connection_id

connection_id

sync_data

sync_data

sync_data

connection_id

AirbyteApiClient

sync_data

airbyte_manager.py

python
import requests
from requests.auth import HTTPBasicAuth

class AirbyteApiClient:
    def __init__(self, base_url, username, password):
        self.base_url = base_url
        self.username = username
        self.password = password

    def sync_data(self, connection_id) -> None:
        """Trigger a sync for the given connection ID in the Airbyte API."""
        url = f"{self.base_url}/connections/sync"
        payload = {
            "connectionId": connection_id
        }
        auth = HTTPBasicAuth(self.username, self.password)
        response = <http://requests.post|requests.post>(url, json=payload, auth=auth)
        if response.status_code == 200:
            print(f"Sync started successfully for connection ID: {connection_id}")
        else:
            raise Exception(f"Failed to start sync: {response.content}")

        return None

airbyte_integration_dag.py

connection_id

sync_data

python
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from airflow.exceptions import AirflowException
from datetime import datetime
from access_afya_elt.common_package.api_clients.airbyte_manager import AirbyteApiClient

def extract_patient_data(connection_id):
    airbyte_client = AirbyteApiClient(base_url="<http://localhost:8000/api/v1>", username="your_username", password="your_password")
    try:
        airbyte_client.sync_data(connection_id=connection_id)
    except Exception as e:
        raise AirflowException(f"Data pipeline task failed: {e}") from e

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2023, 1, 1),
    'retries': 1,
}

dag = DAG(
    'patient_data_extraction_dag',
    default_args=default_args,
    description='A DAG to extract patient data using Airbyte',
    schedule_interval='@daily',
)

extract_patient_data_task = PythonOperator(
    task_id='extract_patient_data',
    python_callable=extract_patient_data,
    op_kwargs={'connection_id': 'your_connection_id'},
    dag=dag,
)

sync_data

connection_id

connection_id

connection_id

sync_data

AirbyteApiClient

connection_id

connection_id

sync_data

connection_id

connection_id

sync_data

sync_data

sync_data

connection_id

AirbyteApiClient

sync_data

airbyte_manager.py

python import requests from requests.auth import HTTPBasicAuth  class AirbyteApiClient:     def __init__(self, base_url, username, password):         self.base_url = base_url         self.username = username         self.password = password      def sync_data(self, connection_id) -> None:         """Trigger a sync for the given connection ID in the Airbyte API."""         url = f"{self.base_url}/connections/sync"         payload = {             "connectionId": connection_id         }         auth = HTTPBasicAuth(self.username, self.password)         response = <http://requests.post|requests.post>(url, json=payload, auth=auth)         if response.status_code == 200:             print(f"Sync started successfully for connection ID: {connection_id}")         else:             raise Exception(f"Failed to start sync: {response.content}")          return None

airbyte_integration_dag.py

connection_id

sync_data

python from airflow import DAG from airflow.operators.python_operator import PythonOperator from airflow.exceptions import AirflowException from datetime import datetime from access_afya_elt.common_package.api_clients.airbyte_manager import AirbyteApiClient  def extract_patient_data(connection_id):     airbyte_client = AirbyteApiClient(base_url="<http://localhost:8000/api/v1>", username="your_username", password="your_password")     try:         airbyte_client.sync_data(connection_id=connection_id)     except Exception as e:         raise AirflowException(f"Data pipeline task failed: {e}") from e  default_args = {     'owner': 'airflow',     'start_date': datetime(2023, 1, 1),     'retries': 1, }  dag = DAG(     'patient_data_extraction_dag',     default_args=default_args,     description='A DAG to extract patient data using Airbyte',     schedule_interval='@daily', )  extract_patient_data_task = PythonOperator(     task_id='extract_patient_data',     python_callable=extract_patient_data,     op_kwargs={'connection_id': 'your_connection_id'},     dag=dag, )

sync_data

connection_id

connection_id

connection_id

sync_data

self.connection_id

connection_id

sync_data

python import requests from requests.auth import HTTPBasicAuth  class AirbyteApiClient:     def __init__(self, base_url, username, password):         self.base_url = base_url         self.username = username         self.password = password      def sync_data(self, connection_id: str) -> None:         """Trigger a sync for the given connection in the Airbyte API."""         url = f"{self.base_url}/connections/sync"         auth = HTTPBasicAuth(self.username, self.password)         response = <http://requests.post|requests.post>(             url, auth=auth, json={"connectionId": connection_id}         )         if response.status_code == 200:             print(f"Sync triggered successfully for connection {connection_id}")         else:             raise Exception(                 "Failed to trigger sync for connection "                 + f"{connection_id}: {response.content}"             )          return None

connection_id

self.connection_id

<http://requests.post|requests.post>

python from airflow import DAG from airflow.operators.python_operator import PythonOperator from airflow.exceptions import AirflowException from datetime import datetime from access_afya_elt.common_package.api_clients.airbyte_manager import AirbyteApiClient  def extract_patient_data(connection_id):     airbyte_client = AirbyteApiClient(base_url="<http://localhost:8000/api/v1>", username="your_username", password="your_password")     try:         airbyte_client.sync_data(connection_id=connection_id)     except Exception as e:         raise AirflowException(f"Data pipeline task failed: {e}") from e  default_args = {     'owner': 'airflow',     'start_date': datetime(2023, 1, 1),     'retries': 1, }  dag = DAG(     'patient_data_extraction_dag',     default_args=default_args,     description='A DAG to extract patient data using Airbyte',     schedule_interval='@daily', )  extract_patient_data_task = PythonOperator(     task_id='extract_patient_data',     python_callable=extract_patient_data,     op_kwargs={'connection_id': 'your_connection_id'},     dag=dag, )

connection_id

sync_data

Internal Server Error: null

java.lang.NullPointerException

sourceId

destinationId

sourceId

destinationId

syncCatalog

syncCatalog

json {   "name": "Dr_Chrono --> Local_JSON",   "workspaceId": "64cf55a5-7a9c-4a15-8928-e3d2eddfaeb7",   "sourceId": "14fa1639-c5f6-4181-8faf-505b54086069",   "destinationId": "362673f3-d948-40b3-b48f-b8896ee8ca4e",   "namespaceDefinition": "source",   "prefix": "",   "operationIds": [],   "syncCatalog": {     "streams": [       {         "stream": {           "name": "patients",           "jsonSchema": {             "type": "object",             "$schema": "<http://json-schema.org/schema#>",             "properties": {               "id": { "type": "number" },               "city": { "type": "string" },               "race": { "type": "string" },               "email": { "type": "string" },               "state": { "type": "string" },               "gender": { "type": "string" },               "address": { "type": "string" },               "zip_code": { "type": "string" },               "last_name": { "type": "string" },               "first_name": { "type": "string" },               "date_of_birth": { "type": "string" }             }           },           "supportedSyncModes": ["full_refresh"]         },         "config": {           "syncMode": "full_refresh",           "destinationSyncMode": "overwrite",           "selected": true         }       }     ]   },   "status": "active" }

POST

plaintext      <http://localhost:8000/api/v1/connections/create>

Content-Type

Key: Content-Type      Value: application/json

Request to <https://drchrono.com/api/patients?since=2024-06-25&page_size=250> failed with status code 401 and error message None

401

access_key

json {{   "sourceId": "your_source_id",   "workspaceId": "your_workspace_id",   "name": "DrChrono Source",   "connectionConfiguration": {{     "base": "USD",     "access_key": "your_api_key",     "start_date": "2021-01-01",     "ignore_weekends": true   }} }}

GET

plaintext      <https://drchrono.com/api/patients?since=2024-06-25&page_size=250>

Authorization

Key: Authorization      Value: Bearer your_api_key

bash curl -X GET "<https://drchrono.com/api/patients?since=2024-06-25&page_size=250>" \      -H "Authorization: Bearer your_api_key"

python import requests from requests.auth import HTTPBasicAuth  class AirbyteApiClient:     def __init__(self, base_url, username, password):         self.base_url = base_url         self.auth = HTTPBasicAuth(username, password)      def list_sources(self):         url = f"{self.base_url}/sources/list"         response = <http://requests.post|requests.post>(url, auth=self.auth)         if response.status_code == 200:             return response.json().get('sources', [])         else:             raise Exception(f"Failed to list sources: {response.content}")      def delete_source(self, source_id):         url = f"{self.base_url}/sources/delete"         payload = {"sourceId": source_id}         response = <http://requests.post|requests.post>(url, json=payload, auth=self.auth)         if response.status_code == 204:             print(f"Source {source_id} deleted successfully")         else:             raise Exception(f"Failed to delete source {source_id}: {response.content}")      def delete_all_sources(self):         sources = self.list_sources()         for source in sources:             self.delete_source(source['sourceId'])  # Initialize the Airbyte API client client = AirbyteApiClient(base_url="<http://localhost:8000/api/v1>", username="your_username", password="your_password")  # Delete all sources try:     client.delete_all_sources() except Exception as e:     print(f"Error: {e}")

AirbyteApiClient

list_sources

/sources/list

delete_source

/sources/delete

delete_all_sources

"your_username"

"your_password"

python
import requests
from requests.auth import HTTPBasicAuth

class AirbyteApiClient:
    def __init__(self, base_url, username, password):
        self.base_url = base_url
        self.auth = HTTPBasicAuth(username, password)

    def list_sources(self):
        url = f"{{self.base_url}}/sources/list"
        response = <http://requests.post|requests.post>(url, auth=self.auth)
        if response.status_code == 200:
            return response.json().get('sources', [])
        else:
            raise Exception(f"Failed to list sources: {{response.content}}")

    def delete_source(self, source_id):
        url = f"{{self.base_url}}/sources/delete"
        payload = {{"sourceId": source_id}}
        response = <http://requests.post|requests.post>(url, json=payload, auth=self.auth)
        if response.status_code == 204:
            print(f"Source {{source_id}} deleted successfully")
        else:
            raise Exception(f"Failed to delete source {{source_id}}: {{response.content}}")

    def delete_percentage_of_sources(self, percentage):
        sources = self.list_sources()
        total_sources = len(sources)
        num_to_delete = int(total_sources * (percentage / 100))
        sources_to_delete = sources[:num_to_delete]
        
        for source in sources_to_delete:
            self.delete_source(source['sourceId'])

# Initialize the Airbyte API client
client = AirbyteApiClient(base_url="<http://localhost:8000/api/v1>", username="your_username", password="your_password")

# Delete 98% of sources
try:
    client.delete_percentage_of_sources(98)
except Exception as e:
    print(f"Error: {{e}}")

AirbyteApiClient

list_sources

/sources/list

delete_percentage_of_sources

python import requests from requests.auth import HTTPBasicAuth  class AirbyteApiClient:     def __init__(self, base_url, username, password):         self.base_url = base_url         self.auth = HTTPBasicAuth(username, password)      def list_sources(self):         url = f"{{self.base_url}}/sources/list"         response = <http://requests.post|requests.post>(url, auth=self.auth)         if response.status_code == 200:             return response.json().get('sources', [])         else:             raise Exception(f"Failed to list sources: {{response.content}}")      def delete_source(self, source_id):         url = f"{{self.base_url}}/sources/delete"         payload = {{"sourceId": source_id}}         response = <http://requests.post|requests.post>(url, json=payload, auth=self.auth)         if response.status_code == 204:             print(f"Source {{source_id}} deleted successfully")         else:             raise Exception(f"Failed to delete source {{source_id}}: {{response.content}}")      def delete_percentage_of_sources(self, percentage):         sources = self.list_sources()         total_sources = len(sources)         num_to_delete = int(total_sources * (percentage / 100))         sources_to_delete = sources[:num_to_delete]                  for source in sources_to_delete:             self.delete_source(source['sourceId'])  # Initialize the Airbyte API client client = AirbyteApiClient(base_url="<http://localhost:8000/api/v1>", username="your_username", password="your_password")  # Delete 98% of sources try:     client.delete_percentage_of_sources(98) except Exception as e:     print(f"Error: {{e}}")

AirbyteApiClient

list_sources

/sources/list

delete_percentage_of_sources

delete_source

/sources/delete

delete_percentage_of_sources

"your_username"

"your_password"

python
import requests
from requests.auth import HTTPBasicAuth

class AirbyteApiClient:
    def __init__(self, base_url, username, password):
        self.base_url = base_url
        self.auth = HTTPBasicAuth(username, password)

    def list_sources(self):
        url = f"{self.base_url}/sources/list"
        response = <http://requests.post|requests.post>(url, auth=self.auth)
        if response.status_code == 200:
            return response.json().get('sources', [])
        else:
            raise Exception(f"Failed to list sources: {response.content}")

    def delete_source(self, source_id):
        url = f"{self.base_url}/sources/delete"
        payload = {"sourceId": source_id}
        response = <http://requests.post|requests.post>(url, json=payload, auth=self.auth)
        if response.status_code == 204:
            print(f"Source {source_id} deleted successfully")
        else:
            raise Exception(f"Failed to delete source {source_id}: {response.content}")

    def delete_percentage_of_sources(self, percentage):
        sources = self.list_sources()
        total_sources = len(sources)
        num_to_delete = int(total_sources * (percentage / 100))
        sources_to_delete = sources[:num_to_delete]
        
        for source in sources_to_delete:
            self.delete_source(source['sourceId'])

# Initialize the Airbyte API client
client = AirbyteApiClient(base_url="<http://localhost:8000/api/v1>", username="your_username", password="your_password")

# Delete 98% of sources
try:
    client.delete_percentage_of_sources(98)
except Exception as e:
    print(f"Error: {e}")

AirbyteApiClient

list_sources

/sources/list

delete_percentage_of_sources

python import requests from requests.auth import HTTPBasicAuth  class AirbyteApiClient:     def __init__(self, base_url, username, password):         self.base_url = base_url         self.auth = HTTPBasicAuth(username, password)      def list_sources(self):         url = f"{self.base_url}/sources/list"         response = <http://requests.post|requests.post>(url, auth=self.auth)         if response.status_code == 200:             return response.json().get('sources', [])         else:             raise Exception(f"Failed to list sources: {response.content}")      def delete_source(self, source_id):         url = f"{self.base_url}/sources/delete"         payload = {"sourceId": source_id}         response = <http://requests.post|requests.post>(url, json=payload, auth=self.auth)         if response.status_code == 204:             print(f"Source {source_id} deleted successfully")         else:             raise Exception(f"Failed to delete source {source_id}: {response.content}")      def delete_percentage_of_sources(self, percentage):         sources = self.list_sources()         total_sources = len(sources)         num_to_delete = int(total_sources * (percentage / 100))         sources_to_delete = sources[:num_to_delete]                  for source in sources_to_delete:             self.delete_source(source['sourceId'])  # Initialize the Airbyte API client client = AirbyteApiClient(base_url="<http://localhost:8000/api/v1>", username="your_username", password="your_password")  # Delete 98% of sources try:     client.delete_percentage_of_sources(98) except Exception as e:     print(f"Error: {e}")

AirbyteApiClient

list_sources

/sources/list

delete_percentage_of_sources

delete_source

/sources/delete

delete_percentage_of_sources

"your_username"

"your_password"

Internal Server Error: That secret was not found in the store!

plaintext
     airbyte_workspace_64cf55a5-7a9c-4a15-8928-e3d2eddfaeb7_secret_4d27c364-4ad9-402c-a187-4c4a24226663_v3

python
import requests
from requests.auth import HTTPBasicAuth

class AirbyteApiClient:
    def __init__(self, base_url, username, password):
        self.base_url = base_url
        self.auth = HTTPBasicAuth(username, password)

    def list_sources(self):
        url = f"{self.base_url}/sources/list"
        response = <http://requests.post|requests.post>(url, auth=self.auth)
        if response.status_code == 200:
            return response.json().get('sources', [])
        else:
            raise Exception(f"Failed to list sources: {response.content}")

    def delete_source(self, source_id):
        url = f"{self.base_url}/sources/delete"
        payload = {"sourceId": source_id}
        response = <http://requests.post|requests.post>(url, json=payload, auth=self.auth)
        if response.status_code == 204:
            print(f"Source {source_id} deleted successfully")
        else:
            raise Exception(f"Failed to delete source {source_id}: {response.content}")

    def delete_percentage_of_sources(self, percentage):
        sources = self.list_sources()
        total_sources = len(sources)
        num_to_delete = int(total_sources * (percentage / 100))
        sources_to_delete = sources[:num_to_delete]
        
        for source in sources_to_delete:
            try:
                self.delete_source(source['sourceId'])
            except Exception as e:
                print(f"Error deleting source {source['sourceId']}: {e}")

# Initialize the Airbyte API client
client = AirbyteApiClient(base_url="<http://localhost:8000/api/v1>", username="your_username", password="your_password")

# Delete 98% of sources
try:
    client.delete_percentage_of_sources(98)
except Exception as e:
    print(f"Error: {e}")

Internal Server Error: That secret was not found in the store!

plaintext      airbyte_workspace_64cf55a5-7a9c-4a15-8928-e3d2eddfaeb7_secret_4d27c364-4ad9-402c-a187-4c4a24226663_v3

python import requests from requests.auth import HTTPBasicAuth  class AirbyteApiClient:     def __init__(self, base_url, username, password):         self.base_url = base_url         self.auth = HTTPBasicAuth(username, password)      def list_sources(self):         url = f"{self.base_url}/sources/list"         response = <http://requests.post|requests.post>(url, auth=self.auth)         if response.status_code == 200:             return response.json().get('sources', [])         else:             raise Exception(f"Failed to list sources: {response.content}")      def delete_source(self, source_id):         url = f"{self.base_url}/sources/delete"         payload = {"sourceId": source_id}         response = <http://requests.post|requests.post>(url, json=payload, auth=self.auth)         if response.status_code == 204:             print(f"Source {source_id} deleted successfully")         else:             raise Exception(f"Failed to delete source {source_id}: {response.content}")      def delete_percentage_of_sources(self, percentage):         sources = self.list_sources()         total_sources = len(sources)         num_to_delete = int(total_sources * (percentage / 100))         sources_to_delete = sources[:num_to_delete]                  for source in sources_to_delete:             try:                 self.delete_source(source['sourceId'])             except Exception as e:                 print(f"Error deleting source {source['sourceId']}: {e}")  # Initialize the Airbyte API client client = AirbyteApiClient(base_url="<http://localhost:8000/api/v1>", username="your_username", password="your_password")  # Delete 98% of sources try:     client.delete_percentage_of_sources(98) except Exception as e:     print(f"Error: {e}")

python import requests from requests.auth import HTTPBasicAuth  class AirbyteApiClient:     def __init__(self, base_url, username, password):         self.base_url = base_url         self.auth = HTTPBasicAuth(username, password)      def list_sources(self):         url = f"{{self.base_url}}/sources/list"         response = <http://requests.post|requests.post>(url, auth=self.auth)         if response.status_code == 200:             return response.json().get('sources', [])         else:             raise Exception(f"Failed to list sources: {{response.content}}")      def delete_source(self, source_id):         url = f"{{self.base_url}}/sources/delete"         payload = {{"sourceId": source_id}}         response = <http://requests.post|requests.post>(url, json=payload, auth=self.auth)         if response.status_code == 204:             print(f"Source {{source_id}} deleted successfully")         else:             raise Exception(f"Failed to delete source {{source_id}}: {{response.content}}")      def delete_percentage_of_sources(self, percentage):         sources = self.list_sources()         total_sources = len(sources)         num_to_delete = int(total_sources * (percentage / 100))         sources_to_delete = sources[:num_to_delete]                  for source in sources_to_delete:             try:                 self.delete_source(source['sourceId'])             except Exception as e:                 print(f"Error deleting source {{source['sourceId']}}: {{e}}")  # Initialize the Airbyte API client client = AirbyteApiClient(base_url="<http://localhost:8000/api/v1>", username="your_username", password="your_password")  # Delete 98% of sources try:     client.delete_percentage_of_sources(98) except Exception as e:     print(f"Error: {{e}}")

delete_percentage_of_sources

try-except

python
import requests
from requests.auth import HTTPBasicAuth

class AirbyteApiClient:
    def __init__(self, base_url, username, password):
        self.base_url = base_url
        self.auth = HTTPBasicAuth(username, password)

    def list_destinations(self):
        url = f"{{self.base_url}}/destinations/list"
        response = <http://requests.post|requests.post>(url, auth=self.auth)
        if response.status_code == 200:
            return response.json().get('destinations', [])
        else:
            raise Exception(f"Failed to list destinations: {{response.content}}")

    def delete_destination(self, destination_id):
        url = f"{{self.base_url}}/destinations/delete"
        payload = {{"destinationId": destination_id}}
        response = <http://requests.post|requests.post>(url, json=payload, auth=self.auth)
        if response.status_code == 204:
            print(f"Destination {{destination_id}} deleted successfully")
        else:
            raise Exception(f"Failed to delete destination {{destination_id}}: {{response.content}}")

    def delete_percentage_of_destinations(self, percentage):
        destinations = self.list_destinations()
        total_destinations = len(destinations)
        num_to_delete = int(total_destinations * (percentage / 100))
        destinations_to_delete = destinations[:num_to_delete]
        
        for destination in destinations_to_delete:
            try:
                self.delete_destination(destination['destinationId'])
            except Exception as e:
                print(f"Error deleting destination {{destination['destinationId']}}: {{e}}")

# Initialize the Airbyte API client
client = AirbyteApiClient(base_url="<http://localhost:8000/api/v1>", username="your_username", password="your_password")

# Delete 98% of destinations
try:
    client.delete_percentage_of_destinations(98)
except Exception as e:
    print(f"Error: {{e}}")

AirbyteApiClient

list_destinations

/destinations/list

python import requests from requests.auth import HTTPBasicAuth  class AirbyteApiClient:     def __init__(self, base_url, username, password):         self.base_url = base_url         self.auth = HTTPBasicAuth(username, password)      def list_destinations(self):         url = f"{{self.base_url}}/destinations/list"         response = <http://requests.post|requests.post>(url, auth=self.auth)         if response.status_code == 200:             return response.json().get('destinations', [])         else:             raise Exception(f"Failed to list destinations: {{response.content}}")      def delete_destination(self, destination_id):         url = f"{{self.base_url}}/destinations/delete"         payload = {{"destinationId": destination_id}}         response = <http://requests.post|requests.post>(url, json=payload, auth=self.auth)         if response.status_code == 204:             print(f"Destination {{destination_id}} deleted successfully")         else:             raise Exception(f"Failed to delete destination {{destination_id}}: {{response.content}}")      def delete_percentage_of_destinations(self, percentage):         destinations = self.list_destinations()         total_destinations = len(destinations)         num_to_delete = int(total_destinations * (percentage / 100))         destinations_to_delete = destinations[:num_to_delete]                  for destination in destinations_to_delete:             try:                 self.delete_destination(destination['destinationId'])             except Exception as e:                 print(f"Error deleting destination {{destination['destinationId']}}: {{e}}")  # Initialize the Airbyte API client client = AirbyteApiClient(base_url="<http://localhost:8000/api/v1>", username="your_username", password="your_password")  # Delete 98% of destinations try:     client.delete_percentage_of_destinations(98) except Exception as e:     print(f"Error: {{e}}")

AirbyteApiClient

list_destinations

/destinations/list

delete_percentage_of_destinations

delete_destination

/destinations/delete

delete_percentage_of_destinations

Error deleting destination {destination['destinationId']}: {e}

/destinations/delete

204 No Content

Error deleting destination {destination['destinationId']}: {e}

/destinations/delete

204 No Content

python import os import requests from requests.auth import HTTPBasicAuth  class AirbyteApiClient:     def __init__(self, base_url=None) -> None:         self.base_url = base_url or os.getenv("AIRBYTE_BASE_URL")         self.username = os.getenv("BASIC_AUTH_USERNAME")         self.password = os.getenv("BASIC_AUTH_PASSWORD")         self.workspaceId = os.getenv("AIRBYTE_WORKSPACE_ID")      def list_destination(self, limit: int = 20, offset: int = 0) -> Any:         url = f"{self.base_url}/destinations/list"         payload = {"workspaceId": self.workspaceId}         params = {"includeDeleted": False, "limit": limit, "offset": offset}         auth = HTTPBasicAuth(self.username, self.password)         response = <http://requests.post|requests.post>(url, params=params, json=payload, auth=auth)         if response.status_code == 200:             return response.json().get('destinations', [])         else:             raise Exception(f"Failed to list destinations: {response.content}")      def delete_destination(self, destination_id):         url = f"{self.base_url}/destinations/delete"         payload = {"destinationId": destination_id}         auth = HTTPBasicAuth(self.username, self.password)         response = <http://requests.post|requests.post>(url, json=payload, auth=auth)         if response.status_code == 204:             print(f"Destination {destination_id} deleted successfully")         else:             raise Exception(f"Failed to delete destination {destination_id}: {response.content}")      def delete_percentage_of_destination(self, percentage):         destinations = self.list_destination()         total_destinations = len(destinations)         num_to_delete = int(total_destinations * (percentage / 100))         destinations_to_delete = destinations[:num_to_delete]          for destination in destinations_to_delete:             try:                 self.delete_destination(destination['destinationId'])             except Exception as e:                 print(f"Error deleting destination {destination['destinationId']}: {e}")  # Initialize the AirbyteApiClient airbyte_client = AirbyteApiClient(base_url=os.getenv("AIRBYTE_BASE_URL"))  # Delete 98% of destinations try:     airbyte_client.delete_percentage_of_destination(98) except Exception as e:     print(f"Error: {e}")

/destinations/delete

204 No Content

"your_username"

ping

telnet

bash
ping your-database-host
telnet your-database-host 5432

bash
airbyte update --connector source-postgres

GET

ping

telnet

bash ping your-database-host telnet your-database-host 5432

bash airbyte update --connector source-postgres

GET

Authorization

200 OK

create_connection

python
import requests
from requests.auth import HTTPBasicAuth

class AirbyteApiClient:
    def __init__(self, base_url, username, password, connection_config):
        self.base_url = base_url
        self.username = username
        self.password = password
        self.connection_config = connection_config

    def load_config(self, config_path):
        # Implement your configuration loading logic here
        pass

    def create_connection(self, **kwargs) -> None:
        """Create a new connection in the Airbyte API."""
        config_data = self.load_config(self.connection_config)
        url = f"{self.base_url}/connections/create"
        payload = {
            "workspaceId": config_data.get("workspaceId"),
            "name": config_data.get("name"),
            "sourceId": config_data.get("sourceId"),
            "destinationId": config_data.get("destinationId"),
            "namespaceDefinition": config_data.get("namespaceDefinition"),
            "connectionConfiguration": config_data.get("connectionConfiguration"),
        }
        payload.update(kwargs)
        auth = HTTPBasicAuth(self.username, self.password)
        response = <http://requests.post|requests.post>(url, json=payload, auth=auth)
        
        if response.status_code == 200:
            response_data = response.json()
            if response_data.get("connectionId"):
                print("Connection created successfully")
            else:
                raise Exception(f"Failed to create connection: {response.content}")
        else:
            raise Exception(f"Failed to create connection: {response.content}")

        return None

connectionId

connectionId

200 OK

create_connection

python import requests from requests.auth import HTTPBasicAuth  class AirbyteApiClient:     def __init__(self, base_url, username, password, connection_config):         self.base_url = base_url         self.username = username         self.password = password         self.connection_config = connection_config      def load_config(self, config_path):         # Implement your configuration loading logic here         pass      def create_connection(self, **kwargs) -> None:         """Create a new connection in the Airbyte API."""         config_data = self.load_config(self.connection_config)         url = f"{self.base_url}/connections/create"         payload = {             "workspaceId": config_data.get("workspaceId"),             "name": config_data.get("name"),             "sourceId": config_data.get("sourceId"),             "destinationId": config_data.get("destinationId"),             "namespaceDefinition": config_data.get("namespaceDefinition"),             "connectionConfiguration": config_data.get("connectionConfiguration"),         }         payload.update(kwargs)         auth = HTTPBasicAuth(self.username, self.password)         response = <http://requests.post|requests.post>(url, json=payload, auth=auth)                  if response.status_code == 200:             response_data = response.json()             if response_data.get("connectionId"):                 print("Connection created successfully")             else:                 raise Exception(f"Failed to create connection: {response.content}")         else:             raise Exception(f"Failed to create connection: {response.content}")          return None

connectionId

connectionId

connectionConfiguration

syncCatalog

scheduleType

status

json
{
    "workspaceId": "64cf55a5-7a9c-4a15-8928-e3d2eddfaeb7",
    "name": "Dr_Chrono-->Local_JSON",
    "sourceId": "29c4947b-38bc-4908-b818-246f886d6fe9",
    "destinationId": "3a97a545-15f1-42a2-81c7-9047bdbd29e6",
    "namespaceDefinition": "source",
    "prefix": "",
    "operationIds": [],
    "syncCatalog": {
        "streams": [
            {
                "stream": {
                    "name": "patients",
                    "jsonSchema": {
                        "type": "object",
                        "properties": {
                            "id": {"type": "number"},
                            "city": {"type": "string"},
                            "race": {"type": "string"},
                            "email": {"type": "string"},
                            "state": {"type": "string"},
                            "gender": {"type": "string"},
                            "address": {"type": "string"},
                            "zip_code": {"type": "string"},
                            "last_name": {"type": "string"},
                            "first_name": {"type": "string"},
                            "date_of_birth": {"type": "string"}
                        }
                    },
                    "supportedSyncModes": ["full_refresh"]
                },
                "config": {
                    "syncMode": "full_refresh",
                    "destinationSyncMode": "overwrite",
                    "selected": true
                }
            }
        ]
    },
    "scheduleType": "manual",
    "status": "active"
}

syncCatalog

scheduleType

status

prefix

operationIds

create_connection

connectionConfiguration

syncCatalog

scheduleType

status

json {     "workspaceId": "64cf55a5-7a9c-4a15-8928-e3d2eddfaeb7",     "name": "Dr_Chrono-->Local_JSON",     "sourceId": "29c4947b-38bc-4908-b818-246f886d6fe9",     "destinationId": "3a97a545-15f1-42a2-81c7-9047bdbd29e6",     "namespaceDefinition": "source",     "prefix": "",     "operationIds": [],     "syncCatalog": {         "streams": [             {                 "stream": {                     "name": "patients",                     "jsonSchema": {                         "type": "object",                         "properties": {                             "id": {"type": "number"},                             "city": {"type": "string"},                             "race": {"type": "string"},                             "email": {"type": "string"},                             "state": {"type": "string"},                             "gender": {"type": "string"},                             "address": {"type": "string"},                             "zip_code": {"type": "string"},                             "last_name": {"type": "string"},                             "first_name": {"type": "string"},                             "date_of_birth": {"type": "string"}                         }                     },                     "supportedSyncModes": ["full_refresh"]                 },                 "config": {                     "syncMode": "full_refresh",                     "destinationSyncMode": "overwrite",                     "selected": true                 }             }         ]     },     "scheduleType": "manual",     "status": "active" }

syncCatalog

scheduleType

status

prefix

operationIds

create_connection

create_connection

python
import requests
from requests.auth import HTTPBasicAuth

class AirbyteApiClient:
    def __init__(self, base_url, username, password, connection_config):
        self.base_url = base_url
        self.username = username
        self.password = password
        self.connection_config = connection_config

    def load_config(self, config_path):
        # Implement your configuration loading logic here
        pass

    def create_connection(self, **kwargs) -> None:
        """Create a new connection in the Airbyte API."""
        config_data = self.load_config(self.connection_config)
        url = f"{self.base_url}/connections/create"
        payload = {
            "workspaceId": config_data.get("workspaceId"),
            "name": config_data.get("name"),
            "sourceId": config_data.get("sourceId"),
            "destinationId": config_data.get("destinationId"),
            "namespaceDefinition": config_data.get("namespaceDefinition"),
            "prefix": "",
            "operationIds": [],
            "syncCatalog": config_data.get("syncCatalog"),
            "scheduleType": "manual",
            "status": "active"
        }
        payload.update(kwargs)
        auth = HTTPBasicAuth(self.username, self.password)
        response = <http://requests.post|requests.post>(url, json=payload, auth=auth)
        
        if response.status_code == 200:
            response_data = response.json()
            if response_data.get("connectionId"):
                print("Connection created successfully")
            else:
                raise Exception(f"Failed to create connection: {response.content}")
        else:
            raise Exception(f"Failed to create connection: {response.content}")

        return None

AirbyteApiClient

load_config

create_connection

200 OK

connectionId

AirbyteApiClient

create_connection

python import requests from requests.auth import HTTPBasicAuth  class AirbyteApiClient:     def __init__(self, base_url, username, password, connection_config):         self.base_url = base_url         self.username = username         self.password = password         self.connection_config = connection_config      def load_config(self, config_path):         # Implement your configuration loading logic here         pass      def create_connection(self, **kwargs) -> None:         """Create a new connection in the Airbyte API."""         config_data = self.load_config(self.connection_config)         url = f"{self.base_url}/connections/create"         payload = {             "workspaceId": config_data.get("workspaceId"),             "name": config_data.get("name"),             "sourceId": config_data.get("sourceId"),             "destinationId": config_data.get("destinationId"),             "namespaceDefinition": config_data.get("namespaceDefinition"),             "prefix": "",             "operationIds": [],             "syncCatalog": config_data.get("syncCatalog"),             "scheduleType": "manual",             "status": "active"         }         payload.update(kwargs)         auth = HTTPBasicAuth(self.username, self.password)         response = <http://requests.post|requests.post>(url, json=payload, auth=auth)                  if response.status_code == 200:             response_data = response.json()             if response_data.get("connectionId"):                 print("Connection created successfully")             else:                 raise Exception(f"Failed to create connection: {response.content}")         else:             raise Exception(f"Failed to create connection: {response.content}")          return None

AirbyteApiClient

load_config

create_connection

200 OK

connectionId

AirbyteApiClient

python # Initialize the Airbyte API client client = AirbyteApiClient(     base_url="<http://localhost:8000/api/v1>",     username="your_username",     password="your_password",     connection_config="path_to_your_config_file" )  # Create a connection try:     client.create_connection() except Exception as e:     print(f"Error: {e}")

"your_username"

"your_password"