Core API

The core package provides foundational classes for XNAT server communication, configuration management, authentication, and error handling.

Client

The XNATClient class is the primary HTTP client for all XNAT REST API operations. It provides automatic retry logic, session management, pagination support, and connection pooling.

Key Features:

• Automatic retry with exponential backoff for transient errors (502, 503, 504)

• Session-based authentication with token caching

• Pagination support for large result sets

• SSL verification control

• Context manager support for automatic cleanup

• Connection pooling via httpx

Basic Usage:

from xnatctl.core.client import XNATClient

# Create and authenticate client
client = XNATClient(
    base_url="https://xnat.example.org",
    username="admin",
    password="secret",
    timeout=60,
    verify_ssl=True
)

client.authenticate()

# Make API calls
response = client.get("/data/projects")

# Use as context manager
with XNATClient(base_url="https://xnat.example.org") as client:
    client.authenticate()
    data = client.get("/data/projects")

Class Reference:

class xnatctl.core.client.XNATClient(base_url, username=None, password=None, session_token=None, timeout=21600, max_retries=3, verify_ssl=True, auto_reauth=False)

HTTP client for XNAT REST API with retry and pagination.

base_url: str

username: str | None = None

password: str | None = None

session_token: str | None = None

timeout: int = 21600

max_retries: int = 3

verify_ssl: bool = True

auto_reauth: bool = False

close()

Close the HTTP client.

__enter__()

__exit__(*args)

property is_authenticated: bool

Check if client has a session token.

authenticate()

Authenticate with username/password and get JSESSIONID.

Returns:

Session token (JSESSIONID).

Raises:

AuthenticationError – If authentication fails.

Return type:

str

invalidate_session()

Logout and clear session token.

get(path, *, params=None, headers=None, timeout=None, stream=False)

GET request.

post(path, *, params=None, json=None, data=None, files=None, headers=None, timeout=None)

POST request.

put(path, *, params=None, json=None, data=None, files=None, headers=None, timeout=None)

PUT request.

delete(path, *, params=None, headers=None, timeout=None)

DELETE request.

paginate(path, *, params=None, page_size=100, result_key='ResultSet.Result')

Paginated GET returning items one by one.

Parameters:

• path (str) – API path.

• params (dict[str, Any] | None) – Additional query parameters.

• page_size (int) – Number of items per page.

• result_key (str) – Dot-separated path to results in response.

Yields:

Individual result items.

get_json(path, *, params=None)

GET request returning JSON.

ping()

Check server connectivity and get version info.

Returns:

Dict with server info.

Raises:

NetworkError – If server is unreachable.

Return type:

dict[str, Any]

whoami()

Get current user information.

Returns:

Dict with user info.

Raises:

AuthenticationError – If not authenticated.

Return type:

dict[str, Any]

__init__(base_url, username=None, password=None, session_token=None, timeout=21600, max_retries=3, verify_ssl=True, auto_reauth=False)

Config

Configuration management with YAML file support and environment variable overrides. Supports multiple server profiles for different XNAT environments.

Configuration File Location:

• ~/.config/xnatctl/config.yaml (default)

Profile Structure:

Each profile defines connection parameters for one XNAT server:

default_profile: production
output_format: table

profiles:
  production:
    url: https://xnat.example.org
    username: admin
    verify_ssl: true
    timeout: 30
    default_project: MYPROJECT

Credential Resolution:

The Config class resolves credentials in this priority order:

1. Environment variables (XNAT_URL, XNAT_USER, XNAT_PASS)

2. Config file profile settings

3. Default values

Class Reference:

class xnatctl.core.config.Config(default_profile='default', output_format='table', profiles=<factory>)

Application configuration.

default_profile: str = 'default'

output_format: str = 'table'

profiles: dict[str, Profile]

classmethod load(config_path=None)

Load config from file with environment variable overrides.

Priority (highest to lowest): 1. Environment variables 2. Config file 3. Defaults

Parameters:

config_path (Path | None) – Optional path to config file.

Returns:

Loaded configuration.

Return type:

Config

save(config_path=None)

Save config to file (excludes secrets).

Parameters:

config_path (Path | None) – Optional path to config file.

get_profile(name=None)

Get profile by name or default.

Parameters:

name (str | None) – Profile name. If None, uses default_profile.

Returns:

Profile configuration.

Raises:

ProfileNotFoundError – If profile doesn’t exist.

Return type:

Profile

has_profile(name)

Check if profile exists.

add_profile(name, url, verify_ssl=True, timeout=21600, default_project=None)

Add or update a profile.

Parameters:

• name (str) – Profile name.

• url (str) – XNAT server URL.

• verify_ssl (bool) – Whether to verify SSL certificates.

• timeout (int) – Request timeout in seconds.

• default_project (str | None) – Default project ID.

Returns:

Created profile.

Return type:

Profile

remove_profile(name)

Remove a profile.

Parameters:

name (str) – Profile name.

Returns:

True if removed, False if didn’t exist.

Return type:

bool

set_default_profile(name)

Set the default profile.

Parameters:

name (str) – Profile name to set as default.

Raises:

ProfileNotFoundError – If profile doesn’t exist.

class xnatctl.core.config.Profile(url, verify_ssl=True, timeout=21600, default_project=None, username=None, password=None, workers=None, overwrite=None, direct_archive=None, archive_mode=None, extract=None)

Configuration profile for an XNAT server.

url: str

verify_ssl: bool = True

timeout: int = 21600

default_project: str | None = None

username: str | None = None

password: str | None = None

workers: int | None = None

overwrite: str | None = None

direct_archive: bool | None = None

archive_mode: str | None = None

extract: bool | None = None

to_dict()

Convert to dictionary for serialization.

classmethod from_dict(data)

Create from dictionary.

Authentication

Session-based authentication with token caching. Handles login, logout, and session validation.

Session Token Storage:

• Cached at ~/.config/xnatctl/.session per profile

• Automatically reused until expiration

• Can be overridden with XNAT_TOKEN environment variable

Authentication management for xnatctl.

Handles credential storage and session token caching.

class xnatctl.core.auth.CachedSession(token, url, username, created_at, expires_at=None)

Cached session token with metadata.

token: str

url: str

username: str

created_at: datetime

expires_at: datetime | None = None

is_expired()

Check if session has expired.

to_dict()

Convert to dictionary for serialization.

classmethod from_dict(data)

Create from dictionary.

class xnatctl.core.auth.AuthManager(cache_file=None)

Manages authentication credentials and session tokens.

get_credentials()

Get credentials from environment variables.

Returns:

Tuple of (username, password).

Return type:

tuple[str | None, str | None]

get_token_from_env()

Get session token from environment variable.

Returns:

Token if set.

Return type:

str | None

save_session(token, url, username, expiry_minutes=15)

Save session token to cache.

Parameters:

• token (str) – Session token (JSESSIONID).

• url (str) – XNAT server URL.

• username (str) – Username used for authentication.

• expiry_minutes (int) – Minutes until session is considered expired.

Returns:

Cached session object.

Return type:

CachedSession

load_session(url=None)

Load cached session token.

Parameters:

url (str | None) – Optional URL to match. If provided, only returns session for that URL.

Returns:

Cached session if valid, None otherwise.

Return type:

CachedSession | None

clear_session()

Clear cached session.

Returns:

True if cache was cleared.

Return type:

bool

has_valid_session(url=None)

Check if there’s a valid cached session.

Parameters:

url (str | None) – Optional URL to match.

Returns:

True if valid session exists.

Return type:

bool

get_session_token(url=None)

Get session token from cache or environment.

Priority: 1. Environment variable (XNAT_TOKEN) 2. Cached session

Parameters:

url (str | None) – Optional URL to match for cached session.

Returns:

Session token if available.

Return type:

str | None

get_session_info(url=None)

Get session information for display.

Parameters:

url (str | None) – Optional URL to match.

Returns:

Dict with session info or None.

Return type:

dict | None

Exceptions

Comprehensive exception hierarchy for error handling.

Exception Hierarchy:

XNATCtlError (base)
├── ConfigurationError
│   └── ProfileNotFoundError
├── AuthenticationError
├── ValidationError
├── NetworkError
│   ├── ConnectionError
│   ├── ServerUnreachableError
│   ├── RetryExhaustedError
│   └── TimeoutError
└── ResourceNotFoundError

Usage Example:

from xnatctl.core.client import XNATClient
from xnatctl.core.exceptions import (
    AuthenticationError,
    ResourceNotFoundError,
    NetworkError
)

try:
    client = XNATClient(base_url="https://xnat.example.org")
    client.authenticate()
except AuthenticationError:
    print("Invalid credentials")
except NetworkError:
    print("Cannot reach server")
except ResourceNotFoundError as e:
    print(f"Resource not found: {e.resource_type} {e.resource_id}")

Exception hierarchy for xnatctl.

Provides typed exceptions for different failure modes with clear error messages.

exception xnatctl.core.exceptions.XNATCtlError(message, details=None)

Bases: Exception

Base exception for all xnatctl errors.

exception xnatctl.core.exceptions.ConfigurationError(message, field=None, value=None)

Bases: XNATCtlError

Error in configuration (missing, invalid, or malformed).

exception xnatctl.core.exceptions.ProfileNotFoundError(profile)

Bases: ConfigurationError

Requested profile does not exist.

exception xnatctl.core.exceptions.ValidationError(message, field=None, value=None)

Bases: XNATCtlError

Input validation failed.

exception xnatctl.core.exceptions.InvalidURLError(url, reason='')

Bases: ValidationError

Invalid URL format.

exception xnatctl.core.exceptions.InvalidPortError(port)

Bases: ValidationError

Invalid port number.

exception xnatctl.core.exceptions.InvalidIdentifierError(identifier_type, value, reason='')

Bases: ValidationError

Invalid XNAT identifier (project, subject, session, scan).

exception xnatctl.core.exceptions.PathValidationError(path, reason)

Bases: ValidationError

Path validation failed.

exception xnatctl.core.exceptions.ConnectionError(message, url=None)

Bases: XNATCtlError

Base class for connection-related errors.

exception xnatctl.core.exceptions.NetworkError(url, cause=None)

Bases: ConnectionError

Network-level error (DNS, TCP, TLS).

exception xnatctl.core.exceptions.ServerUnreachableError(url)

Bases: ConnectionError

Server is not reachable.

exception xnatctl.core.exceptions.TimeoutError(url, timeout)

Bases: ConnectionError

Request timed out.

exception xnatctl.core.exceptions.RetryExhaustedError(operation, attempts, last_error=None)

Bases: ConnectionError

All retry attempts failed.

exception xnatctl.core.exceptions.AuthenticationError(url=None, reason='')

Bases: XNATCtlError

Authentication failed.

exception xnatctl.core.exceptions.SessionExpiredError(url=None)

Bases: AuthenticationError

Session has expired.

exception xnatctl.core.exceptions.PermissionDeniedError(resource, operation='access', url=None)

Bases: AuthenticationError

User lacks permission for the requested operation.

exception xnatctl.core.exceptions.ResourceError(message, resource_type=None, resource_id=None)

Bases: XNATCtlError

Error related to XNAT resources.

exception xnatctl.core.exceptions.ResourceNotFoundError(resource_type, resource_id)

Bases: ResourceError

Requested resource does not exist.

exception xnatctl.core.exceptions.ResourceExistsError(resource_type, resource_id)

Bases: ResourceError

Resource already exists.

exception xnatctl.core.exceptions.OperationError(operation, message, details=None)

Bases: XNATCtlError

Error during an operation.

exception xnatctl.core.exceptions.UploadError(message, file_path=None, details=None)

Bases: OperationError

Error during upload.

