etiket-sdk 0.3.0b1

Python SDK for the eTiKeT platform

eTiKeT SDK

Python SDK for the eTiKeT platform. The Sync Agent runs as a background service on your computer, automatically synchronizing your experimental data to qHarbor. This SDK allows you to easily interact with the Sync Agent and manage your sync sources.

Installation

pip install etiket-sdk

---

Sync Agent

The SyncAgent class provides methods to check the status of the background sync service, start and stop it, and view recent errors. For more information about debugging, see the Debugging section.

Checking Status

from etiket_sdk.sync import SyncAgent

# Get the current status
status = SyncAgent.status()
print(status)

Output:

Sync Agent Status:
  Agent Service: running
  API Service: running
  Agent Status: running
  Iterations: 142
  Last Update: 2025-12-16 10:30:00

The status shows:

• Agent Service: Whether the sync agent service is installed and running
• API Service: Whether the sync API service is running (the service that allows the SDK to interact with the Sync Agent)
• Agent Status: Current state (running, stopped, no_connection, unavailable). When the sync agent is not running, the API service is not available.
• Iterations: Number of sync cycles completed
• Last Update: When the status was last updated

Starting and Stopping

# Stop the sync agent
SyncAgent.stop()

# Start the sync agent
SyncAgent.start()

Viewing Errors

If sync isn't working, check for agent-level errors. Note: always check the timestamps to see if errors are recent.

# Get recent errors
errors = SyncAgent.errors()
for error in errors:
    print(error)

# Or fetch only the unread inbox
unread = SyncAgent.errors(viewed=False)

Errors carry a viewed flag so you can acknowledge what you've already triaged:

# Mark a single error as viewed (or unviewed)
unread[0].mark_viewed()

# Bulk-acknowledge — returns the count of rows actually flipped
flipped = SyncAgent.mark_all_errors_viewed()

These are system-level errors that block sync from running entirely (e.g., not logged in, no network connection).

---

Sync Sources

A Sync Source defines where data comes from and how it should be synchronized. Each source uses a backend that knows how to read data from a specific format (e.g., QCoDeS, Quantify, or a folder of files).

Listing Sources

from etiket_sdk.sync import SyncSources

# List all configured sync sources
sources = SyncSources.list()
print(sources)

Output:

['QH datasets' (backend='etiket_sync_agent_native'), 'my_sync_source' (backend='etiket_sync_agent_folderbase')]

This lists all configured sync sources. Use SyncSources.get(name) to get full details for a specific source.

Getting Source Details

# Get a specific source by name
source = SyncSources.get("my_sync_source")
print(source)

Output:

SyncSource: my_sync_source
  id: 2
  backend: etiket_sync_agent_folderbase
  status: scanning
  items: 28/28 synced, 0 failed
  last_update: 2025-12-15 14:54:00

The output shows sync progress at a glance.

Viewing Source Errors

When the sync source is unable to run because of an error, the status will be error. You can fetch and view errors for a source:

source = SyncSources.get("my_source")
source.print_errors()

# Filter by viewed flag — pass viewed=False to print only unread errors
source.print_errors(viewed=False)

Errors carry a viewed flag so you can acknowledge what you've already triaged:

unread = source.errors(viewed=False)
unread[0].mark_viewed()                # mark one
unread[0].mark_viewed(viewed=False)    # un-view if you change your mind

# Bulk-acknowledge — returns the count of rows actually flipped
flipped = source.mark_all_errors_viewed()

Listing Available Backends

Before creating a source, see what backends are available:

# List backend identifiers
print(SyncSources.backends())
# ['etiket_sync_agent_folderbase', 'etiket_sync_agent_quantify', 'etiket_sync_agent_qcodes', ...]

# Get details for a specific backend
backend = SyncSources.backend("etiket_sync_agent_folderbase")
print(backend)

Generating a Create Example

Each backend can generate a copy-paste ready example:

SyncSources.backend("etiket_sync_agent_folderbase").example()

Output:

# Copy-paste this example to create a sync source:
SyncSources.create(
    name="my_folderbase_source",
    backend_identifier="etiket_sync_agent_folderbase",
    config_data={
        "root_directory": "<path to folder>",
        "is_server_folder": False,
    },
)

Creating a Sync Source

from etiket_sdk.sync import SyncSources

