Supported Methods

The Fivetran Connector SDK supports these main methods:

• update() is required for every connector. It defines how your connector retrieves and delivers data during sync.
• schema() is optional, but most production connectors need it. It defines the connector's output schema.
• configuration_form() is optional. It defines setup form fields and setup tests for the connector.

update()

Required method that defines how your connector retrieves and delivers data during sync and stores progress in state.

Signature

update(configuration: dict, state: dict)

Parameters

Example

def update(configuration, state):
    log.info("Starting sync.")
    records = get_data(configuration)
    for record in records:
        op.upsert(table="table1", data=record)
    op.checkpoint(state={"table1_cursor": "2024-08-14T02:01:00Z"})

Notes

• Required for every connector.
• Called automatically at the start of each sync.

schema()

Optional method that defines your connector’s output schema. Most connectors need this method. If you don't define the primary key(s) for a table, we create a surrogate primary key column named _fivetran_id. It contains a hashed value generated from full set of the row's values. See our system columns documentation for more details.

If not defined, Fivetran infers schema from data.

Signature

schema(configuration: dict)

Parameters

Returns

Returns list of table definition dictionaries as JSON object.

The dictionary should use the following keys:

• The schema key is optional. Use it to declare which schema a table belongs to. Without it, the table goes to the destination schema named after the connection; with it, the table gets its own named schema in the destination. In your upsert(), update(), delete(), and truncate() calls, pass the same schema key you declared. Fivetran uses it to match each row to the right table definition and apply your primary keys and column types. The value must be a non-empty string and cannot be the connection name. You can mix table definitions with and without schema. For more information, see Declaring multiple schemas.
• The table key is required and specifies the name of the table.
• The primary_key key is optional but recommended. The value is a list of one or more primary key column names. The content of the list is used as the table's primary key; a single entry means a simple primary key while multiple entries are combined to create a composite primary key for the table. We recommend that you provide primary keys for your tables.
• The column key is optional, its value is a dictionary of key-value pairs where the key is the column name and the value is data type.

Example

def schema(configuration: dict):
    return [{
        "table": "company",
        "primary_key": ["id"],
        "columns": {"id": "INT", "name": "STRING"}
    }]

Example with multiple schemas

The examples above omit schema, so the table goes to the destination schema named after the connection. Use schema when your connector syncs tables from more than one source schema:

def schema(configuration: dict):
    return [
        {
            "schema": "sales",
            "table": "orders",
            "primary_key": ["id"],
            "columns": {"id": "LONG", "amount": "FLOAT"}
        },
        {
            "schema": "support",
            "table": "tickets",
            "primary_key": ["id"],
            "columns": {"id": "LONG", "status": "STRING"}
        }
    ]

Notes

• We reject duplicate table definitions and any schema named like the connection, in any letter case. You can mix table definitions with and without schema. Tables without schema belong to the destination schema named after the connection. A table definition applies only to its exact combination of schema and table name.
• If you change an existing single-schema connector to declare schemas, we don't backfill the previously synced data into the tables in their new schemas, and you must re-sync the connection yourself. For more information, see Declaring multiple schemas.
• We infer the schema for data you send us if you do not define it. However, if you want to set a primary key for a table or configure columns to have specific data types, then use this method.
• Our Connector SDK GitHub Repo has many examples of how to use the schema() method.
  • For an example of defining a schema with one table, see our weather example.
  • For an example with multiple tables, see our using_pd_dataframes example.
  • For an example of a table with multiple keys, see our multiple_tables_with_cursors example.
  • For an example of how columns in a schema response can be defined with specific data types, see our specified_types example.
• If you don't provide the primary key to use in a table, Fivetran creates a surrogate primary key column named _fivetran_id, which is a hashed value generated based on the row's set of values. See our system columns documentation for more details.
• If a new row is received with a different set of columns, we calculate the hash from the new row's values, including values from any new columns. This leads to all rows being treated and written as distinct rows, which may be perceived as data integrity issues in the destination. In this case, you may have to drop and re-sync the connection to preserve data integrity. Thus, we recommend defining primary keys for your tables to avoid unexpected behavior.
• The values you provide for the table, primary_key and column keys are renamed based on our renaming rules so that the corresponding names in the destination may differ from how they are set in your code. If you want the table and column names in the destination to exactly match the names you set in your code, we recommend adhering to the renaming rules ensuring the names align with the pattern and character set of transformed names. This means that names of the tables and columns in your source may not exactly match the corresponding names in the destination.
• If you need to change primary key selections for a table, drop the table in your destination and then select Resync all historical data on the connection's Setup tab in your dashboard. Doing so maintains data integrity across all records.
• Fivetran maintains a list of system columns that are automatically created in your destination for every connector. We recommend that you do not use these reserved names for your own columns to avoid conflicts.

Example of undefined table or primary key

Assume Fivetran receives the following row for a table not defined in the schema or defined without a primary key:

The generated hashed value would be from the values of the three columns.

Consider we receive the same row with an additional column, bar:

The generated hashed value would differ from the first row as the hashed value is calculated from the values of all the columns. This results in two distinct rows written in the destination.

Syncing empty tables and columns

Fivetran creates tables and columns in your destination for any column declared in the schema() method, even if there is no data sent for that column.

For more information, see our Features documentation.

configuration_form()

Optional method that defines the setup form shown for your Connector SDK connection. Use this method when you want users to provide configuration values in a connection setup form in the Fivetran dashboard instead of relying only on a manually created configuration.json file.

Signature

configuration_form()

Returns

Returns a ConfigurationForm object containing form fields and optional setup tests.

Example

def configuration_form():
    form = ConfigurationForm()
    form.add_field(form_field.TextField(
        name="api_key",
        label="API Key",
        field_type=form_field.TextField.password,
        required=True,
    ))
    form.add_test(label="Test connection", func=connection_test)
    return form

connector = Connector(
    update=update,  # required; add schema=schema if you define a schema
    configuration_form=configuration_form,
)

Notes

• The configuration_form() method is optional.
• Pass it to the Connector object using configuration_form=configuration_form, as shown in the example above.
• Field values are passed to schema() and update() through the configuration dictionary.
• For supported field types, setup tests, and local configuration generation, see our Setup Form documentation.