Services API

The services package provides high-level Python interfaces for XNAT REST API operations. Each service class encapsulates operations for a specific resource type (projects, subjects, sessions, etc.).

Service Layer Architecture

Design Pattern:

All services follow the service layer pattern:

1. Instantiate with authenticated client

2. Call service methods (list, get, create, update, delete)

3. Receive typed model objects (not raw JSON)

Benefits:

• Type-safe operations with Pydantic models

• Automatic retry and error handling

• Consistent pagination and filtering

• Clean separation of concerns

Common Usage Pattern:

from xnatctl.core.client import XNATClient
from xnatctl.services.projects import ProjectService

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

# Instantiate service
service = ProjectService(client)

# Call service methods
projects = service.list()
project = service.get("MYPROJECT")

Base Service

Foundation class providing common HTTP method wrappers and pagination utilities.

class xnatctl.services.base.BaseService(client)

Bases: object

Base service class with common functionality.

Projects Service

Manage XNAT projects: list, inspect, create, and configure.

Hierarchy:

Project (top-level)
└── Subject
    └── Session
        └── Scan
            └── Resource

Common Operations:

from xnatctl.services.projects import ProjectService

service = ProjectService(client)

# List all accessible projects
projects = service.list()

# Get specific project details
project = service.get("MYPROJECT")
print(f"Name: {project.name}")
print(f"PI: {project.pi}")
print(f"Subjects: {project.subject_count}")

# Create new project
new_project = service.create(
    project_id="NEWPROJECT",
    name="New Project",
    pi_firstname="Jane",
    pi_lastname="Smith",
    accessibility="private"
)

class xnatctl.services.projects.ProjectService(client)

Bases: BaseService

Service for XNAT project operations.

list(accessible=True, limit=None)

List projects.

Parameters:

• accessible (bool) – Only list accessible projects

• limit (int | None) – Maximum number of results

Returns:

List of Project objects

Return type:

list[Project]

get(project_id)

Get project details.

Parameters:

project_id (str) – Project ID

Returns:

Project object

Raises:

ResourceNotFoundError – If project not found

Return type:

Project

create(project_id, name=None, description=None, keywords=None, pi_firstname=None, pi_lastname=None, accessibility='private')

Create a new project.

Parameters:

• project_id (str) – Project ID (must be unique)

• name (str | None) – Project name (defaults to project_id)

• description (str | None) – Project description

• keywords (str | None) – Comma-separated keywords

• pi_firstname (str | None) – PI first name

• pi_lastname (str | None) – PI last name

• accessibility (str) – Access level (private, protected, public)

Returns:

Created Project object

Return type:

Project

delete(project_id, remove_files=False)

Delete a project.

Parameters:

• project_id (str) – Project ID

• remove_files (bool) – Also remove files from filesystem

Returns:

True if successful

Return type:

bool

get_subjects(project_id, limit=None)

Get subjects in a project.

Parameters:

• project_id (str) – Project ID

• limit (int | None) – Maximum number of results

Returns:

List of subject data dicts

Return type:

list[dict[str, Any]]

get_sessions(project_id, limit=None)

Get sessions/experiments in a project.

Parameters:

• project_id (str) – Project ID

• limit (int | None) – Maximum number of results

Returns:

List of session data dicts

Return type:

list[dict[str, Any]]

set_accessibility(project_id, accessibility)

Set project accessibility level.

Parameters:

• project_id (str) – Project ID

• accessibility (str) – Access level (private, protected, public)

Returns:

True if successful

Return type:

bool

Subjects Service

Manage subjects (participants) within projects.

Subject Lifecycle:

1. Create subject in a project

2. Add imaging sessions to subject

3. Optionally rename or delete subject

Common Operations:

from xnatctl.services.subjects import SubjectService

service = SubjectService(client)

# List subjects in a project
subjects = service.list(project="MYPROJECT")

# Get specific subject
subject = service.get(
    project="MYPROJECT",
    subject="SUB001"
)

# Rename subject
service.rename(
    project="MYPROJECT",
    subject="SUB001",
    new_label="PARTICIPANT001"
)

# Delete subject (WARNING: destructive)
service.delete(
    project="MYPROJECT",
    subject="SUB001"
)

class xnatctl.services.subjects.SubjectService(client)

Bases: BaseService

Service for XNAT subject operations.

list(project=None, limit=None, columns=None)

List subjects.

Parameters:

• project (str | None) – Filter by project ID

• limit (int | None) – Maximum number of results

• columns (list[str] | None) – Specific columns to retrieve

Returns:

List of Subject objects

Return type:

list[Subject]

get(subject_id, project=None)

Get subject details.

Parameters:

• subject_id (str) – Subject ID or label

• project (str | None) – Project ID (required if using label)

Returns:

Subject object

Raises:

ResourceNotFoundError – If subject not found

Return type:

Subject

create(project, label, group=None, gender=None, yob=None)

Create a new subject.

Parameters:

• project (str) – Project ID

• label (str) – Subject label

• group (str | None) – Subject group

• gender (str | None) – Gender (male, female, other, unknown)

• yob (int | None) – Year of birth

