Creating a connector¶

A connector is a plugin that defines how the sync agent reads datasets from a particular data format or acquisition system. Pre-built connectors exist for QCoDeS, Quantify, Labber, Core Tools, and generic folders — if your data lives in a format that isn't covered, you can write your own. These connectors are open source, so you can install any of them from PyPI and read its source for inspiration.

This guide explains the two kinds of connector, the package you build, and the two classes you implement.

Types of connectors¶

Every connector is one of two kinds, depending on where your datasets live and how they are discovered:

Database — the sync agent does not know which datasets exist, so you tell it which ones need syncing. The defining trait is that you enumerate the datasets yourself; the classic case is a database that assigns a new id to each measurement, but "database" is meant loosely here — the source can be anything you track ids in (even a file). You implement two things: a function that returns the identifiers of datasets that still need syncing, and a function that turns each identifier into a synced dataset.
File-based — the sync agent discovers datasets automatically by watching the file system. It assumes datasets live at a known depth in a folder tree, so folders (or files) at that depth are registered as datasets without you listing them. You implement a single function that turns a detected folder into a synced dataset.

Kind	Base class	You implement
Database	`SyncSourceDatabaseBase`	`get_new_datasets()` + `sync_dataset_normal()`
File-based	`SyncSourceFileBase`	`root_path()` + `sync_dataset_normal()`

Both kinds optionally implement check_live_dataset() and sync_dataset_live() for live synchronization.

Package structure¶

A connector is an ordinary Python package with a small, fixed layout:

etiket-sync-agent-my-connector/
├── pyproject.toml                       # package config + entry point
└── etiket_sync_agent_my_connector/
    ├── __init__.py                      # exports the sync and config classes
    ├── my_connector_config_class.py       # the configuration dataclass
    └── my_connector_sync_class.py         # the sync logic

You don't have to create these by hand — the scaffold generator below produces exactly this structure.

Generating the structure automatically¶

First install the core sync-agent package, which provides the base classes and the scaffold-generator CLI:

pip install etiket_sync_agent

Then generate a connector package. Pick --base database or --base file depending on the type:

# A database-backed connector
generate_sync_agent_scaffold my-connector \
    --base database \
    --default-scope REQUIRED

# A file-based connector that detects datasets two folders deep, with live sync
generate_sync_agent_scaffold my-connector \
    --base file \
    --level 2 \
    --live-sync

Key options:

Option	Description
`--base`	`database` or `file` — picks the base class template
`--default-scope`	Scope policy: `REQUIRED` or `DISABLED`
`--live-sync`	Include a live-sync method stub
`--level`	(file-based only) directory depth at which datasets are detected
`--dependency`	Add a package dependency (repeatable)

The generated package has the boilerplate in place; you fill in the configuration and sync logic described below.

Technical detail: how connectors are discovered

Connectors are distributed as ordinary Python packages. Each declares an entry point in the etiket_sync_agent.connectors group (the scaffold writes this for you):

# in your connector package's pyproject.toml
[project.entry-points."etiket_sync_agent.connectors"]
my_connector_sync_agent = "etiket_sync_agent_my_connector:MyConnectorSync"

When the package is installed into the sync agent's environment, the agent finds it through this entry point automatically — there is no central registry to update.

The configuration class¶

The configuration class lives in my_connector_config_class.py. It is a dataclass describing the parameters a user supplies when creating a sync source — these become the config_data dict in the SDK. Implement an async validate() method that checks the configuration and raises ValueError on problems.

import dataclasses, pathlib
from typing import Optional

@dataclasses.dataclass
class MyConnectorConfigData:
    database_path: pathlib.Path

    def __post_init__(self):
        self.database_path = pathlib.Path(self.database_path).expanduser()

    async def validate(self, current_sync_source=None):
        path = self.database_path.resolve(strict=True)
        if not path.is_file():
            raise ValueError(f"Not a file: {path}")
        return True

The sync class¶

The sync class lives in my_connector_sync_class.py. It declares a few class variables and implements the methods for its base type. All methods are staticmethods and may be async.

DatabaseFile-based