source = SyncSources.create(
    name="my_folderbase_source",
    backend_identifier="etiket_sync_agent_folderbase",
    config_data={
        "root_directory": "~/Downloads/MyData",
        "is_server_folder": False,
    },
    default_scope="6c319227-f6f8-4bc6-b7f9-ed278a8fb040",
)

Scope Options:

The default_scope and autoassign_scope_mapping options depend on the backend:

• Required default scope (e.g., FolderBase): The default_scope must be provided. All items sync to this scope.

• Optional/No default scope (e.g., core_tools, QCoDeS): Items may have scope identifiers that need to be mapped to scopes. Use autoassign_scope_mapping=True to automatically match identifiers to scope names:

source = SyncSources.create(
    name="my_coretools_source",
    backend_identifier="etiket_sync_agent_coretools",
    config_data={"db_path": "/path/to/db"},
    autoassign_scope_mapping=True,  # auto-map identifiers to scopes
)

See Scope Mappings for more details on managing scope mappings.

Updating a Sync Source

source = SyncSources.get("my_source")

# Rename a source
source.update(rename_to="new_name")

# Update configuration
source.update(config_data={"is_server_folder": True})

Note: Some configuration changes may require a sync agent restart to take effect.

Deleting a Sync Source

source = SyncSources.get("my_source")
source.delete()

Source Ownership

Some backends (e.g. those that need a per-user API token to upload) attach an owner to each sync source — that's the user whose stored credentials get used for uploads. You can check, claim, or release ownership from the SDK:

source = SyncSources.get("my_source")
print(source.owner)            # the current owner's user_sub, or None

# Claim ownership for the currently logged-in user. 
source.take_ownership()

# Release current ownership.
source.release_ownership()

release_ownership() is idempotent — calling it on an already-unowned source is a no-op. Backends that don't have an owner concept (e.g. etiket_sync_agent_native) will reject take_ownership() with SyncSourceBackendHasNoOwner:

from etiket_sdk.sync import SyncSourceBackendHasNoOwner

try:
    source.take_ownership()
except SyncSourceBackendHasNoOwner:
    print("This backend doesn't support per-user ownership")

Starting and Stopping Sync Sources

You can pause and resume individual sync sources without affecting others:

source = SyncSources.get("my_source")

# Stop a running sync source
source.stop()

# Start a stopped sync source
source.start()

Status transitions:

• Start: Changes status from stopped or error to syncing. Raises SyncSourceAlreadyRunning if the source is already syncing or scanning.
• Stop: Changes status from syncing, scanning, or error to stopped. Raises SyncSourceAlreadyStopped if the source is already stopped.

from etiket_sdk.sync import (
    SyncSourceAlreadyRunning,
    SyncSourceAlreadyStopped,
)

source = SyncSources.get("my_source")

try:
    source.start()
except SyncSourceAlreadyRunning:
    print("Source is already running!")

try:
    source.stop()
except SyncSourceAlreadyStopped:
    print("Source is already stopped!")

Resetting Sync Items

If many items have failed and you want to retry them all at once (instead of resetting each individually), you can reset all failed sync items for a source:

source = SyncSources.get("my_source")

# Reset only failed items (retry them)
count = source.reset_sync_items()
print(f"Reset {count} failed items")

# Reset ALL items (including successful ones, forcing re-sync)
count = source.reset_sync_items(include_successful=True)
print(f"Reset {count} items")

This is useful when:

• Multiple items failed due to a temporary issue (e.g. permissions, ...)
• You need to re-sync everything after changing the default scope
• You want to force a full re-sync of all data

Reporting Errors

If something isn't working, you can send an error report to qHarbor support:

source = SyncSources.get("my_source")
source.report_error("Sync keeps failing with permission denied")

Scope Mappings

For sync sources that don't have a required default scope (like core_tools), sync items may have scope identifiers (project names, etc.) that need to be mapped to specific scopes in qHarbor. This allows items from different folders to be synced to different scopes.

Note: If your sync source has a required default scope, all items are synced to that scope and mappings are not needed.

Enabling Auto-Assign:

When creating or updating a sync source, you can enable automatic scope mapping. When enabled, the sync agent will automatically create mappings for identifiers that match exactly one scope name (case-insensitive):

