Unstructured File Uploads Private Preview

Unstructured file uploads let you send PDFs, images, archives, or other binary content to your destination alongside structured metadata. The Connector SDK supports these uploads through an optional file parameter on the upsert() and update() operations.

Fivetran always uploads the file first, then the metadata row. If a sync fails during or after a file upload, we retry the entire operation from the last checkpoint.

Existing connectors that do not use the file parameter continue to work unchanged.

File uploads are supported only for destinations that support Unstructured file replication. Using file uploads with unsupported destinations causes the sync to fail.
File uploads are not supported in Hybrid Deployment (HD) mode.
Your connector must work with open streams, HTTP responses, or in-memory objects like BytesIO. It cannot accept file paths as strings from users or save files to disk during the sync.

Using `file` with `upsert()` and `update()`

To use file uploads in your connector, import the FileUpload class:

from fivetran_connector_sdk import FileUpload

For more information, see Required Declarations.

Signature

op.upsert(table="name", data=data, file=file_upload)
op.update(table="name", modified=data, file=file_upload)

The file parameter takes a FileUpload object that bundles the destination file path, binary input stream, and optional file size for integrity checking. If omitted, the operation behaves as a standard upsert() or update() with no file upload.

`FileUpload` dataclass

from dataclasses import dataclass
from typing import Optional

@dataclass(frozen=True)
class FileUpload:
    """File input for upload operations."""
    path: str                              # Path in the table's namespace (stored in _fivetran_file_path)
    stream                                 # Any stream with read(size) -> bytes method
    expected_bytes: Optional[int] = None   # Optional file size for integrity verification

Parameter	Type	Required	Description
`path`	str	Yes	Path within the table's namespace for the file, such as `"invoices/2026/123.pdf"`. The destination validates the path and enforces any size or character limits. Leading and trailing whitespace is automatically trimmed, so `" abc.pdf"` and `"abc.pdf"` are treated as the same value. The SDK stores this value in the auto-created `_fivetran_file_path` column. This is the path within your table's namespace, not a full cloud storage URL like `s3://...`. Do not create or set this column manually. If you include `_fivetran_file_path` in your data dictionary, the SDK overwrites it with the correct path and logs a warning.
`stream`	Any stream object	Yes	Any stream with a `read(size) -> bytes` method. Compatible types include `io.BytesIO`, file handles from `open("file.pdf", "rb")`, and `requests.raw`. The SDK reads the stream incrementally and does not close it. Your code manages the stream lifecycle. The stream does not need to support `seek()` or have a known length upfront.
`expected_bytes`	int	No	Expected file size in bytes. If provided, the SDK verifies the actual bytes written match this value and fails the sync if they don't match, for example, if the stream was truncated. A zero-byte file is supported; set `expected_bytes = 0` to verify the file is empty. Omit if the size is unknown.

Examples

These are quickstart examples that use inline code to show how to upload a file using the file parameter.

Upload a file from an HTTP response

The stream must contain raw, decoded bytes. If your source compresses the response (for example, with gzip or deflate), decode it first. When using the requests library, always set response.raw.decode_content = True before passing response.raw to FileUpload. Without this, the SDK uploads compressed bytes and the file is corrupted in the destination.

import requests
from fivetran_connector_sdk import FileUpload, Operations as op

for invoice in api.list_invoices():
    destination_path = f"invoices/2026/{invoice['id']}.pdf"
    with requests.get(invoice["pdf_url"], stream=True) as response:
        response.raise_for_status()
        response.raw.decode_content = True  # Ensure decompressed bytes
        op.upsert(
            table="invoices",
            data={
                "id": invoice["id"],
                "updated_at": invoice["updated_at"],
                "size": invoice["size"],
            },
            file=FileUpload(
                path=destination_path,
                stream=response.raw,
                expected_bytes=invoice["size"],  # Optional integrity check
            ),
        )

op.checkpoint(state={"cursor": invoice["updated_at"]})

Upload a file from bytes in memory

import io
from fivetran_connector_sdk import FileUpload, Operations as op

file_bytes = b"PDF binary content here..."
file_stream = io.BytesIO(file_bytes)

op.upsert(
    table="generated_files",
    data={"id": "gen_123", "format": "pdf"},
    file=FileUpload(
        path="generated/file_123.pdf",
        stream=file_stream,
        expected_bytes=len(file_bytes),
    ),
)

The upsert(), update(), and checkpoint() operations are thread-safe, so you can safely upload files from multiple threads in parallel using ThreadPoolExecutor or similar approaches.

For complete working examples, see the Connector SDK unstructured data examples on GitHub:

file_lifecycle: covers the full file operation lifecycle, including upload, update, and delete.
stream_examples: demonstrates different streaming approaches, including HTTP responses, BytesIO, and file handles.

Updating files

To update a file for an existing row, call upsert() or update() with the same primary key and a new FileUpload.

Use update() if you only want to replace the file and optionally update specific columns. Other columns remain unchanged. Use upsert() only if you need to replace all columns or if the row might not exist yet. When using upsert(), you must provide all metadata column values; columns you don't provide are set to null.

# Replaces only the file and specified columns
op.update(
    table="documents",
    modified={
        "doc_id": "123",           # Primary key (required)
        "doc_name": "updated.pdf"  # Only the columns you want to change
    },
    file=FileUpload(path="updated.pdf", stream=new_stream)
)

# Replaces all columns — unprovided columns are set to null
op.upsert(
    table="documents",
    data={
        "doc_id": "123",
        "doc_name": "updated.pdf",
        "doc_type": "invoice",
        "created_at": "2026-01-01"
    },
    file=FileUpload(path="updated.pdf", stream=new_stream)
)

When you update a file:

Same path: The destination replaces the existing file. This operation is idempotent.
Different path: The destination uploads the new file, deletes the old staged file, and updates the path reference.

Deleting files

Use op.delete() to soft-delete the metadata row:

op.delete(table="documents", keys={"doc_id": "123"})

This soft-deletes the row, but the file remains in the destination stage and Fivetran does not remove it automatically. You cannot delete only the file while keeping the metadata row.

Local testing

Run fivetran debug to test file uploads locally. The SDK writes uploaded files to:

<project directory>/files/storage/<schema or defaultSchema>/<table>/<your_file_path>

Inspect the files in this directory to verify the content is correct.