Implement get_new_datasets() to report which datasets still need syncing, and sync_dataset_normal() to sync one. Each item you return is a SyncItemCreate with a data_identifier (a stable string id for the dataset) and a required sync_priority (an int or float that increases for newer datasets). The agent remembers the highest sync_priority it has synced and passes that item back as last_sync_item, so get_new_datasets() should return only datasets with a higher priority:

import typing
from etiket_sync_agent.connectors.sync_source_abstract import (
    SyncSourceDatabaseBase, ScopeRequirement,
)
from etiket_sync_agent.schemas import SyncItemSchema, SyncItemCreate
from etiket_sync_agent.sync.sync_records.manager import SyncRecordManager
from etiket_sync_agent.sync.sync_utilities import SyncUtilities

class MyConnectorSync(SyncSourceDatabaseBase):
    sync_agent_name: typing.ClassVar[str] = "MyConnector"
    config_data_class: typing.ClassVar[type] = MyConnectorConfigData
    scope_requirement: typing.ClassVar[ScopeRequirement] = ScopeRequirement.REQUIRED
    supports_scope_mapping: typing.ClassVar[bool] = False
    live_sync_implemented: typing.ClassVar[bool] = False
    has_owner: typing.ClassVar[bool] = True

    @staticmethod
    async def get_new_datasets(
        config_data: MyConnectorConfigData,
        last_sync_item: SyncItemSchema | None,
    ) -> typing.List[SyncItemCreate]:
        # last_sync_item is the highest-priority item synced so far, or None on
        # the first run. Query your source for everything with a higher priority.
        last_priority = -1
        if last_sync_item is not None:
            last_priority = last_sync_item.sync_priority
        new_records = my_query_for_records_newer_than(config_data, last_priority)
        return [
            SyncItemCreate(
                data_identifier=str(record.id),
                sync_priority=record.priority,
            )
            for record in new_records
        ]

    @staticmethod
    async def sync_dataset_normal(
        config_data: MyConnectorConfigData,
        sync_item: SyncItemSchema,
        sync_record: SyncRecordManager,
    ):
        with sync_record.task("Create or update dataset"):
            await SyncUtilities.create_or_update_dataset(...)
        with sync_record.task("Upload files"):
            await SyncUtilities.upload_file(...)

    @staticmethod
    async def check_live_dataset(
        config_data: MyConnectorConfigData,
        sync_item: SyncItemSchema,
        max_priority: bool,
    ) -> bool:
        # Return True while the dataset is still being written to.
        return False

Implement root_path() to return the directory to scan, and sync_dataset_normal() to sync a detected folder:

import typing
from pathlib import Path
from etiket_sync_agent.connectors.sync_source_abstract import (
    SyncSourceFileBase, ScopeRequirement,
)
from etiket_sync_agent.schemas import SyncItemSchema
from etiket_sync_agent.sync.sync_records.manager import SyncRecordManager
from etiket_sync_agent.sync.sync_utilities import SyncUtilities

class MyConnectorSync(SyncSourceFileBase):
    sync_agent_name: typing.ClassVar[str] = "MyConnector"
    config_data_class: typing.ClassVar[type] = MyConnectorConfigData
    scope_requirement: typing.ClassVar[ScopeRequirement] = ScopeRequirement.REQUIRED
    supports_scope_mapping: typing.ClassVar[bool] = False
    live_sync_implemented: typing.ClassVar[bool] = False
    has_owner: typing.ClassVar[bool] = True
    level: typing.ClassVar[int] = -1        # -1 = detect datasets at any depth
    is_single_file: typing.ClassVar[bool] = False

    @staticmethod
    def root_path(config_data: MyConnectorConfigData) -> Path:
        # example
        return Path(config_data.my_directory)

    @staticmethod
    async def sync_dataset_normal(
        config_data: MyConnectorConfigData,
        sync_item: SyncItemSchema,
        sync_record: SyncRecordManager,
    ):
        # example
        dataset_path = Path(config_data.my_directory) / sync_item.data_identifier
        with sync_record.task("Create dataset"):
            await SyncUtilities.create_or_update_dataset(...)
        with sync_record.task("Upload files"):
            for file_path in dataset_path.glob("*"):
                if file_path.is_file():
                    await SyncUtilities.upload_file(...)

    @staticmethod
    async def check_live_dataset(
        config_data: MyConnectorConfigData,
        sync_item: SyncItemSchema,
        max_priority: bool,
    ) -> bool:
        return False