exception xnatctl.core.exceptions.DownloadError(message, resource=None, details=None)

Bases: OperationError

Error during download.

exception xnatctl.core.exceptions.BatchOperationError(operation, succeeded, failed, errors)

Bases: OperationError

Error in batch operation with partial success.

exception xnatctl.core.exceptions.DicomError(message, file_path=None)

Bases: XNATCtlError

Error related to DICOM operations.

exception xnatctl.core.exceptions.DicomParseError(file_path, reason='')

Bases: DicomError

Failed to parse DICOM file.

exception xnatctl.core.exceptions.DicomStoreError(message, host=None, port=None)

Bases: DicomError

DICOM C-STORE operation failed.

exception xnatctl.core.exceptions.TransferError(message, details=None)

Bases: OperationError

Error during project transfer.

exception xnatctl.core.exceptions.TransferConflictError(entity_type, local_id, remote_id, reason)

Bases: TransferError

Conflict detected on destination during transfer.

exception xnatctl.core.exceptions.TransferCircuitBreakerError(failures, max_failures)

Bases: TransferError

Too many consecutive transfer failures.

exception xnatctl.core.exceptions.TransferVerificationError(entity_id, expected, actual)

Bases: TransferError

Post-transfer verification failed.

exception xnatctl.core.exceptions.TransferConfigError(message, field=None)

Bases: TransferError

Invalid transfer configuration.

Validation

Input validation utilities for XNAT resource identifiers, URLs, and parameters.

Validators:

• validate_server_url(url: str) -> str - Normalize and validate XNAT server URLs

• validate_project_id(project_id: str) -> None - Validate project ID format

• validate_subject_label(label: str) -> None - Validate subject label format

• validate_session_label(label: str) -> None - Validate session label format

Input validation module for xnatctl.

Provides comprehensive validation for URLs, ports, identifiers, paths, and DICOM-specific values.

xnatctl.core.validation.validate_server_url(url)

Validate XNAT server URL and return normalized form.

Parameters:

url (str) – Server URL to validate.

Returns:

Normalized URL (trailing slash removed).

Raises:

InvalidURLError – If URL is malformed or uses unsupported scheme.

Return type:

str

xnatctl.core.validation.validate_url_or_none(url)

Validate URL if provided, or return None.

xnatctl.core.validation.validate_port(port, allow_none=False)

Validate port number.

Parameters:

• port (int | str | None) – Port number to validate.

• allow_none (bool) – If True, None is a valid value.

Returns:

Validated port number or None.

Raises:

InvalidPortError – If port is invalid.

Return type:

int | None

xnatctl.core.validation.validate_xnat_identifier(value, identifier_type='identifier', *, allow_empty=False, max_length=64)

Validate an XNAT identifier (project, subject, session, scan ID).

Parameters:

• value (str) – Identifier value to validate.

• identifier_type (str) – Type name for error messages.

• allow_empty (bool) – If True, empty string is valid.

• max_length (int) – Maximum allowed length.

Returns:

Validated and stripped identifier.

Raises:

InvalidIdentifierError – If identifier is invalid.

Return type:

str

xnatctl.core.validation.validate_project_id(project)

Validate XNAT project ID.

xnatctl.core.validation.validate_subject_id(subject)

Validate XNAT subject ID.

xnatctl.core.validation.validate_session_id(session)

Validate XNAT session/experiment ID.

xnatctl.core.validation.validate_scan_id(scan_id)

Validate XNAT scan ID (typically numeric but XNAT allows strings).

xnatctl.core.validation.validate_resource_label(label)

Validate XNAT resource label (more flexible than other identifiers).

xnatctl.core.validation.validate_ae_title(ae_title, field_name='AE Title')

Validate DICOM Application Entity Title.

Per DICOM standard: 1-16 printable ASCII characters, no backslash.

xnatctl.core.validation.validate_path_exists(path, *, must_be_file=False, must_be_dir=False, description='path')

Validate that a path exists and optionally check its type.

