Managing synchronization from Python¶
The etiket-sdk package lets you manage synchronization programmatically: control the sync agent, configure sync sources, map scopes, and inspect or retry individual datasets. This is ideal for servers without a graphical environment and for scripted, reproducible setups.
Installation¶
The SDK talks to the sync agent, so it must be installed and running. See qDrive installation for more information.
The sync agent¶
The SyncAgent class reports the status of the background service and lets you start or stop it.
from etiket_sdk.sync import SyncAgent
status = SyncAgent.status()
print(status)
SyncAgent.stop()
SyncAgent.start()
The status reports whether the agent and API services are running, the current agent state (running, stopped, no_connection, unavailable), the number of completed sync cycles, and the time of the last update. If the status is not running, you can always call SyncAgent.start().
Sync sources¶
A sync source defines where data comes from and how it is synchronized. Each source uses a connector that knows how to read a specific data format (e.g. the QUAlibrate data structure or a QCoDeS database).
Listing and inspecting sources¶
from etiket_sdk.sync import SyncSources
# All configured sources
sources = SyncSources.list()
# Full details for one source, including status and item counts
source = SyncSources.get("my_sync_source")
print(source)
Choosing a connector¶
# Available connector identifiers
print(SyncSources.connectors())
# Details for a specific connector (its name, parameters, scope policy)
connector = SyncSources.connector("etiket_sync_agent_folderbase")
print(connector)
# Print a copy-paste-ready example for creating a source with this connector
connector.example()
Creating a source¶
A source needs a name, a connector identifier, and the connector-specific config_data. The example below uses FolderBase; for the exact config_data of each connector, see the connectors pages.
from etiket_sdk.sync import SyncSources
source = SyncSources.create(
name="my_folderbase_source",
connector_identifier="etiket_sync_agent_folderbase",
config_data={
"root_directory": "~/Downloads/MyData",
"is_server_folder": False,
},
default_scope="6c319227-f6f8-4bc6-b7f9-ed278a8fb040",
)
How the scope is set depends on the connector:
- Required default scope (e.g. FolderBase): you must pass
default_scope. Datasets go there . - Mapped scopes (e.g. Core Tools): datasets carry scope-local identifiers that are mapped to scopes. Pass
autoassign_scope_mapping=Trueto auto-match identifiers to scope names, or manage scope mappings manually.
Finding scope UUIDs
A default_scope is a scope UUID. List the scopes you have access to with from qdrive.scopes import get_scopes; get_scopes().
Updating, starting, stopping, and deleting¶
source = SyncSources.get("my_source")
source.update(rename_to="new_name")
source.update(config_data={"is_server_folder": True})
source.stop() # pause this source
source.start() # resume it
source.delete()
Note
Some configuration changes require a sync agent restart to take effect.
Scope mappings¶
For sources without a required default scope, datasets carry scope-local identifiers (project names, etc.) that are mapped to scopes. For example:
import uuid
source = SyncSources.get("my_source")
# Which identifiers still need a scope?
print(source.unmapped_identifiers())
# Auto-assign identifiers that match exactly one scope name
source.autoassign_mappings()
# Or map manually -- here scope_A is the dataset's scope-local identifier
source.create_mapping("scope_A", uuid.UUID("12345678-1234-1234-1234-123456789abc"))
source.update_mapping("scope_A", uuid.UUID("87654321-4321-4321-4321-cba987654321"))
source.delete_mapping("scope_A")
Set autoassign_scope_mapping=True when creating or updating a source to have matching identifiers mapped automatically on every sync cycle.
Retrying failed items¶
source = SyncSources.get("my_source")
# Retry only the failed items
count = source.reset_sync_items()
# Force a full re-sync of everything
count = source.reset_sync_items(include_successful=True)
Note
Re-syncing never duplicates data: if the dataset already exists on the server (matched by its identifier), the agent updates that dataset instead of creating a new one.
Sync items¶
A sync item is a single dataset queued for synchronization.
source = SyncSources.get("my_source")
# List items, optionally filtered or searched
items = source.items()
not_synced = source.items(is_synchronized=False) # everything not yet synced
matches = source.items(query="measurement_2024")
# Each item has a .status: pending, synchronized, error, or no_mapping
errored = [it for it in not_synced if it.status == "error"]
# Inspect one item and its detailed sync log
item = source.item(123)
print(item.status)
print(item.sync_record)
item.prioritize() # sync this one next
item.reset() # clear attempts and retry
Debugging¶
Errors surface at three levels, and how you inspect them reflects what each level is:
- The sync agent and sync sources are long-running processes. They accumulate an error log over time, which you read with
.errors()and acknowledge with theviewedflag — the two behave identically. This log is cumulative: it keeps past errors, so entries don't necessarily mean something is wrong now — check the timestamps and the currentstatusto judge what's live. - A sync item is a single dataset to be synced, with a definite
status(pending,synchronized,error, orno_mapping— waiting for a scope mapping) and anattemptscount. It has no rolling error log; instead, its diagnostics live in the item's sync record (item.sync_record), which logs the steps attempted during the last sync and where a failure occurred.
When debugging, work from the top down:
from etiket_sdk.sync import SyncAgent, SyncSources
# 1. Agent level — system errors that block all sync (login, network)
print(SyncAgent.status())
for error in SyncAgent.errors(viewed=False):
print(error)
# 2. Source level — configuration or connector errors
source = SyncSources.get("my_source")
for error in source.errors(viewed=False): # or source.print_errors()
print(error)
# 3. Item level — failures syncing individual datasets, found in each item's sync record
for item in source.items(is_synchronized=False):
print(item)
# view in detail what steps were undertaken and where an error occurred:
print(source.item(123).sync_record)
If you can't resolve an issue, send a report to support (this emails the support team with the relevant details attached):
SyncAgent.report_error("Sync agent keeps crashing on startup")
source.report_error("Sync keeps failing with permission denied")
source.item(123).report_error("This file always fails to upload")
Service logs¶
Both local services write to a rotating log file. Read them through the SDK without needing to know where they live:
from etiket_sdk.admin import Logs, LogService
for line in Logs.tail(LogService.api, n=500):
print(line)
Managing connectors¶
The Connectors class lists, installs, updates, and removes connector packages.
from etiket_sdk.sync import Connectors
print(Connectors.list()) # installed connectors
connector = Connectors.get("etiket_sync_agent_qcodes") # details for one
Install and update a connector either from PyPI or from a local path:
Useful when developing your own connector. Use editable=True to pick up code changes without reinstalling.
Remove a connector by its package name:
To build a connector for a data format that isn't covered, see Creating a connector.