Class variables¶

Variable	Meaning
`sync_agent_name`	Human-readable display name
`config_data_class`	The configuration dataclass (above)
`scope_requirement`	Whether a `default_scope` must be set when creating the source: `ScopeRequirement.REQUIRED` (the user must provide one) or `DISABLED` (none needed — datasets get their scope from `scope_identifier`)
`supports_scope_mapping`	Whether items carry a `scope_identifier` (set in `get_new_datasets`) to be mapped to a scope — used when there is no single default scope
`live_sync_implemented`	Whether `sync_dataset_live` is implemented
`has_owner`	Whether an item may name an `owner` to override the source's default owner, so that user's credentials are used for the upload
`level` (file-based)	Directory depth at which datasets are detected
`is_single_file` (file-based)	`True` if each dataset is a single file rather than a folder

Two of these pair with optional fields you can set on each SyncItemCreate in get_new_datasets:

scope_identifier — a string (e.g. a project name) that is mapped to a scope when supports_scope_mapping=True. The Core Tools connector is a good example. If omitted, the source's default scope is used.
owner — overrides the source's default owner for that dataset (requires has_owner=True), so the named user's credentials are used to upload it.

For file-based connectors, level controls how deep below root_path() the agent looks for datasets, and is_single_file decides whether a dataset is a folder or one file. For example, given this tree:

root/                         <- root_path()
└── 2024-01-01/               <- level 1
    └── experiment_42/        <- level 2
        ├── data.hdf5         <- level 3
        └── analysis.hdf5