Returns:

Created Subject object

Return type:

Subject

delete(subject_id, project=None, remove_files=False, force=False)

Delete a subject.

By default, refuses to delete a subject that still has experiments attached (XNAT would cascade-delete the experiments). Pass force=True to override this safety check — but only do so after explicitly reassigning or archiving the experiments yourself.

Parameters:

• subject_id (str) – Subject ID or label.

• project (str | None) – Project ID (required when using a label).

• remove_files (bool) – Also remove files from filesystem.

• force (bool) – Skip the “subject has experiments” safety check. Use with extreme caution; cascade-deletes all attached experiments.

Returns:

True if successful.

Raises:

RuntimeError – If the subject has attached experiments and force=False.

Return type:

bool

rename(subject_id, new_label, project=None)

Rename a subject.

Parameters:

• subject_id (str) – Current subject ID or label

• new_label (str) – New label

• project (str | None) – Project ID

Returns:

Updated Subject object

Return type:

Subject

rename_batch(project, mapping, dry_run=False)

Batch rename subjects using a mapping.

Parameters:

• project (str) – Project ID

• mapping (dict[str, str]) – Dict of old_label -> new_label

• dry_run (bool) – Preview changes without applying

Returns:

Summary dict with renamed, skipped, errors

Return type:

dict[str, Any]

rename_pattern(project, match_pattern, to_template, dry_run=False, merge=False)

Rename subjects matching a pattern.

Parameters:

• project (str) – Project ID

• match_pattern (str) – Regex pattern with capture groups

• to_template (str) – Template using {1}, {2} for groups

• dry_run (bool) – Preview changes without applying

• merge (bool) – Allow merging into existing subjects

Returns:

Summary dict with renamed, merged, skipped, errors

Return type:

dict[str, Any]

get_sessions(subject_id, project=None)

Get sessions for a subject.

Parameters:

• subject_id (str) – Subject ID

• project (str | None) – Project ID

Returns:

List of session data dicts

Return type:

list[dict[str, Any]]

merge_subjects(project, source_label, target_label, dry_run=False)

Merge source subject into target subject.

This moves all experiments/sessions from the source subject to the target subject, then deletes the empty source subject.

Use this when renaming would result in a duplicate - the experiments are consolidated under the target subject.

Parameters:

• project (str) – Project ID

• source_label (str) – Source subject label (will be deleted)

• target_label (str) – Target subject label (will receive experiments)

• dry_run (bool) – Preview changes without applying

Returns:

• experiments_moved: number of experiments moved

• source_deleted: whether source was deleted

• experiments: list of moved experiment IDs

Return type:

Summary dict with

Raises:

ResourceNotFoundError – If source or target not found

Sessions Service

Manage imaging sessions (experiments) containing scan data.

Session Types:

Sessions represent data collection events. Common types include:

• Baseline scans

• Follow-up visits

• Longitudinal timepoints

• Multi-modal acquisitions

Filtering by Modality:

from xnatctl.services.sessions import SessionService

service = SessionService(client)

# List all MR sessions in a project
mr_sessions = service.list(
    project="MYPROJECT",
    modality="MR"
)

# List sessions for a specific subject
subject_sessions = service.list(
    project="MYPROJECT",
    subject="SUB001"
)

# Get session details
session = service.get(
    session_id="XNAT_E00001",
    project="MYPROJECT"
)

print(f"Date: {session.session_date}")
print(f"Scans: {session.scan_count}")
print(f"Scanner: {session.scanner}")

class xnatctl.services.sessions.SessionService(client)

Bases: BaseService

Service for XNAT session/experiment operations.

list(project=None, subject=None, modality=None, limit=None, columns=None)

List sessions/experiments.

Parameters:

• project (str | None) – Filter by project ID

• subject (str | None) – Filter by subject ID

• modality (str | None) – Filter by modality (MR, PET, CT)

• limit (int | None) – Maximum number of results

• columns (list[str] | None) – Specific columns to retrieve

Returns:

List of Session objects

Return type:

list[Session]

get(session_id, project=None)

Get session details.

Parameters:

• session_id (str) – Session ID or label

• project (str | None) – Project ID (helps with label lookup)

Returns:

Session object

Raises:

ResourceNotFoundError – If session not found

Return type:

Session

create(project, subject, label, xsi_type='xnat:mrSessionData', date=None, time=None, visit_id=None, modality=None)

Create a new session/experiment.

Parameters:

• project (str) – Project ID

• subject (str) – Subject ID or label

• label (str) – Session label

• xsi_type (str) – XSI type (xnat:mrSessionData, xnat:petSessionData, etc)

• date (str | None) – Session date (YYYY-MM-DD)

• time (str | None) – Session time (HH:MM:SS)

• visit_id (str | None) – Visit identifier

• modality (str | None) – Modality (overrides xsi_type if provided)

Returns:

Created Session object

Return type:

Session

delete(session_id, project=None, remove_files=False)

Delete a session.

Parameters:

• session_id (str) – Session ID

• project (str | None) – Project ID

• remove_files (bool) – Also remove files from filesystem

Returns:

True if successful

Return type:

bool

get_scans(session_id, project=None)

Get scans for a session.

Parameters:

• session_id (str) – Session ID

• project (str | None) – Project ID

Returns:

List of scan data dicts

Return type:

list[dict[str, Any]]

get_resources(session_id, project=None)

Get resources for a session.

Parameters:

• session_id (str) – Session ID

• project (str | None) – Project ID

Returns:

List of resource data dicts

Return type:

list[dict[str, Any]]

set_field(session_id, field, value, project=None)

Set a field value on a session.

Parameters:

• session_id (str) – Session ID

• field (str) – Field name (e.g., ‘note’, ‘acquisition_site’)

• value (str) – Field value

• project (str | None) – Project ID

Returns:

True if successful

Return type:

bool

share(session_id, target_project, label=None, primary=False)

Share a session with another project.

Parameters:

• session_id (str) – Session ID

• target_project (str) – Target project ID

• label (str | None) – New label in target project

• primary (bool) – Make target the primary project

Returns:

True if successful

Return type:

bool

Scans Service

Manage individual imaging scans (series) within sessions.

Scan Operations:

from xnatctl.services.scans import ScanService

service = ScanService(client)

# List scans in a session
scans = service.list(
    project="MYPROJECT",
    session="SESSION01"
)

for scan in scans:
    print(f"{scan.scan_id}: {scan.type}")
    print(f"  Quality: {scan.quality}")
    print(f"  Files: {scan.file_count}")

# Delete a scan
service.delete(
    project="MYPROJECT",
    session="SESSION01",
    scan="1"
)

class xnatctl.services.scans.ScanService(client)

Bases: BaseService

Service for XNAT scan operations.

list(session_id, project=None, columns=None)

List scans in a session.

Parameters:

• session_id (str) – Session ID

• project (str | None) – Project ID (optional)

• columns (list[str] | None) – Specific columns to retrieve

Returns:

List of Scan objects

Return type:

list[Scan]

get(session_id, scan_id, project=None)

Get scan details.

Parameters:

• session_id (str) – Session ID

• scan_id (str) – Scan ID

• project (str | None) – Project ID (optional)

Returns:

Scan object

Raises:

ResourceNotFoundError – If scan not found

Return type:

Scan

delete(session_id, scan_id, project=None, remove_files=False)

Delete a scan.

Parameters:

• session_id (str) – Session ID

• scan_id (str) – Scan ID

• project (str | None) – Project ID

• remove_files (bool) – Also remove files from filesystem

Returns:

True if successful

Return type:

bool

delete_multiple(session_id, scan_ids, project=None, remove_files=False, parallel=True, workers=4, progress_callback=None)

Delete multiple scans.

Parameters:

• session_id (str) – Session ID

• scan_ids (list[str]) – List of scan IDs to delete (“*” for all)

• project (str | None) – Project ID

• remove_files (bool) – Also remove files

• parallel (bool) – Use parallel deletion

• workers (int) – Number of parallel workers

• progress_callback (Callable[[int, int, str], None] | None) – Callback(current, total, scan_id)

Returns:

Summary dict with deleted, failed, errors

Return type:

dict[str, Any]

get_resources(session_id, scan_id, project=None)

Get resources for a scan.

Parameters:

• session_id (str) – Session ID

• scan_id (str) – Scan ID

• project (str | None) – Project ID

Returns:

List of resource data dicts

Return type:

list[dict[str, Any]]

set_quality(session_id, scan_id, quality, project=None)

Set scan quality assessment.

Parameters:

• session_id (str) – Session ID

• scan_id (str) – Scan ID

• quality (str) – Quality value (usable, questionable, unusable)

• project (str | None) – Project ID

Returns:

True if successful

Return type:

bool

set_note(session_id, scan_id, note, project=None)

Set scan note.

Parameters:

• session_id (str) – Session ID

• scan_id (str) – Scan ID

• note (str) – Note text

• project (str | None) – Project ID

Returns:

True if successful

Return type:

bool

Resources Service

Manage file resources attached to XNAT objects.

Resource Types:

Common resource categories:

• DICOM - Raw DICOM files

• NIFTI - Converted NIfTI volumes

• SNAPSHOTS - Preview images

• QC - Quality control reports

• PROCESSED - Analysis outputs

Upload and Download:

from xnatctl.services.resources import ResourceService

service = ResourceService(client)

# List resources
resources = service.list(
    project="MYPROJECT",
    session="SESSION01"
)

# Upload files to a resource
service.upload(
    project="MYPROJECT",
    session="SESSION01",
    resource="PROCESSED",
    files=["analysis.nii.gz", "report.pdf"]
)

# Download resource
service.download(
    project="MYPROJECT",
    session="SESSION01",
    resource="DICOM",
    dest="./downloads/"
)

class xnatctl.services.resources.ResourceService(client)

Bases: BaseService

Service for XNAT resource operations.

list(session_id, scan_id=None, project=None)

List resources for a session or scan.

Parameters:

• session_id (str) – Session ID

• scan_id (str | None) – Scan ID (for scan-level resources)

