"""Prearchive service for XNAT prearchive operations."""
from __future__ import annotations
import builtins
from typing import Any
from urllib.parse import quote
from xnatctl.core.exceptions import ResourceNotFoundError
from .base import BaseService
def _quote_path_segment(value: str) -> str:
"""Encode a single REST path segment for XNAT service URIs."""
return quote(value, safe="").replace(".", "%2E")
[docs]
class PrearchiveService(BaseService):
"""Service for XNAT prearchive operations."""
[docs]
def list(
self,
project: str | None = None,
) -> builtins.list[dict[str, Any]]:
"""List prearchive sessions.
Args:
project: Filter by project ID
Returns:
List of prearchive session dicts
"""
if project:
path = f"/data/prearchive/projects/{_quote_path_segment(project)}"
else:
path = "/data/prearchive"
params = {"format": "json"}
data = self._get(path, params=params)
return self._extract_results(data)
[docs]
def get(
self,
project: str,
timestamp: str,
session_name: str,
) -> dict[str, Any]:
"""Get prearchive session details.
Args:
project: Project ID
timestamp: Prearchive timestamp
session_name: Session name in prearchive
Returns:
Prearchive session dict
Raises:
ResourceNotFoundError: If session not found
"""
path = (
f"/data/prearchive/projects/{_quote_path_segment(project)}"
f"/{_quote_path_segment(timestamp)}/{_quote_path_segment(session_name)}"
)
params = {"format": "json"}
try:
data = self._get(path, params=params)
results = self._extract_results(data)
if results:
return results[0]
raise ResourceNotFoundError(
"prearchive session", f"{project}/{timestamp}/{session_name}"
)
except Exception as e:
if "404" in str(e):
raise ResourceNotFoundError(
"prearchive session",
f"{project}/{timestamp}/{session_name}",
) from e
raise
[docs]
def archive(
self,
project: str,
timestamp: str,
session_name: str,
subject: str | None = None,
experiment_label: str | None = None,
overwrite: bool = False,
) -> dict[str, Any]:
"""Archive a session from prearchive.
Args:
project: Project ID
timestamp: Prearchive timestamp
session_name: Session name in prearchive
subject: Target subject ID (optional, uses DICOM if not provided)
experiment_label: Target session label
overwrite: Overwrite existing session data
Returns:
Result dict with archived session info
"""
encoded_project = _quote_path_segment(project)
encoded_timestamp = _quote_path_segment(timestamp)
encoded_session_name = _quote_path_segment(session_name)
src = f"/prearchive/projects/{encoded_project}/{encoded_timestamp}/{encoded_session_name}"
data: dict[str, Any] = {"src": src}
resolved_subject = subject
if experiment_label and resolved_subject is None:
session = self.get(project, timestamp, session_name)
resolved_subject = session.get("subject")
if not resolved_subject:
raise ValueError("Cannot archive to a specific experiment label without a subject")
if resolved_subject:
dest = (
f"/archive/projects/{encoded_project}"
f"/subjects/{_quote_path_segment(resolved_subject)}"
)
if experiment_label:
dest = f"{dest}/experiments/{_quote_path_segment(experiment_label)}"
data["dest"] = dest
if overwrite:
data["overwrite"] = "delete"
result = self._post("/data/services/archive", data=data)
return {
"success": True,
"project": project,
"session": session_name,
"result": result,
}
[docs]
def delete(
self,
project: str,
timestamp: str,
session_name: str,
) -> bool:
"""Delete a session from prearchive.
Args:
project: Project ID
timestamp: Prearchive timestamp
session_name: Session name in prearchive
Returns:
True if successful
"""
path = (
f"/data/prearchive/projects/{_quote_path_segment(project)}"
f"/{_quote_path_segment(timestamp)}/{_quote_path_segment(session_name)}"
)
return self._delete(path)
[docs]
def rebuild(
self,
project: str,
timestamp: str,
session_name: str,
) -> dict[str, Any]:
"""Rebuild/refresh a prearchive session.
Args:
project: Project ID
timestamp: Prearchive timestamp
session_name: Session name in prearchive
Returns:
Result dict
"""
path = (
f"/data/prearchive/projects/{_quote_path_segment(project)}"
f"/{_quote_path_segment(timestamp)}/{_quote_path_segment(session_name)}"
)
params = {"action": "rebuild"}
result = self._post(path, params=params)
return {
"success": True,
"project": project,
"session": session_name,
"result": result,
}
[docs]
def move(
self,
project: str,
timestamp: str,
session_name: str,
target_project: str,
) -> dict[str, Any]:
"""Move a prearchive session to another project.
Args:
project: Source project ID
timestamp: Prearchive timestamp
session_name: Session name in prearchive
target_project: Target project ID
Returns:
Result dict
"""
path = (
f"/data/prearchive/projects/{_quote_path_segment(project)}"
f"/{_quote_path_segment(timestamp)}/{_quote_path_segment(session_name)}"
)
params = {
"action": "move",
"newProject": target_project,
}
result = self._post(path, params=params)
return {
"success": True,
"source_project": project,
"target_project": target_project,
"session": session_name,
"result": result,
}
[docs]
def get_scans(
self,
project: str,
timestamp: str,
session_name: str,
) -> builtins.list[dict[str, Any]]:
"""Get scans from a prearchive session.
Args:
project: Project ID
timestamp: Prearchive timestamp
session_name: Session name in prearchive
Returns:
List of scan dicts
"""
path = (
f"/data/prearchive/projects/{_quote_path_segment(project)}"
f"/{_quote_path_segment(timestamp)}/{_quote_path_segment(session_name)}/scans"
)
params = {"format": "json"}
data = self._get(path, params=params)
return self._extract_results(data)