level = 2, is_single_file = False → each experiment_42/ folder is one dataset (all files inside belong to it).
level = 3, is_single_file = True → each .hdf5 file at depth 3 is its own dataset.
level = -1 → any folder containing a _QH_dataset_info.yaml file is a dataset, at any depth (FolderBase's default).

Live synchronization¶

If you set live_sync_implemented = True, also implement sync_dataset_live() (same signature as sync_dataset_normal()). The agent calls it for datasets that are still being acquired, and check_live_dataset() decides when a dataset is still "live".

Tooling for the sync methods¶

Your sync methods receive the sync item (sync_item) and a sync_record. Use the SyncUtilities helpers to create the dataset and upload data — each takes a DatasetInfo or FileInfo object, the sync item, and the sync_record:

from datetime import datetime
from etiket_sync_agent.schemas import SyncItemSchema
from etiket_sync_agent.sync.sync_records.manager import SyncRecordManager
from etiket_sync_agent.sync.sync_utilities import (
    SyncUtilities, DatasetInfo, FileInfo, FileType,
)

async def sync_dataset_normal(
    config_data: MyConnectorConfigData,
    sync_item: SyncItemSchema,
    sync_record: SyncRecordManager,
):
    with sync_record.task("Create or update the dataset"):
        ds_info = DatasetInfo(
            name="my dataset",
            dataset_uuid=sync_item.dataset_uuid,
            scope_uuid=sync_item.scope_uuid,
            created=datetime.now(),
            attributes={"sample": "Q7-R3"},
        )
        # first argument is live_mode (False for a normal sync)
        await SyncUtilities.create_or_update_dataset(
            False, sync_item, ds_info, sync_record
        )

    with sync_record.task("Upload files"):
        f_info = FileInfo(name="measurement", file_name="measurement.hdf5",
                           created=datetime.now(), file_type=FileType.HDF5_NETCDF)
        await SyncUtilities.upload_xarray(xr_dataset, sync_item, f_info, sync_record)
        # also available: upload_file(path, ...), upload_json(obj, ...)
        # and add_thumbnail(id, path, ...) — see below

The sync_record groups work into logged steps and is what users inspect when debugging:

with sync_record.task("..."): — records the step's timing and result (and any exception raised inside it).
sync_record.add_log("...") — add an informational log line.
sync_record.add_error(message, error, stacktrace=None) — record a non-fatal error.

Dataset thumbnails¶

A connector can give a dataset thumbnails: the images shown next to it in the dataset list of the DataQruiser app, which makes a long list of measurements much easier to scan. Hand add_thumbnail() an image and the number it should get — 0 is the image the dataset is shown with, 1, 2, ... are extras:

from etiket_sync_agent.sync.sync_utilities import SyncUtilities, IMAGE_SUFFIXES

with sync_record.task("Add thumbnails"):
    images = sorted(p for p in dataset_path.glob("*") if p.suffix.lower() in IMAGE_SUFFIXES)
    for id, image_path in enumerate(images):
        await SyncUtilities.add_thumbnail(id, image_path, sync_item, sync_record)

Each image is downscaled and re-encoded to roughly 100 kB, so thumbnails stay quick to load. Upload the image itself as a regular file as well if users should be able to open it at full resolution.

Number the images in a stable order, so that re-syncing a dataset maps every image onto the same thumbnail again instead of creating new versions.

Note

add_thumbnail() expects an image. A missing file, a file that is not an image, or an image that cannot be read is reported as an error on the sync record (and returns False) rather than passing silently — so filter the files you offer it on IMAGE_SUFFIXES, as above. A broken image never stops the rest of the dataset from syncing.

Install and use your connector¶

Install your connector into the same environment as the sync agent so it is discovered. While developing, an editable install lets you iterate without reinstalling:

etiket-services pip install -e .

You can also install it through the SDK:

from etiket_sdk.sync import Connectors

Connectors.install_from_local("/path/to/etiket-sync-agent-my-connector", editable=True)
print(Connectors.list())   # your connector should now appear

Once installed, create a sync source that uses it — see Managing synchronization from Python.

Reference implementations

The best reference is the existing connectors. Study the etiket-sync-agent-qcodes (database) and FolderBase (file) packages for complete, working examples. Need help? Contact support@qharbor.nl.

Testing your connector¶

You don't need a full test suite to get going — these levels run from cheapest to most thorough.

1. Unit-test the config class. validate() should accept good configuration and reject bad — call it directly:

import asyncio, pytest
from etiket_sync_agent_my_connector import MyConnectorConfigData

with pytest.raises(ValueError):
    asyncio.run(MyConnectorConfigData(database_path="/does/not/exist.db").validate())

If validate() also guards against configuration that must be unique across sources — e.g. two sync sources shouldn't point at the same folder or database — you can test that too. The check queries the agent's existing sources, so you build a stand-in source and mock the lookup to return it:

Test duplicate rejection

import asyncio, pytest
from unittest.mock import patch
from etiket_sync_agent.crud.sync_sources import crud_sync_sources
from etiket_sync_agent.models.sync_sources import SyncSources, SyncSourceStatus
from etiket_sync_agent_my_connector import MyConnectorConfigData

cfg = MyConnectorConfigData(database_path="tests/fixtures/example.db")

# a source that already syncs the same data
existing_source = SyncSources(
    name="other_source",
    status=SyncSourceStatus.SYNCING,
    connector="etiket_sync_agent_my_connector",
    config_data={"database_path": "tests/fixtures/example.db"},
)

async def list_existing(_session):
    return [existing_source]

with (
    patch("etiket_sync_agent.db.get_db_session_context"),
    patch.object(crud_sync_sources, "list_sync_sources", list_existing),
):
    with pytest.raises(ValueError):
        asyncio.run(cfg.validate())

etiket-sync-agent-qcodes's tests/test_config_cls.py shows the same approach as reusable fixtures, plus the no-conflict case (an empty source list → validate() passes).

2. Unit-test discovery in isolation. The discovery method is pure and easy to call directly: point it at a small fixture source and check what it returns. For a database connector:

import asyncio
from etiket_sync_agent_my_connector import MyConnectorSync, MyConnectorConfigData

config = MyConnectorConfigData(database_path="tests/fixtures/example.db")
items = asyncio.run(MyConnectorSync.get_new_datasets(config, None))

assert len(items) == 3                       # 3 datasets in the fixture
assert items[0].data_identifier == "run-1"

# sync_priority drives incremental sync, so check it's set and ordered oldest → newest
priorities = [i.sync_priority for i in items]
assert priorities == sorted(priorities)      # ascending
assert len(set(priorities)) == len(priorities)   # unique per dataset

Also test the incremental case: given a last_sync_item, get_new_datasets must return only datasets with a higher sync_priority. It reads nothing else from that item, so a mock is enough:

Test incremental discovery with a mocked last_sync_item

import asyncio
from unittest.mock import MagicMock
from etiket_sync_agent_my_connector import MyConnectorSync, MyConnectorConfigData

config = MyConnectorConfigData(database_path="tests/fixtures/example.db")
all_items = asyncio.run(MyConnectorSync.get_new_datasets(config, None))

# get_new_datasets only reads .sync_priority from last_sync_item
last = MagicMock()
last.sync_priority = all_items[-1].sync_priority   # pretend we synced up to here

newer = asyncio.run(MyConnectorSync.get_new_datasets(config, last))
assert newer == []                               # nothing newer than the last item

For a file-based connector, check that root_path(config) is correct and that your expected folders are detected.

3. Run sync_dataset_normal offline with mocked uploads. sync_dataset_normal reaches the server through SyncUtilities. Patch those calls so they just print what would happen, and you can exercise your whole sync_dataset_normal — without a server or login — to confirm it reads the right files and builds the right dataset/file metadata.

A print-only mock harness

import asyncio, uuid
from unittest.mock import MagicMock, patch
from etiket_sync_agent_my_connector import MyConnectorSync, MyConnectorConfigData

# sync_record: .task(...) as a no-op context manager
sync_record = MagicMock()
sync_record.task.return_value.__enter__ = MagicMock()
sync_record.task.return_value.__exit__ = MagicMock(return_value=False)

# sync_item: set the fields your connector reads
sync_item = MagicMock()
sync_item.data_identifier = "run-1"
sync_item.dataset_uuid = uuid.uuid4()
sync_item.scope_uuid = uuid.uuid4()

async def fake_create(live, item, ds_info, rec):
    print(f"create/update dataset: {ds_info.name}  attrs={ds_info.attributes}")

async def fake_upload(data, item, f_info, rec):
    print(f"upload file: {f_info.file_name}")

config = MyConnectorConfigData(database_path="tests/fixtures/example.db")

# patch SyncUtilities on the module where your sync class imports it
su = "etiket_sync_agent_my_connector.my_connector_sync_class.SyncUtilities"
with (
    patch(f"{su}.create_or_update_dataset", side_effect=fake_create),
    patch(f"{su}.upload_file", side_effect=fake_upload),
    patch(f"{su}.upload_xarray", side_effect=fake_upload),
    patch(f"{su}.upload_json", side_effect=fake_upload),
):
    asyncio.run(MyConnectorSync.sync_dataset_normal(config, sync_item, sync_record))

Running this prints the create/upload calls your connector would make, e.g.:

create/update dataset: run-1  attrs={'sample': 'Q7-R3'}
upload file: measurement.hdf5

4. Verify the real sync. For a true integration check, install your connector (editable), create a sync source against a small test dataset, let it sync, then inspect the result with the SDK:

from etiket_sdk.sync import SyncSources

source = SyncSources.get("my_test_source")
print(source)                                # status + item counts
for item in source.items(is_synchronized=False):
    print(item)                              # anything that failed
print(source.item(123).sync_record)          # step-by-step log of one item

See Debugging for the full set of inspection tools.

Tip

For deeper automated tests, the pre-built connector packages (e.g. etiket-sync-agent-qcodes) ship complete pytest suites worth using as a template — they use exactly this MagicMock + patch approach.

Converters¶

File-based connectors can also run converters that transform files during sync (for example CSV → HDF5 so they plot automatically). A converter is a separate package, registered through its own entry point and discovered the same way as a connector. See File converters for the full guide.