Parameters:

• path (str | Path) – Path to validate.

• must_be_file (bool) – If True, path must be a file.

• must_be_dir (bool) – If True, path must be a directory.

• description (str) – Description for error messages.

Returns:

Resolved Path object.

Raises:

PathValidationError – If path is invalid or doesn’t meet requirements.

Return type:

Path

xnatctl.core.validation.validate_path_writable(path, description='path')

Validate that a path is writable (parent directory exists and is writable).

Parameters:

• path (str | Path) – Path to validate.

• description (str) – Description for error messages.

Returns:

Resolved Path object.

Raises:

PathValidationError – If path is not writable.

Return type:

Path

xnatctl.core.validation.validate_archive_path(path)

Validate that path is a supported archive file.

Parameters:

path (str | Path) – Path to archive file.

Returns:

Resolved Path object.

Raises:

PathValidationError – If path is not a valid archive.

Return type:

Path

xnatctl.core.validation.validate_dicom_directory(path)

Validate that path is a directory suitable for DICOM files.

xnatctl.core.validation.validate_timeout(value, field_name='timeout', *, min_value=1, max_value=2592000, default=21600)

Validate timeout value in seconds.

Parameters:

• value (int | float | str | None) – Timeout value to validate.

• field_name (str) – Field name for error messages.

• min_value (int) – Minimum allowed value.

• max_value (int) – Maximum allowed value.

• default (int) – Default value if None.

Returns:

Validated timeout in seconds.

Raises:

ConfigurationError – If timeout is invalid.

Return type:

int

xnatctl.core.validation.validate_workers(value, field_name='workers', *, min_value=1, max_value=100, default=4)

Validate worker count for parallel operations.

Parameters:

• value (int | str | None) – Worker count to validate.

• field_name (str) – Field name for error messages.

• min_value (int) – Minimum allowed value.

• max_value (int) – Maximum allowed value.

• default (int) – Default value if None.

Returns:

Validated worker count.

Raises:

ConfigurationError – If value is invalid.

Return type:

int

xnatctl.core.validation.validate_regex_pattern(pattern, field_name='pattern')

Validate and compile a regex pattern.

Parameters:

• pattern (str) – Regex pattern string.

• field_name (str) – Field name for error messages.

Returns:

Compiled regex pattern.

Raises:

ConfigurationError – If pattern is invalid.

Return type:

Pattern[str]

xnatctl.core.validation.validate_scan_ids_input(scan_input)

Validate and parse scan IDs input from CLI.

Accepts: - “*” for all scans (returns None) - Comma-separated list: “1,2,3,4” - Single ID: “1”

Parameters:

scan_input (str) – Raw scan IDs input string.

Returns:

List of scan IDs or None for all scans.

Raises:

InvalidIdentifierError – If any scan ID is invalid.

Return type:

list[str] | None

xnatctl.core.validation.validate_project_list(projects_input)

Validate and parse comma-separated project IDs.

Parameters:

projects_input (str) – Comma-separated project IDs.

Returns:

List of validated project IDs.

Raises:

InvalidIdentifierError – If any project ID is invalid.

Return type:

list[str]

Output

Output formatting utilities for JSON, table, and quiet modes. Uses Rich for terminal rendering.

Supported Formats:

• json - Machine-readable JSON output

• table - Human-readable table with borders and alignment

• quiet - Minimal output (IDs only)

Usage Example:

from xnatctl.core.output import OutputFormatter

formatter = OutputFormatter(format="table")
formatter.print_table(
    data=[{"id": "proj1", "name": "Project 1"}],
    columns=["id", "name"]
)

Output formatting for xnatctl.

Provides consistent output in JSON, table, and quiet modes using Rich.

class xnatctl.core.output.OutputFormat(*values)

Output format options.

JSON = 'json'

TABLE = 'table'

classmethod from_string(value)

Create from string value.

xnatctl.core.output.print_table(rows, columns, *, title=None, column_labels=None)

Print data as a Rich table.

