eTiKeT API

REST and Python API for the eTiKeT platform. This package provides the local API service that allows external tools (like the eTiKeT SDK) to interact with the sync agent and manage local authentication.

Overview

The eTiKeT API runs as a local service. It provides:

• REST API: HTTP endpoints for authentication, sync management, and more
• Python API: Direct Python functions for internal use

Note: Most users should use the eTiKeT SDK instead of calling this API directly. The SDK provides a simpler, higher-level interface.

Installation

pip install etiket-api

Running the Service

Normally, the sync API runs as a system service in the background (managed by the eTiKeT installer). However, you can run it manually for development or debugging:

python3.11 -m etiket_api

By default, the service runs on http://localhost:10410. The interactive API documentation is available at http://localhost:10410/docs (Swagger UI).

REST API Overview

The REST API is available at http://localhost:10410/api/v1/. Interactive API documentation is available at http://localhost:10410/docs (Swagger UI).

Local API Tokens

Manage API tokens stored on this machine. Tokens allow background sync to authenticate with the server without an active login session.

Sync Agent

Manage the background sync agent service.

Sync Sources

Manage sync sources — configurations that define where data comes from.

Sync Items

Manage sync items — individual datasets to be synchronized.

Sync Connectors

Manage sync connector packages.

Sync Converters

Manage converter packages for file format conversion.

Admin

Inspect service state for support and debugging.

Error Handling

All API errors return a consistent JSON error model:

{
  "error": "sync_source_not_found",
  "detail": "Sync source with id 123 not found"
}

• error: A machine-readable error code (snake_case)
• detail: A human-readable error message (optional)

Error Codes Reference

General Errors

Auth Errors

Sync Agent Errors

Sync Source Errors

Sync Item Errors

Connector Errors

Converter Errors

Admin Errors

Python API

The Python API provides direct access to functionality. It's primarily intended for internal use by the REST API layer and the sync agent itself.

Local API Tokens

from etiket_api.python_api.auth.local.api_tokens import (
    create_api_token,
    list_api_tokens,
    delete_api_token,
    check_api_token,
)

# Create a token for the current user
token = await create_api_token(name="my-sync-token")
print(token.api_token)  # only shown on create

# List all local tokens
tokens = await list_api_tokens()
for t in tokens:
    print(f"{t.user_name}: {t.api_key_name} ({t.server_url})")

# Check if a token is still valid
result = await check_api_token(api_key_uid="...")
print(result.valid)

# Delete a token (revokes on server first)
await delete_api_token(api_key_uid="...")

Sync Agent

from etiket_api.python_api.sync.sync_agent import (
    get_sync_status,
    get_sync_errors,
    start_sync_agent,
    stop_sync_agent,
    set_sync_status_error_viewed,
    mark_all_sync_status_errors_viewed,
)

# Get current status
status = await get_sync_status()

# Fetch only unviewed errors (UI "unread" pane)
unread = await get_sync_errors(limit=10, skip=0, viewed=False)

# Mark one as viewed; or mark all of them
await set_sync_status_error_viewed(error_id=unread[0].id, viewed=True)
flipped = await mark_all_sync_status_errors_viewed()

# Start/stop the agent
await start_sync_agent()
await stop_sync_agent()

Sync Sources

from etiket_api.python_api.sync.sync_sources import (
    list_sync_sources,
    get_sync_source,
    get_sync_source_by_name,
    update_sync_source,
    delete_sync_source,
    start_sync_source,
    stop_sync_source,
    list_sync_source_definitions,
    reset_sync_source_sync_items,
    take_ownership,
    release_ownership,
    # Scope mapping functions
    create_scope_mapping,
    update_scope_mapping,
    delete_scope_mapping,
    get_unmapped_scope_identifiers,
    autoassign_scope_mappings,
    # Exceptions
    SyncSourceAlreadyRunningError,
    SyncSourceAlreadyStoppedError,
)

# List all sources
sources = list_sync_sources()

# Get a specific source
source = get_sync_source(source_id=1)
source = get_sync_source_by_name("my_source")

# Start/stop a sync source
try:
    source = start_sync_source(source_id=1)
except SyncSourceAlreadyRunningError:
    print("Source is already running")

try:
    source = stop_sync_source(source_id=1)
except SyncSourceAlreadyStoppedError:
    print("Source is already stopped")

# Reset failed sync items for a source (to retry them)
reset_count = reset_sync_source_sync_items(source_id=1)
print(f"Reset {reset_count} failed items")

# Reset ALL sync items (including successful ones)
reset_count = reset_sync_source_sync_items(source_id=1, include_successful=True)
print(f"Reset {reset_count} items (all)")

# List available connectors
connectors = list_sync_source_definitions()

# Claim ownership of a sync source for the current user (overwrites any existing owner)
source = await take_ownership(source_id=1)
print(source.owner)

# Release your own ownership (errors if you are not the current owner)
source = await release_ownership(source_id=1)
assert source.owner is None