# Enable auto-assign when creating a source
source = SyncSources.create(
    name="my_source",
    backend_identifier="etiket_sync_agent_folderbase",
    config_data={"root_directory": "/data", "is_server_folder": False},
    autoassign_scope_mapping=True,  # Enable automatic scope mapping
)

# Or enable it on an existing source
source = SyncSources.get("my_source")
source.update(autoassign_scope_mapping=True)

Viewing Current Mappings:

source = SyncSources.get("my_source")

# View existing mappings
for mapping in source.scope_mappings:
    print(f"{mapping.scope_identifier} -> {mapping.scope_uuid}")

# Check how many items need mapping
print(f"Items needing mapping: {source.items_unmapped_scope_count}")

Finding Unmapped Identifiers:

source = SyncSources.get("my_source")
unmapped = source.unmapped_identifiers()
print(f"Need mapping: {unmapped}")
# ['scope_A', 'scope_B', ...]

Manual Auto-Assign:

If autoassign_scope_mapping is not enabled on the source, you can manually trigger auto-assignment. This creates mappings for identifiers that match exactly one scope name (case-insensitive):

source = SyncSources.get("my_source")
count = source.autoassign_mappings()
print(f"Created {count} mappings automatically")

Note: When autoassign_scope_mapping=True is set on the sync source, this happens automatically during each sync cycle, so manual calls are not needed.

Manual Mapping Management:

For identifiers that don't match a scope name or match multiple scopes, you can create mappings manually:

import uuid

source = SyncSources.get("my_source")

# Create a new mapping
source.create_mapping(
    scope_identifier="scope_A",
    scope_uuid=uuid.UUID("12345678-1234-1234-1234-123456789abc"),
)

# Update an existing mapping
source.update_mapping(
    scope_identifier="scope_A",
    scope_uuid=uuid.UUID("87654321-4321-4321-4321-cba987654321"),
)

# Delete a mapping
source.delete_mapping(scope_identifier="scope_A")

Handling Mapping Errors:

from etiket_sdk.sync import (
    ScopeMappingNotFound,
    ScopeMappingAlreadyExists,
    ScopeMappingInvalidScope,
)

source = SyncSources.get("my_source")

try:
    source.create_mapping("scope_A", some_uuid)
except ScopeMappingAlreadyExists:
    print("Mapping already exists - use update_mapping instead")
except ScopeMappingInvalidScope:
    print("The target scope UUID doesn't exist")

---

Sync Items

A Sync Item represents a single dataset that needs to be synchronized. Sync items are automatically created when the sync agent scans your sources.

Listing Items

from etiket_sdk.sync import SyncSources

source = SyncSources.get("my_source")

# List all items for a source
items = source.items()
for item in items:
    print(item)

# Filter by sync status
failed_items = source.items(is_synchronized=False)

# Search by data identifier
items = source.items(query="measurement_2024")

Getting Item Details

source = SyncSources.get("my_source")

# Get a specific item by ID
item = source.item(123)
print(item)

# View the sync record with detailed logs
print(item.sync_record)

Prioritizing an Item

If you need an item to sync immediately:

source = SyncSources.get("my_source")
item = source.item(123)
item.prioritize()

Resetting a Failed Item

If an item failed and you want to retry:

source = SyncSources.get("my_source")
item = source.item(123)
item.reset()

This clears the attempt counter and marks it for retry.

Reporting Errors

source = SyncSources.get("my_source")
item = source.item(123)
item.report_error("This file keeps failing to upload")

---

Backends

In the Sync Sources section, we used backends to create sync sources. The Backends class lets you manage installed backend packages — install new ones, update existing ones, or uninstall them.

Backends are implementations that know how to synchronize data from specific data acquisition software. We provide pre-built backends for popular packages:

Backend	Package	Description
etiket_sync_agent_qcodes	QCoDeS	Sync QCoDeS datasets
etiket_sync_agent_quantify	Quantify	Sync Quantify datasets
etiket_sync_agent_labber	Labber	Sync Labber log files
etiket_sync_agent_coretools	Core Tools	Sync Core Tools datasets
etiket_sync_agent_folderbase	FolderBase	Sync files from a folder

Installing Backends

from etiket_sdk.sync import Backends

# List installed backends
print(Backends.list())

# Get a specific backend and print its details
backend = Backends.get("etiket_sync_agent_qcodes")
print(backend)