• project (str | None) – Project ID

Returns:

List of Resource objects

Return type:

list[Resource]

get(session_id, resource_label, scan_id=None, project=None)

Get resource details.

Parameters:

• session_id (str) – Session ID

• resource_label (str) – Resource label

• scan_id (str | None) – Scan ID (for scan-level resources)

• project (str | None) – Project ID

Returns:

Resource object

Raises:

ResourceNotFoundError – If resource not found

Return type:

Resource

list_files(session_id, resource_label, scan_id=None, project=None)

List files in a resource.

Parameters:

• session_id (str) – Session ID

• resource_label (str) – Resource label

• scan_id (str | None) – Scan ID (for scan-level resources)

• project (str | None) – Project ID

Returns:

List of ResourceFile objects

Return type:

list[ResourceFile]

create(session_id, resource_label, scan_id=None, project=None, format=None, content=None)

Create a new resource.

Parameters:

• session_id (str) – Session ID

• resource_label (str) – Resource label

• scan_id (str | None) – Scan ID (for scan-level resources)

• project (str | None) – Project ID

• format (str | None) – Resource format

• content (str | None) – Content type

Returns:

Created Resource object

Return type:

Resource

delete(session_id, resource_label, scan_id=None, project=None, remove_files=True)

Delete a resource.

Parameters:

• session_id (str) – Session ID

• resource_label (str) – Resource label

• scan_id (str | None) – Scan ID (for scan-level resources)

• project (str | None) – Project ID

• remove_files (bool) – Also remove files from filesystem

Returns:

True if successful

Return type:

bool

upload_file(session_id, resource_label, file_path, scan_id=None, project=None, extract=False, overwrite=False)

Upload a file to a resource.

Parameters:

• session_id (str) – Session ID

• resource_label (str) – Resource label

• file_path (Path) – Local file path

• scan_id (str | None) – Scan ID (for scan-level resources)

• project (str | None) – Project ID

• extract (bool) – Extract ZIP/TAR files after upload

• overwrite (bool) – Overwrite existing files

Returns:

Upload result dict

Return type:

dict[str, object]

upload_directory(session_id, resource_label, directory_path, scan_id=None, project=None, overwrite=False)

Upload a directory to a resource (creates ZIP first).

Parameters:

• session_id (str) – Session ID

• resource_label (str) – Resource label

• directory_path (Path) – Local directory path

• scan_id (str | None) – Scan ID (for scan-level resources)

• project (str | None) – Project ID

• overwrite (bool) – Overwrite existing files

Returns:

Upload result dict

Return type:

dict[str, object]

Downloads Service

High-performance parallel download operations with resume support.

Features:

• Multi-threaded parallel downloads

• Resume support for interrupted transfers

• Progress tracking with Rich progress bars

• Automatic retry on transient failures

• Checksum verification

Parallel Download Example:

from xnatctl.services.downloads import DownloadService

service = DownloadService(client)

# Download with 8 parallel workers
service.download_session(
    project="MYPROJECT",
    session="SESSION01",
    dest="./data/",
    workers=8,
    show_progress=True
)

Download service for XNAT download operations.

class xnatctl.services.downloads.DownloadService(client)

Service for XNAT download operations.

download_session(session_id, output_dir, project=None, include_resources=True, include_assessors=False, pattern=None, resume=False, verify=False, parallel=True, workers=4, progress_callback=None)

Download session data.

Parameters:

• session_id (str) – Session ID

• output_dir (Path) – Output directory path

• project (str | None) – Project ID

• include_resources (bool) – Include session-level resources

• include_assessors (bool) – Include assessor data

• pattern (str | None) – File pattern filter

• resume (bool) – Resume interrupted download

• verify (bool) – Verify checksums after download

• parallel (bool) – Use parallel downloads

• workers (int) – Number of parallel workers

• progress_callback (Callable[[DownloadProgress], None] | None) – Progress callback function

Returns:

DownloadSummary with results

Return type:

DownloadSummary

download_resource(session_id, resource_label, output_dir, scan_id=None, project=None, extract=False, zip_filename=None, progress_callback=None)

Download a specific resource.

Parameters:

• session_id (str) – Session ID

• resource_label (str) – Resource label

• output_dir (Path) – Output directory

• scan_id (str | None) – Scan ID (for scan-level resources)

• project (str | None) – Project ID

• extract (bool) – Extract ZIP files (default: False)

• zip_filename (str | None) – Custom ZIP filename (default: {resource_label}.zip)

• progress_callback (Callable[[DownloadProgress], None] | None) – Progress callback

Returns:

DownloadSummary with results

Return type:

DownloadSummary

download_scan(session_id, scan_id, output_dir, project=None, resource=None, progress_callback=None)

Download a specific scan.

Parameters:

• session_id (str) – Session ID

• scan_id (str) – Scan ID

• output_dir (Path) – Output directory

• project (str | None) – Project ID

• resource (str | None) – Resource type to download (None = all resources)

• progress_callback (Callable[[DownloadProgress], None] | None) – Progress callback

Returns:

DownloadSummary with results

Return type:

DownloadSummary

