API Reference¶

class Deposition:
    """Local deposition session. Created via deposit_init() or deposit_resume()."""

    def __init__(
        self,
        store: SessionStore,
        api_client: ApiClient,
        check_runner: CheckRunnerProtocol,
    ) -> None:
        self._store = store
        self._api_client = api_client
        self._check_runner = check_runner
        self._session = store.get_session()

    @property
    def session_id(self) -> str:
        """Unique ID of the local session."""
        return self._session.session_id

    @property
    def remote_dep_id(self) -> str | None:
        """Remote deposition ID, populated after deposit() is called."""
        return self._session.remote_dep_id

    @property
    def site_base_url(self) -> str | None:
        """Remote deposition site root, populated after deposit() is called."""
        return self._session.site_base_url

    @property
    def site_url(self) -> str | None:
        """Remote deposition site URL, populated after deposit() is called."""
        return self._session.site_url

    def set_experiment_type(self, experiment_type: ExperimentType) -> None:
        """Set or update the experiment type for this deposition."""
        self._store.update_experiment_type(experiment_type)
        self._session.experiment_type = experiment_type

    def set_em_params(
        self,
        em_subtype: EMSubType | None = None,
        coordinates: bool | None = None,
    ) -> None:
        """Set EM-specific parameters for this deposition."""
        self._store.update_em_params(em_subtype, coordinates)
        self._session.em_subtype = em_subtype
        self._session.coordinates = coordinates

    def add_file(self, file_path: str, file_type: FileType) -> str:
        """Register a local file for this deposition.

        Returns:
            A file_id (UUID string) to reference this file in check methods.

        Raises:
            FileNotFoundError: If file_path does not exist.
        """
        path = Path(file_path)
        if not path.exists():
            raise FileNotFoundError(f"File not found: {file_path}")
        stat = path.stat()
        file_mtime = datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc)
        md5 = _md5_of_file(path)
        file_id = str(uuid.uuid4())
        local_file = LocalFile(
            file_id=file_id,
            session_id=self._session.session_id,
            file_path=str(path.resolve()),
            file_type=file_type,
            md5=md5,
            file_mtime=file_mtime,
        )
        self._store.add_file(local_file)
        return file_id

    def has_file(self, file_path: str) -> bool:
        """Check if file has already been registered for this deposition."""
        return self._store.has_file(file_path)

    def remove_file(self, file_id: str) -> None:
        """Remove a file from this local session by its file_id."""
        self._store.remove_file(file_id)

    def set_voxel_values(
        self,
        file_id: str,
        spacing_x: float,
        spacing_y: float,
        spacing_z: float,
        contour: float,
    ) -> None:
        """Set voxel spacing and contour level for a map file."""
        self._store.set_voxel_values(file_id, spacing_x, spacing_y, spacing_z, contour)

    def check_required_files(self) -> CheckReport:
        """Check that the session contains all required files for the experiment type."""
        files = self._store.get_all_files()
        return self._check_runner.check_required_files(files, self._session.experiment_type, self._session.em_subtype)

    def check_mmcif_file(self, file_id: str) -> CheckReport:
        """Check that the file identified by file_id is a valid mmCIF."""
        file = self._store.get_file(file_id)
        return self._check_runner.check_mmcif_file(file)

    def check_mmcif_category(self, file_id: str, category: str) -> CheckReport:
        """Check that the mmCIF file contains the given category."""
        file = self._store.get_file(file_id)
        return self._check_runner.check_mmcif_category(file, category)

    def check_mmcif_field(self, file_id: str, category: str, field: str) -> CheckReport:
        """Check that the mmCIF file contains the given field in the given category."""
        file = self._store.get_file(file_id)
        return self._check_runner.check_mmcif_field(file, category, field)

    def check_file_type(self, file_id: str, file_type: FileType) -> CheckReport:
        """Check that the file matches the expected FileType."""
        file = self._store.get_file(file_id)
        return self._check_runner.check_file_type(file, file_type)

    def deposit(self) -> str:
        """Submit this deposition to the OneDep API.

        Creates a remote deposition, uploads all registered files, and triggers
        processing. Returns immediately without waiting for processing to finish.

        Returns:
            The remote deposition ID (e.g. "D_8000000001").

        Raises:
            ValueError: If experiment_type has not been set.
            ApiError: If any API call fails.
        """
        if self._session.experiment_type is None:
            raise ValueError(
                "experiment_type must be set before calling deposit(). "
                "Use set_experiment_type() or pass experiment_type to deposit_init()."
            )
        if self._session.remote_dep_id is None:
            experiment = Experiment(
                exp_type=self._session.experiment_type,
                coordinates=self._session.coordinates if self._session.coordinates is not None else True,
                subtype=self._session.em_subtype,
            )
            remote_dep = self._api_client.create_deposition(
                email=self._session.email,
                users=self._session.users,
                country=self._session.country,
                experiments=[experiment],
            )
            dep_id = remote_dep.dep_id
            site_base_url = remote_dep.site_base_url or getattr(self._api_client, "site_base_url", None)
            self._store.set_remote_dep_id(
                dep_id,
                site_url=remote_dep.site_url,
                site_base_url=site_base_url,
            )
            self._session.remote_dep_id = dep_id
            self._session.site_base_url = site_base_url
            self._session.site_url = remote_dep.site_url
        else:
            dep_id = self._session.remote_dep_id

        for file in self._store.get_all_files():
            deposited = self._api_client.upload_file(dep_id, file.file_path, file.file_type)
            if file.voxel:
                v = file.voxel
                self._api_client.update_metadata(
                    dep_id,
                    deposited.file_id,
                    spacing_x=v["spacing_x"],
                    spacing_y=v["spacing_y"],
                    spacing_z=v["spacing_z"],
                    contour=v["contour"],
                    description="",
                )

        self._api_client.process(dep_id)
        return dep_id

    def get_status(self) -> DepositStatus | DepositError:
        """Return the current processing status of the remote deposition.

        Raises:
            RuntimeError: If deposit() has not been called yet.
        """
        if self._session.remote_dep_id is None:
            raise RuntimeError(
                "deposit() has not been called yet for this session. "
                "Call deposit() first to obtain a remote deposition ID."
            )
        return self._api_client.get_status(self._session.remote_dep_id)

    def get_experiment_file_types(self) -> list[FileType]:
        """Return the accepted file types for the current experiment type."""
        session = self._session
        if not session or not session.experiment_type:
            return []
        exptype = session.experiment_type.value
        if exptype is None:
            return []
        subschemas = CheckRunner.subschemas
        if exptype not in subschemas:
            return []
        provider = LocalSchemaProvider(DepositConfig().local_schema_cache_dir)
        schema = provider.get_schema(exptype)
        filetypes = schema.get("enum", [])
        return filetypes

    def close(self) -> None:
        """Close the underlying session store connection."""
        self._store.close()

    def __enter__(self) -> Deposition:
        return self

    def __exit__(self, *args: object) -> None:
        self.close()