# Install from PyPI
Backends.install_from_pypi("etiket-sync-agent-qcodes")

# Install a specific version
Backends.install_from_pypi("qcodes", version="0.45.0")

# Install from local development path
Backends.install_from_local("/path/to/my-backend")

# Install in editable mode (for development)
Backends.install_from_local("/path/to/my-backend", editable=True)

# Update a backend
Backends.update_from_pypi("etiket-sync-agent-qcodes")

# Uninstall a backend
Backends.uninstall("etiket_sync_agent_qcodes")

You can also create your own backends. See the Backend Development Guide for more information.

---

Converters

Converters automatically transform files from one format to another during sync. For example, you can convert CSV files to HDF5 so they can be plotted automatically in qHarbor.

Converters are supported when using the FolderBase backend.

Installing Converters

from etiket_sdk.sync import Converters

# List installed converters
print(Converters.list())

# Get a specific converter and print its details
converter = Converters.get("my_converter")
print(converter)

# Install from PyPI
Converters.install_from_pypi("my-converter-package")

# Install from local development path
Converters.install_from_local("/path/to/my-converter")

# Update a converter
Converters.update_from_pypi("my-converter-package")

# Uninstall a converter
Converters.uninstall("my_converter")

For information on creating your own converters, see the Converter Development Guide.

---

Debugging

When sync isn't working, errors are caught at three levels. Debug from top to bottom:

1. Sync Agent Errors

These are system-level errors that block sync entirely (e.g., not logged in, no network).

from etiket_sdk.sync import SyncAgent

# Check if the agent is running
status = SyncAgent.status()
print(status)

# If not running or status shows issues, check errors
errors = SyncAgent.errors()
for error in errors:
    print(error)

2. Sync Source Errors

These are source-level errors that prevent items from being discovered or processed.

from etiket_sdk.sync import SyncSources

# Check the source status
source = SyncSources.get("my_source")
print(source)  # Shows status and item counts

# View recent errors
source.print_errors()

Look for:

• Configuration errors
• Backend connection issues
• Permission problems

3. Sync Item Errors

These are item-level errors that occur while syncing individual items.

from etiket_sdk.sync import SyncSources

source = SyncSources.get("my_source")

# List failed items
failed = source.items(is_synchronized=False)
for item in failed:
    print(item)

# Get detailed error info for a specific item
item = source.item(123)
print(item.sync_record)  # Shows detailed logs and errors

Debugging Workflow

1. Start at the top: Check SyncAgent.status() — is it running? Any login or network issues?
2. Check agent errors: SyncAgent.errors() — any agent-level errors?
3. Check source status: SyncSources.get("name") — is the source in an error state?
4. Check source errors: source.print_errors() — any configuration issues?
5. Check failed items: source.items(is_synchronized=False) — which items failed?
6. Get item details: source.item(id).sync_record — what went wrong?

Reporting Issues

If you can't resolve an issue, send an error report to qHarbor support. This sends an automatic email to qHarbor support with error details attached:

# Report agent issues
SyncAgent.report_error("Sync agent keeps crashing on startup")

# Report source issues
source = SyncSources.get("my_source")
source.report_error("Sync keeps failing with permission denied")

# Report item issues
item = source.item(123)
item.report_error("This file always fails to upload")

---

Admin / Logs

Both local services (sync API and sync agent) write to a rotating log file in the shared data directory. The SDK exposes them via the Logs class so you don't need to know where they live on disk.

from etiket_sdk.admin import Logs, LogService

# Last 500 records of the API service log
for line in Logs.tail(LogService.api, n=500):
    print(line)

# Full sync-agent log (current file plus the rotation backup, up to ~20 MB)
all_lines = Logs.tail(LogService.sync_agent, n=-1)

n counts logical records, not physical lines: multi-line entries (e.g. stack traces) come back as a single string with embedded \n, so a tail request will never slice through the middle of a traceback.

Raises LogFileNotFound if the requested service hasn't run yet (its log file doesn't exist on disk).

---

Configuration

By default, the SDK connects to the local sync service at localhost:10410. You can change this if needed (though this is generally not necessary):

from etiket_sdk._client import EtiketClient

# Configure once at application startup
EtiketClient.configure(base_url="http://my-server:10410")