download_scans(session_id, scan_ids, output_dir, project=None, subject=None, resource=None, zip_filename=None, extract=False, cleanup=True, progress_callback=None)

Download multiple scans in a single request.

Uses XNAT’s comma-separated scan ID feature for efficient batch downloads. When resource is None, downloads ALL files (DICOM + SNAPSHOTS).

Parameters:

• session_id (str) – Session ID or label

• scan_ids (list[str]) – List of scan IDs (or [“ALL”] for all scans)

• output_dir (Path) – Output directory

• project (str | None) – Project ID (required when using session label)

• subject (str | None) – Subject ID/label (optional, narrows experiment lookup)

• resource (str | None) – Resource type (None = all resources, “DICOM” = DICOM only)

• zip_filename (str | None) – Output ZIP filename (default: scans.zip)

• extract (bool) – Extract ZIP after download

• cleanup (bool) – Remove ZIP after successful extraction (with extract=True)

• progress_callback (Callable[[DownloadProgress], None] | None) – Progress callback

Returns:

DownloadSummary with results

Return type:

DownloadSummary

Uploads Service

High-performance parallel upload operations for DICOM and file resources.

Upload Strategies:

xnatctl supports two DICOM upload strategies:

1. Gradual DICOM (default): REST API upload with parallel workers

2. Prearchive: Upload to staging area for review before archiving

Features:

• Multi-threaded parallel uploads

• Automatic retry with exponential backoff

• Progress tracking

• Sequential fallback for failed files

• Thread-local HTTP clients for stability

Parallel Upload Example:

from xnatctl.services.uploads import UploadService

service = UploadService(client)

# Upload DICOM files with 8 workers
service.upload_dicom(
    project="MYPROJECT",
    subject="SUB001",
    session="SESSION01",
    files=["scan001.dcm", "scan002.dcm", ...],
    workers=8,
    show_progress=True
)

Error Handling:

Failed uploads are automatically retried at lower concurrency, with a final sequential retry pass to maximize completion rate on flaky networks.

Upload service for XNAT upload operations.

Provides UploadService with methods for all upload transports: - REST batch upload (simple ZIP batches via import service) - Parallel REST upload (batched archives with parallel workers) - DICOM C-STORE upload (pynetdicom-based network transfer) - Resource upload (file/directory upload to session resources)

Public utility functions (collect_dicom_files, split_into_batches, etc.) are available for direct import and testing.

class xnatctl.services.uploads.DICOMStoreSummary(total_files, sent, failed, log_dir, workspace, success)

Summary of a DICOM C-STORE operation.

total_files: int

sent: int

failed: int

log_dir: Path

workspace: Path

success: bool

xnatctl.services.uploads.collect_dicom_files(root, *, include_extensionless=True)

Recursively collect DICOM-like files under a root directory.

Parameters:

• root (Path) – Root directory to search.

• include_extensionless (bool) – If True, include files without extensions (common for raw DICOM from scanners).

Returns:

Sorted list of file paths.

Raises:

ValueError – If root is not a directory.

Return type:

list[Path]

xnatctl.services.uploads.archive_destination_params(project, direct_archive)

Return the querystring keys that route a POST /data/services/import.

• Direct-archive path: Direct-Archive=true — handled by the DICOM-zip and gradual-DICOM import handlers; bypasses the prearchive and writes straight to the project archive.

• Prearchive path: dest=/prearchive/projects/{project} — the documented destination form. Direct-Archive=false alone is equivalent to “use standard upload mechanism”; we prefer the explicit dest because it is self-describing and matches the PrearchiveService pattern used elsewhere in this repo.

Caveat: neither form can prevent a project-configured auto-archive. XNAT’s prearchive_code on the project (0=manual, 4/5=auto) is the authoritative switch. When a project has auto-archive enabled, a session uploaded via either of these paths will land in prearchive momentarily then be auto-archived by the server. To force prearchive-only behaviour, the project’s prearchive setting must be changed to “Leave in prearchive” (prearchive_code=0). There is no per-upload import-service override for this on XNAT 1.8+.

xnatctl.services.uploads.split_into_batches(files, batch_size)

Split files into batches of specified size.

Parameters:

• files (Sequence[Path]) – Sequence of file paths to split.

• batch_size (int) – Maximum files per batch.

Returns:

List of batches, each batch being a list of paths.

Return type:

list[list[Path]]

xnatctl.services.uploads.split_into_n_batches(files, num_batches)

Split files into N roughly equal batches using round-robin.

Parameters:

• files (Sequence[Path]) – Sequence of file paths to split.

• num_batches (int) – Number of batches to create.

Returns:

List of batches, each batch being a list of paths.

Return type:

list[list[Path]]

xnatctl.services.uploads.is_retryable_status(status_code)

Check if an HTTP status code warrants a retry.

Retryable: 400 (XNAT transient), 429 (rate limit), 5xx (server errors). Non-retryable: 2xx (success), 401/403 (auth), other 4xx (client error).

xnatctl.services.uploads.upload_with_retry(upload_fn, *, max_retries=5, backoff_base=2, label='upload')