Parameters:

• rows (Sequence[dict[str, Any]]) – List of dictionaries with data.

• columns (Sequence[str]) – Column keys to display.

• title (str | None) – Optional table title.

• column_labels (dict[str, str] | None) – Optional mapping of column keys to display labels.

xnatctl.core.output.print_key_value(data, *, title=None, key_labels=None)

Print key-value pairs in a formatted way.

Parameters:

• data (dict[str, Any]) – Dictionary of key-value pairs.

• title (str | None) – Optional title.

• key_labels (dict[str, str] | None) – Optional mapping of keys to display labels.

xnatctl.core.output.print_json(data, *, indent=2)

Print data as JSON.

Parameters:

• data (Any) – Data to print.

• indent (int) – Indentation level.

xnatctl.core.output.print_output(data, *, format=OutputFormat.TABLE, columns=None, column_labels=None, title=None, quiet=False, id_field='id')

Print data in the specified format.

Parameters:

• data (Any) – Data to print (dict, list, or scalar).

• format (OutputFormat) – Output format.

• columns (Sequence[str] | None) – Columns for table format.

• column_labels (dict[str, str] | None) – Labels for columns.

• title (str | None) – Optional title.

• quiet (bool) – If True, only print IDs.

• id_field (str) – Field to use for IDs in quiet mode.

xnatctl.core.output.print_error(message)

Print error message to stderr.

The message is routed through redact_url_query() so that URLs embedded in the error never leak secret-shaped query values.

xnatctl.core.output.print_warning(message)

Print warning message to stderr.

The message is routed through redact_url_query() so that URLs embedded in the warning never leak secret-shaped query values.

xnatctl.core.output.print_success(message)

Print success message.

xnatctl.core.output.print_info(message)

Print info message.

xnatctl.core.output.create_progress()

Create a Rich progress bar.

Returns:

Progress instance.

Return type:

Progress

xnatctl.core.output.create_spinner()

Create a spinner for indeterminate progress.

Returns:

Progress instance with spinner only.

Return type:

Progress

Logging

Structured logging utilities with configurable verbosity levels.

Logging utilities for xnatctl.

Provides structured logging with audit trail support.

xnatctl.core.logging.setup_logging(level=20, *, quiet=False, verbose=False)

Configure logging for xnatctl.

Parameters:

• level (int) – Base logging level.

• quiet (bool) – If True, only show errors.

• verbose (bool) – If True, show debug messages.

xnatctl.core.logging.get_logger(name)

Get a logger instance.

Parameters:

name (str) – Logger name (typically __name__).

Returns:

Logger instance.

Return type:

Logger

class xnatctl.core.logging.LogContext(operation, logger=None, **context)

Context manager for structured logging with context fields.

log(level, message, *args)

Log a message with context.

Parameters:

• level (int) – Log level.

• message (str) – Message format string.

• *args (Any) – Format arguments.

info(message, *args)

Log info message.

warning(message, *args)

Log warning message.

error(message, *args)

Log error message.

debug(message, *args)

Log debug message.

xnatctl.core.logging.log_context(operation, logger=None, **context)

Context manager for structured logging.

Parameters:

• operation (str) – Name of the operation.

• logger (Logger | None) – Logger instance.

• **context (Any) – Additional context fields.

Yields:

LogContext instance.

class xnatctl.core.logging.AuditLogger(logger=None)

Logger for audit trail of operations.

log_operation(operation, *, project=None, subject=None, session=None, user=None, success=True, details=None)

Log an auditable operation.

Parameters:

• operation (str) – Name of the operation.

• project (str | None) – Project ID.

• subject (str | None) – Subject ID.

• session (str | None) – Session ID.

• user (str | None) – Username performing the operation.

• success (bool) – Whether operation succeeded.

• details (dict[str, Any] | None) – Additional details.

xnatctl.core.logging.get_audit_logger()

Get the audit logger instance.

Returns:

AuditLogger instance.

Return type:

AuditLogger