This repository is owned and maintained by the Fivetran Professional Services team. It provides a Python-based framework for interacting with the Fivetran REST API to support automation, monitoring, and advanced integration workflows.

The repository also includes example solutions from customer engagements covering:

• AI-driven workflows
• HVR (High Volume Replication) automation
• MCP-based orchestration patterns
• Fivetran Connector SDK integrations

These examples are designed to be educational and practical, helping teams understand what’s possible with the Fivetran platform and accelerate real-world implementations.

If you’d like guidance, customization, or hands-on support, book a demo with Fivetran Professional Services to connect with our experts.

This repository is provided as a reference implementation to help users better understand and work with Fivetran automations. It is intended as a starting point for custom solutions that may require additional design, hardening, or scaling for enterprise use.

• Fivetran API Docs
• Postman Collection
• REST API FAQs

• The function 'atlas' is a general-purpose function to interact with the Fivetran API. It takes three parameters: method, endpoint, and payload.
• The method parameter determines the HTTP method to use (GET, POST, PATCH, DELETE). The endpoint parameter specifies the API endpoint to interact with. The payload parameter is used to send data in the case of POST or PATCH requests.
• The function constructs the full URL for the API request, sets up the headers including the authorization, and makes the request using the requests library.
• If the request is successful, it returns the JSON response. If the request fails, it prints an error message and returns None.
• The function uses exception handling to catch any errors that occur during the request and to raise an exception if the HTTP status code indicates an error.

• Install libraries:
  • requests
  • json
  • colorama
• Set the api_key and api_secret variables to your Fivetran API credentials.
• Run the script with the desired method, endpoint, and payload.

• Create New User
• Create New Team
• Create New Group Webhook
• Certificate Management
• Create New Connector
• Resync Table
• Create New Connector in Multiple Destinations
• Create New Destinations
• Run a dbt Transformation
• Update Connector Settings
• Delete Connector
• Get Connector Status
• Get Connector Status using Pagination
• Get a list of connectors in a specific status
• Manage Schema Configurations
• Pause or Sync connectors and log the action(s)
• Re-write SQL files

This Python script is designed to interact with an API, specifically the Fivetran API, to retrieve and display the status of connectors. It uses the requests library to send HTTP requests and the colorama library to colorize the output. Step-by-step Breakdown

The script begins by importing the necessary Python libraries. These include requests for making HTTP requests, json for handling JSON data, and colorama for colorizing the terminal output.

        import requests
        from requests.auth import HTTPBasicAuth
        import json
        import colorama
        from colorama import Fore, Back, Style

This function is used to send HTTP requests to the Fivetran API. It takes three parameters: method (the HTTP method), endpoint (the API endpoint), and payload (the request body for POST and PATCH requests). It constructs the request, sends it, and returns the response as a JSON object.

        def atlas(method, endpoint, payload):
        base_url = 'https://api.fivetran.com/v1'
        h = {
        'Authorization': f'Bearer {api_key}:{api_secret}'
        }
        url = f'{base_url}/{endpoint}'
        ...

The script then specifies the parameters for the API request. In this case, it's making a GET request to the 'groups/{group_id}/connectors' endpoint.

        group_id = ''
        method = 'GET'
        endpoint = 'groups/' + group_id + '/connectors'
        payload = ''

The script calls the atlas function with the specified parameters and stores the response.

response = atlas(method, endpoint, payload)

Finally, the script checks if the response is not None, prints the request and response details, and iterates over the 'items' in the response data, printing the 'service', 'sync_state', and 'sync_frequency' for each item.

     if response is not None:
      print(Fore.CYAN + 'Call: ' + method + ' ' + endpoint + ' ' + str(payload))
      print(Fore.GREEN + 'Response: ' + response['code'])
      cdata_list =  response['data']
      ctimeline  =  cdata_list['items']
  for c in ctimeline:
      print(Fore.MAGENTA + 'Type:' + c['service'] + Fore.BLUE + ' Status:' + c['status']['sync_state'] + Fore.YELLOW + ' Frequency:' + str(c['sync_frequency']))

This Python script is designed to interact with an API, specifically the Fivetran API, to pause a given connector and log the actions. It uses the requests library to send HTTP requests and the colorama library to colorize the output. Step-by-step Breakdown

  import requests
  from requests.auth import HTTPBasicAuth
  import json
  import colorama
  from colorama import Fore
  import os
  import logging
  from logging.handlers import RotatingFileHandler

This function is used to make HTTP requests to the Fivetran API. It takes three parameters: the HTTP method (GET, POST, PATCH, DELETE), the API endpoint, and the payload (data to send with the request). The function constructs the request, sends it, and logs the result.

   def atlas(method, endpoint, payload):
    base_url = 'https://api.fivetran.com/v1'
    h = {
        'Authorization': f'Bearer {api_key}:{api_secret}'
    }
    url = f'{base_url}/{endpoint}'

    try:
        if method == 'GET':
            response = requests.get(url, headers=h, auth=a)
        elif method == 'POST':
            response = requests.post(url, headers=h, json=payload, auth=a)
        elif method == 'PATCH':
            response = requests.patch(url, headers=h, json=payload, auth=a)
        elif method == 'DELETE':
            response = requests.delete(url, headers=h, auth=a)
        else:
            raise ValueError('Invalid request method.')

        response.raise_for_status()  # Raise exception

        logger.info(f'Successful {method} request to {url}')
        return response.json()
    except requests.exceptions.RequestException as e:
        logger.error(f'Request failed: {e}')
        print(f'Request failed: {e}')
        return None

The script sets up a logger that writes to a file (api_framework.log). If the log file exceeds 10MB, it is overwritten. The logger is set to log INFO level messages and above. A rotating file handler is added to the logger, which keeps the last 3 log files when the current log file reaches 10MB.

   log_file = "/api_framework.log"
   log_size = 10 * 1024 * 1024  # 10 MB
    
    #Check if the log file size exceeds 10MB
    if os.path.exists(log_file) and os.path.getsize(log_file) >= log_size:
        # If it does, overwrite the file
        open(log_file, 'w').close()
    
    logger = logging.getLogger(__name__)
    logger.setLevel(logging.INFO)
    
    #Add a rotating handler
    handler = RotatingFileHandler(log_file, maxBytes=log_size, backupCount=3)
    logger.addHandler(handler)

The script constructs a request to the Fivetran API to pause a connector (identified by connector_id). The HTTP method is PATCH, the endpoint is connectors/{connector_id}, and the payload is {"paused": True}.

    connector_id = ''
    method = 'PATCH'  #'POST' 'PATCH' 'DELETE' 'GET'
    endpoint = 'connectors/' + connector_id 
    payload = {"paused": True}
    
    #Submit
    response = atlas(method, endpoint, payload)

The script calls the atlas function to send the request and get the response. If the response is not None, it prints the request details and response in different colors. In this example, we successfully paused a connector and logged the activity.

    if response is not None:
      print(Fore.CYAN + 'Call: ' + method + ' ' + endpoint + ' ' + str(payload))
      print(Fore.GREEN +  'Response: ' + response['code'])
      print(Fore.MAGENTA + str(response))

Book a consultation with a Fivetran Services expert.

• This solution was written by: Elijah Davis
• Special Thanks: Angel Hernandez & Jimmy Hooker