Execute an upload function with retry on transient HTTP errors.

Parameters:

• upload_fn (Callable[[], Response]) – Callable that performs the upload and returns an httpx.Response. Will be called multiple times on retry – must be idempotent.

• max_retries (int) – Maximum number of retries (default: 5).

• backoff_base (int) – Base for exponential backoff in seconds (default: 2).

• label (str) – Label for log messages.

Returns:

The httpx.Response from a successful attempt.

Raises:

The last exception if all retries are exhausted and no response was obtained. –

Return type:

Response

class xnatctl.services.uploads.UploadService(client)

Service for XNAT upload operations.

Provides methods for all upload transports: REST batch, parallel REST, DICOM C-STORE, and resource uploads.

upload_dicom(project, subject, session, source_path, overwrite=False, quarantine=False, batch_size=500, parallel=True, workers=4, progress_callback=None)

Upload DICOM files via simple REST batch (ZIP per batch).

Parameters:

• project (str) – Project ID.

• subject (str) – Subject label.

• session (str) – Session label.

• source_path (Path) – Path to DICOM files (directory or ZIP).

• overwrite (bool) – Overwrite existing scans.

• quarantine (bool) – Send to prearchive instead.

• batch_size (int) – Files per upload batch.

• parallel (bool) – Use parallel uploads.

• workers (int) – Number of parallel workers.

• progress_callback (Callable[[UploadProgress], None] | None) – Progress callback function.

Returns:

UploadSummary with results.

Return type:

UploadSummary

upload_dicom_parallel(source_dir, project, subject, session, *, username=None, password=None, upload_workers=4, archive_workers=4, archive_format='tar', import_handler='DICOM-zip', ignore_unparsable=True, overwrite='delete', direct_archive=True, timeout=21600, progress_callback=None)

Upload DICOM files using parallel batched archives via REST import.

High-throughput upload that: 1. Collects DICOM files from the source directory 2. Splits files into N batches (N = upload_workers) 3. Creates archives in parallel 4. Uploads archives in parallel with per-thread HTTP clients

Parameters:

• source_dir (Path) – Directory containing DICOM files.

• project (str) – Target project ID.

• subject (str) – Target subject label.

• session (str) – Target session label.

• username (str | None) – XNAT username (override for per-thread auth).

• password (str | None) – XNAT password (override for per-thread auth).

• upload_workers (int) – Parallel upload workers (default: 4).

• archive_workers (int) – Parallel archive workers (default: 4).

• archive_format (str) – Archive format, “tar” or “zip” (default: tar).

• import_handler (str) – XNAT import handler (default: DICOM-zip).

• ignore_unparsable (bool) – Skip unparsable DICOM files (default: True).

• overwrite (str) – Overwrite mode: none, append, delete (default: delete).

• direct_archive (bool) – Use direct archive vs prearchive (default: True).

• timeout (int) – HTTP timeout in seconds.

• progress_callback (Callable[[UploadProgress], None] | None) – Optional callback for progress updates.

Returns:

UploadSummary with results.

Return type:

UploadSummary

upload_dicom_store(dicom_root, host, called_aet, *, port=104, calling_aet='XNATCTL', workers=4, cleanup=True)

Send DICOM files to an SCP using C-STORE.

This method: 1. Verifies connectivity with C-ECHO 2. Collects DICOM files from the root directory 3. Splits files into batches for parallel associations 4. Sends files using multiple concurrent C-STORE associations

Parameters:

• dicom_root (Path) – Directory containing DICOM files.

• host (str) – DICOM SCP host.

• called_aet (str) – Remote AE title.

• port (int) – DICOM SCP port (default: 104).

• calling_aet (str) – Our AE title (default: XNATCTL).

• workers (int) – Number of parallel associations (default: 4).

• cleanup (bool) – Remove temporary workspace on completion (default: True).

Returns:

DICOMStoreSummary with results.

Raises:

• ImportError – If pydicom/pynetdicom are not installed.

• ValueError – If dicom_root is not a directory.

• RuntimeError – If C-ECHO fails or no DICOM files found.

Return type:

DICOMStoreSummary

upload_dicom_gradual(source_path, project, subject, session, *, workers=4, direct_archive=True, progress_callback=None)

Upload DICOM files using the gradual-DICOM handler (parallel per-file).

Each file is uploaded individually to the XNAT import service using the gradual-DICOM handler, which lets XNAT parse each file on ingest. Files are uploaded in parallel using per-thread HTTP clients.

Accepts directories or ZIP archives. ZIP archives are extracted to a temporary directory before upload. Only DICOM-like files are sent: known DICOM extensions plus extensionless files commonly produced by scanners.

Parameters:

• source_path (Path) – Directory or ZIP file containing DICOM files.

• project (str) – Target project ID.

• subject (str) – Target subject label.

• session (str) – Target session label.

• workers (int) – Number of parallel upload workers (default: 4).

• direct_archive (bool) – Use direct archive vs prearchive (default: True).

• progress_callback (Callable[[UploadProgress], None] | None) – Optional callback for progress updates.

Returns:

UploadSummary with results.

Raises:

• ValueError – If source_path is not a directory or ZIP file.

• FileNotFoundError – If source_path does not exist.

Return type:

UploadSummary

upload_dicom_gradual_files(*, files, project, subject, session, workers=4, direct_archive=True, progress_callback=None)

Upload a specific list of DICOM files via the gradual-DICOM handler.

Unlike upload_dicom_gradual(), this method uploads only the files explicitly provided and does not scan any directories.

Parameters:

• files (Sequence[Path]) – Explicit list of files to upload.

• project (str) – Target project ID.

• subject (str) – Target subject label.

• session (str) – Target session label.

• direct_archive (bool) – Use direct archive vs prearchive (default: True).

• workers (int) – Number of parallel upload workers.

• progress_callback (Callable[[UploadProgress], None] | None) – Optional callback for progress updates.

Returns:

UploadSummary with results.

Raises:

• FileNotFoundError – If any provided path does not exist.

• ValueError – If any provided path is not a file.

Return type:

UploadSummary

upload_resource(session_id, resource_label, source_path, scan_id=None, project=None, extract=False, overwrite=False, progress_callback=None)

Upload files to a resource.

Parameters:

• session_id (str) – Session ID.

• resource_label (str) – Resource label.

• source_path (Path) – File or directory to upload.

• scan_id (str | None) – Scan ID (for scan-level resources).

• project (str | None) – Project ID.

• extract (bool) – Extract ZIP/TAR after upload.

• overwrite (bool) – Overwrite existing files.

• progress_callback (Callable[[UploadProgress], None] | None) – Progress callback.

Returns:

UploadSummary with results.

Return type:

UploadSummary

Prearchive Service

Manage the XNAT prearchive staging area for reviewing uploads before archiving.

Prearchive Workflow:

1. Upload DICOM files to prearchive

2. Review session metadata and quality

3. Archive to final location or delete

Operations:

from xnatctl.services.prearchive import PrearchiveService

service = PrearchiveService(client)

# List prearchive sessions
sessions = service.list(project="MYPROJECT")

for session in sessions:
    print(f"{session.name} - {session.status}")
    print(f"  Uploaded: {session.upload_date}")
    print(f"  Scans: {session.scan_count}")

# Archive session from prearchive
service.archive(
    project="MYPROJECT",
    timestamp="20240101_120000",
    session="SESSION01"
)

# Delete prearchive session
service.delete(
    project="MYPROJECT",
    timestamp="20240101_120000",
    session="SESSION01"
)

class xnatctl.services.prearchive.PrearchiveService(client)

Bases: BaseService

Service for XNAT prearchive operations.

list(project=None)

List prearchive sessions.

Parameters:

project (str | None) – Filter by project ID

Returns:

List of prearchive session dicts

Return type:

list[dict[str, Any]]

get(project, timestamp, session_name)

Get prearchive session details.

Parameters:

• project (str) – Project ID

• timestamp (str) – Prearchive timestamp

• session_name (str) – Session name in prearchive

Returns:

Prearchive session dict

Raises:

ResourceNotFoundError – If session not found

Return type:

dict[str, Any]

archive(project, timestamp, session_name, subject=None, experiment_label=None, overwrite=False)

Archive a session from prearchive.

Parameters:

• project (str) – Project ID

• timestamp (str) – Prearchive timestamp

• session_name (str) – Session name in prearchive

• subject (str | None) – Target subject ID (optional, uses DICOM if not provided)

• experiment_label (str | None) – Target session label

• overwrite (bool) – Overwrite existing session data

Returns:

Result dict with archived session info

Return type:

dict[str, Any]

delete(project, timestamp, session_name)

Delete a session from prearchive.

Parameters:

• project (str) – Project ID

• timestamp (str) – Prearchive timestamp

• session_name (str) – Session name in prearchive

Returns:

True if successful

Return type:

bool

rebuild(project, timestamp, session_name)

Rebuild/refresh a prearchive session.

Parameters:

• project (str) – Project ID

• timestamp (str) – Prearchive timestamp

• session_name (str) – Session name in prearchive

Returns:

Result dict

Return type:

dict[str, Any]

move(project, timestamp, session_name, target_project)

Move a prearchive session to another project.

Parameters:

• project (str) – Source project ID

• timestamp (str) – Prearchive timestamp

• session_name (str) – Session name in prearchive

• target_project (str) – Target project ID

Returns:

Result dict

Return type:

dict[str, Any]

get_scans(project, timestamp, session_name)

Get scans from a prearchive session.

Parameters:

• project (str) – Project ID

• timestamp (str) – Prearchive timestamp

• session_name (str) – Session name in prearchive

Returns:

List of scan dicts

Return type:

list[dict[str, Any]]

Pipelines Service

Execute and monitor XNAT processing pipelines.

Pipeline Operations:

from xnatctl.services.pipelines import PipelineService

service = PipelineService(client)

# List available pipelines
pipelines = service.list(project="MYPROJECT")

# Run pipeline on a session
run_id = service.run(
    project="MYPROJECT",
    session="SESSION01",
    pipeline="DicomToNifti"
)