@dataclass
class DepositConfig:
    """Resolved configuration for talking to a OneDep deposition site.

    Built by :meth:`load`, which resolves fields from three sources in order
    of increasing priority: the on-disk config file (``config_path``),
    environment variables, and keyword argument overrides. See ``load`` for
    the full resolution order, including how ``access_token`` and
    ``refresh_token`` are populated from the ``[auths.<fqdn>]`` section.

    Attributes:
        access_token: Current OAuth access token, or None if not authenticated.
        refresh_token: Current OAuth refresh token, or None if not authenticated.
        hostname: Base URL of the OneDep deposition site to talk to.
        ssl_verify: Whether to verify TLS certificates on API requests.
        redirect: Whether to follow site-redirect responses from the API.
        allowed_redirect_domain: Domain suffix that redirect targets must belong to.
        fetch_local_schema: Whether to use the bundled local JSON schemas instead
            of fetching them from schema_base_url.
        local_schema_cache_dir: Directory containing the bundled local JSON schemas.
        schema_base_url: Base URL to fetch remote JSON schemas from.
        schema_cache_dir: Local directory used to cache remote JSON schemas.
        session_dir: Local directory used to store deposition session state.
        config_path: Path to the TOML config file read and written by this class.
    """

    access_token: str | None = None
    refresh_token: str | None = None
    hostname: str = "https://deposit.wwpdb.org/deposition"
    ssl_verify: bool = True
    redirect: bool = True
    allowed_redirect_domain: str = "wwpdb.org"
    fetch_local_schema: bool = True
    local_schema_cache_dir: Path = field(default_factory=lambda: Path(__file__).parent / "schemas" / "json")
    schema_base_url: str = "https://schemas.wwpdb.org/nextdep"
    schema_cache_dir: Path = field(default_factory=lambda: Path.home() / ".onedep" / "schemas")
    session_dir: Path = field(default_factory=lambda: Path.home() / ".onedep" / "sessions")
    config_path: Path = field(default_factory=lambda: Path.home() / ".config" / "onedep" / "config.toml")

    @staticmethod
    def _load_toml_file(path: Path) -> dict:
        if not path.exists():
            return {}
        try:
            with path.open("rb") as fp:
                return tomllib.load(fp)
        except tomllib.TOMLDecodeError as exc:
            raise ConfigError(f"Failed to parse {path}: {exc}") from exc

    @classmethod
    def load(cls, **overrides: object) -> DepositConfig:
        """Resolve a DepositConfig from the config file, environment, and overrides.

        Resolution order (lowest to highest priority):

        1. The TOML config file at ``config_path`` (default
           ``~/.config/onedep/config.toml``), read from its ``[default]`` section.
        2. Environment variables (``ONEDEP_ACCESS_TOKEN``, ``ONEDEP_REFRESH_TOKEN``,
           ``ONEDEP_HOSTNAME``, ``ONEDEP_SSL_VERIFY``, ``ONEDEP_REDIRECT``,
           ``ONEDEP_SCHEMA_URL``).
        3. Keyword arguments passed to this method.

        After resolving ``hostname``, also reads the matching ``[auths.<fqdn>]``
        section from the config file and populates ``access_token`` and
        ``refresh_token`` from it -- unless either token was explicitly set
        via an environment variable (``ONEDEP_ACCESS_TOKEN`` or
        ``ONEDEP_REFRESH_TOKEN``) or passed as a keyword argument, in which
        case file-based token loading is skipped entirely.

        Args:
            **overrides: Field values that take priority over the config file
                and environment variables. Pass config_path to read from a
                non-default config file location.

        Returns:
            A fully resolved DepositConfig.

        Raises:
            ConfigError: If the config file contains invalid TOML, an invalid
                boolean environment variable value, or malformed token data
                in an [auths.<fqdn>] section.
        """
        valid_fields = {f.name for f in fields(cls)}
        path_fields = {"local_schema_cache_dir", "schema_cache_dir", "session_dir", "config_path"}
        merged: dict[str, object] = {}

        config_path_override = overrides.pop("config_path", None)
        config_file = (
            Path(config_path_override)  # type: ignore[arg-type]
            if config_path_override is not None
            else DepositConfig().config_path
        )
        merged["config_path"] = config_file

        raw = cls._load_toml_file(config_file)
        section = raw.get("default", {})
        for key, value in section.items():
            if key in valid_fields and key not in ("config_path", "access_token", "refresh_token"):
                if key == "hostname" and value == "":
                    continue
                merged[key] = value

        for env_var, (field_name, coerce) in _ENV_MAP.items():
            raw_val = os.environ.get(env_var)
            if raw_val is not None:
                value = coerce(raw_val)
                if field_name == "hostname" and value == "":
                    continue
                merged[field_name] = value

        for key, value in overrides.items():
            if key in valid_fields:
                merged[key] = value

        # Load tokens from [auths.<fqdn>] unless explicitly overridden via kwargs
        # If either token was explicitly overridden, skip file-based token loading entirely.
        if "access_token" not in merged and "refresh_token" not in merged:
            hostname_val = str(merged.get("hostname", next(f.default for f in fields(cls) if f.name == "hostname")))
            fqdn_key = _hostname_to_fqdn_key(hostname_val)
            if fqdn_key:
                entry = raw.get("auths", {}).get(fqdn_key)
                if isinstance(entry, dict):
                    acc = entry.get("access_token")
                    ref = entry.get("refresh_token")
                    if acc is not None or ref is not None:
                        if acc is not None and not isinstance(acc, str):
                            raise ConfigError(f"Malformed token data in [auths.{fqdn_key}]")
                        if not isinstance(ref, str):
                            raise ConfigError(f"Malformed token data in [auths.{fqdn_key}]")
                        merged["access_token"] = acc
                        merged["refresh_token"] = ref

        for key in path_fields:
            if key in merged and not isinstance(merged[key], Path):
                merged[key] = Path(merged[key])  # type: ignore[arg-type]

        return cls(**merged)  # type: ignore[arg-type]

    def _read_toml(self) -> dict:
        return DepositConfig._load_toml_file(self.config_path)

    def _write_toml(self, data: dict) -> None:
        self.config_path.parent.mkdir(parents=True, exist_ok=True)
        tmp = self.config_path.with_suffix(".toml.tmp")
        try:
            tmp.write_text(tomli_w.dumps(data), encoding="utf-8")
            os.replace(tmp, self.config_path)
        except ConfigError:
            tmp.unlink(missing_ok=True)
            raise
        except Exception as exc:
            tmp.unlink(missing_ok=True)
            raise ConfigError(f"Failed to write {self.config_path}: {exc}") from exc

    def read_auth_entry(self, key: str) -> dict | None:
        data = self._read_toml()
        auths = data.get("auths", {})
        if not isinstance(auths, dict):
            raise ConfigError("Malformed [auths] section in config.toml")
        entry = auths.get(key)
        if entry is None:
            return None
        if not isinstance(entry, dict):
            raise ConfigError(f"Malformed [auths.{key}] entry in config.toml")
        return entry

    def read_auth_entries(self) -> dict[str, dict]:
        data = self._read_toml()
        auths = data.get("auths", {})
        if not isinstance(auths, dict):
            raise ConfigError("Malformed [auths] section in config.toml")
        entries: dict[str, dict] = {}
        for key, entry in auths.items():
            if not isinstance(entry, dict):
                raise ConfigError(f"Malformed [auths.{key}] entry in config.toml")
            entries[key] = dict(entry)
        return entries

    def write_auth_entry(self, key: str, entry: dict) -> None:
        data = self._read_toml()
        auths = data.setdefault("auths", {})
        if not isinstance(auths, dict):
            raise ConfigError("Malformed [auths] section in config.toml")
        auths[key] = entry
        self._write_toml(data)

    def delete_auth_entry(self, key: str) -> None:
        data = self._read_toml()
        auths = data.get("auths")
        if not isinstance(auths, dict) or key not in auths:
            return
        del auths[key]
        self._write_toml(data)