# Error-log "viewed" management
from etiket_api.python_api.sync.sync_sources import (
    read_sync_source_errors,
    set_sync_source_error_viewed,
    mark_all_sync_source_errors_viewed,
)

# Fetch only unviewed errors (UI "unread" pane)
unread = await read_sync_source_errors(source_id=1, viewed=False)

# Mark one error as viewed, or un-view it
await set_sync_source_error_viewed(error_id=unread[0].id, viewed=True)

# Mark everything as viewed in one shot; returns the count flipped
flipped = await mark_all_sync_source_errors_viewed(source_id=1)

Scope Mappings

Scope mappings are used when a connector provides per-item scope routing. Each sync item may have a scopeIdentifier string that needs to be mapped to an actual scope UUID.

import uuid
from etiket_api.python_api.sync.sync_sources import (
    create_scope_mapping,
    update_scope_mapping,
    delete_scope_mapping,
    get_unmapped_scope_identifiers,
    autoassign_scope_mappings,
)

# Find identifiers that don't have a mapping yet
unmapped = get_unmapped_scope_identifiers(sync_source_id=1)
print(f"Unmapped identifiers: {unmapped}")

# Create a mapping from identifier to scope UUID
mapping = create_scope_mapping(
    sync_source_id=1,
    scope_identifier="my_scope_identifier",
    scope_uuid=uuid.UUID("12345678-1234-1234-1234-123456789abc"),
)

# Update an existing mapping to point to a different scope
mapping = update_scope_mapping(
    sync_source_id=1,
    scope_identifier="my_scope_identifier",
    scope_uuid=uuid.UUID("87654321-4321-4321-4321-cba987654321"),
)

# Delete a mapping
delete_scope_mapping(sync_source_id=1, scope_identifier="my_scope_identifier")

# Auto-assign mappings based on scope name matching
# (creates mappings where identifier matches exactly one scope name)
created_count = autoassign_scope_mappings(sync_source_id=1)
print(f"Auto-assigned {created_count} mappings")

Sync Items

from etiket_api.python_api.sync.sync_items import (
    list_sync_items,
    get_sync_item,
    reset_sync_item,
    prioritize_sync_item,
)

# List items for a source
items = list_sync_items(source_id=1)

# Get a specific item
item = get_sync_item(sync_item_id=123)

# Reset or prioritize an item
reset_sync_item(sync_item_id=123)
prioritize_sync_item(sync_item_id=123)

Connectors

from etiket_api.python_api.sync.connectors import (
    list_connectors,
    install_connector_from_pypi,
    install_connector_from_local_path,
    update_connector_from_pypi,
    uninstall_connector,
)

# List installed connectors
connectors = list_connectors()

# Install from PyPI
install_connector_from_pypi("etiket-sync-agent-qcodes")

# Install from local path
install_connector_from_local_path("/path/to/my-connector")

# Update from PyPI
update_connector_from_pypi("etiket-sync-agent-qcodes")

# Uninstall
uninstall_connector("etiket_sync_agent_qcodes")

Converters

from etiket_api.python_api.sync.converters import (
    list_converters,
    install_converter_from_pypi,
    install_converter_from_local_path,
    update_converter_from_pypi,
    uninstall_converter,
)

# List installed converters
converters = list_converters()

# Install from PyPI
install_converter_from_pypi("my-converter-package")

# Install from local path
install_converter_from_local_path("/path/to/my-converter")

# Update from PyPI
update_converter_from_pypi("my-converter-package")

# Uninstall
uninstall_converter("my_converter")

Admin / Logs

from etiket_api.python_api.admin.logs import (
    LogService,
    get_recent_log_lines,
)

# Read the last 500 lines of the API's log file
lines = get_recent_log_lines(LogService.api, n=500)
for line in lines:
    print(line)

# Read the entire log file (up to the 10 MB rotation cap)
all_lines = get_recent_log_lines(LogService.sync_agent, n=-1)

Architecture

┌─────────────────────────────────────────────────────────────┐
│                      eTiKeT SDK                             │
│                   (User-facing Python SDK)                  │
└─────────────────────────┬───────────────────────────────────┘
                          │ HTTP
                          ▼
┌─────────────────────────────────────────────────────────────┐
│                      eTiKeT API                             │
│  ┌──────────────────┐   ┌──────────────────────────────┐    │
│  │    REST API      │   │        Python API            │    │
│  │    (FastAPI)     │──▶│   auth/local, sync/agent,    │    │
│  │                  │   │   sync/sources, sync/items.. │    │
│  └──────────────────┘   └──────────────────────────────┘    │
└─────────────────────────────────────────────────────────────┘
                          │
                          ▼
┌─────────────────────────────────────────────────────────────┐
│                   eTiKeT Sync Agent                         │
│            (Background synchronization service)             │
└─────────────────────────────────────────────────────────────┘

Requirements

• Python 3.10+
• etiket_sync_agent
• etiket_client