# Check pipeline status
status = service.status(
    project="MYPROJECT",
    run_id=run_id
)
print(f"Status: {status}")

class xnatctl.services.pipelines.PipelineService(client)

Bases: BaseService

Service for XNAT pipeline operations.

list(project=None)

List available pipelines.

Parameters:

project (str | None) – Filter by project ID

Returns:

List of pipeline dicts

Return type:

list[dict[str, Any]]

get(pipeline_name, project=None)

Get pipeline details.

Parameters:

• pipeline_name (str) – Pipeline name

• project (str | None) – Project ID

Returns:

Pipeline details dict

Raises:

ResourceNotFoundError – If pipeline not found

Return type:

dict[str, Any]

run(pipeline_name, experiment_id, project=None, params=None)

Run a pipeline on an experiment.

Parameters:

• pipeline_name (str) – Pipeline name

• experiment_id (str) – Experiment/session ID

• project (str | None) – Project ID

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

Returns:

Job information dict with job ID

Return type:

dict[str, Any]

status(job_id)

Get pipeline job status.

Parameters:

job_id (str) – Job ID

Returns:

Job status dict

Return type:

dict[str, Any]

wait(job_id, timeout=3600, poll_interval=30, progress_callback=None)

Wait for a pipeline job to complete.

Parameters:

• job_id (str) – Job ID

• timeout (int) – Maximum wait time in seconds

• poll_interval (int) – Seconds between status checks

• progress_callback (Callable[[dict[str, Any]], None] | None) – Called with status on each poll

Returns:

Final job status dict

Raises:

OperationError – If job fails or times out

Return type:

dict[str, Any]

cancel(job_id)

Cancel a running pipeline job.

Parameters:

job_id (str) – Job ID

Returns:

True if cancelled successfully

Return type:

bool

list_jobs(experiment_id=None, project=None, status=None, limit=100)

List pipeline jobs.

Parameters:

• experiment_id (str | None) – Filter by experiment

• project (str | None) – Filter by project

• status (str | None) – Filter by status

• limit (int) – Maximum results

Returns:

List of job dicts

Return type:

list[dict[str, Any]]

Admin Service

Administrative operations including catalog refresh, user management, and audit logs.

Admin Operations:

from xnatctl.services.admin import AdminService

service = AdminService(client)

# Refresh project catalogs
service.refresh_catalogs(project="MYPROJECT")

# List users
users = service.list_users()

# View audit logs
logs = service.audit_logs(
    project="MYPROJECT",
    limit=100
)

class xnatctl.services.admin.AdminService(client)

Bases: BaseService

Service for XNAT administrative operations.

refresh_catalogs(project, experiments=None, options=None, limit=None, parallel=True, workers=4, progress_callback=None)

Refresh catalog XMLs for project experiments.

Parameters:

• project (str) – Project ID

• experiments (list[str] | None) – Specific experiment IDs (or all if None)

• options (list[str] | None) – Refresh options (checksum, delete, append, populateStats)

• limit (int | None) – Limit number of experiments

• parallel (bool) – Use parallel execution

• workers (int) – Number of parallel workers

• progress_callback (Callable[[int, int, str], None] | None) – Callback(current, total, experiment_id)

Returns:

Summary dict with refreshed, failed, errors

Return type:

dict[str, Any]

add_user_to_groups(username, groups, projects=None, role='member')

Add a user to XNAT groups.

Parameters:

• username (str) – XNAT username

• groups (list[str]) – Group names to add user to

• projects (list[str] | None) – Project IDs (expands group names per project)

• role (str) – Role (owner, member, collaborator)

Returns:

Summary dict with added, failed, errors

Return type:

dict[str, Any]

remove_user_from_groups(username, groups, projects=None)

Remove a user from XNAT groups.

Parameters:

• username (str) – XNAT username

• groups (list[str]) – Group names to remove user from

• projects (list[str] | None) – Project IDs

Returns:

Summary dict with removed, failed, errors

Return type:

dict[str, Any]

list_users(project=None)

List users.

Parameters:

project (str | None) – Filter by project

Returns:

List of user dicts

Return type:

list[dict[str, Any]]

get_user(username)

Get user details.

Parameters:

username (str) – Username

Returns:

User details dict

Return type:

dict[str, Any]

audit_log(project=None, username=None, action=None, since=None, limit=100)

Get audit log entries.

Parameters:

• project (str | None) – Filter by project

• username (str | None) – Filter by username

• action (str | None) – Filter by action type

• since (str | None) – Time filter (e.g., “7d”, “2024-01-01”)

• limit (int) – Maximum results

Returns:

List of audit log entries

Return type:

list[dict[str, Any]]

get_server_info()

Get XNAT server information.

Returns:

Server info dict with version, build info, etc.

Return type:

dict[str, Any]

get_site_config(key=None)

Get site configuration.

Parameters:

key (str | None) – Specific config key (or all if None)

Returns:

Configuration dict

Return type:

dict[str, Any]

set_site_config(key, value)

Set site configuration value.

Parameters:

• key (str) – Config key

• value (Any) – Config value

Returns:

True if successful

Return type:

bool