"""Project-wide configuration and platform credentials helpers.
This module is the canonical source of truth for ``CONFIG_FILE`` and all
configuration management functions (``load_config``, ``save_config``,
``get_active_profile``, …). ``uniqc.backend_adapter.config`` re-exports from here
so that patching ``uniqc.config.CONFIG_FILE`` propagates to every import path.
All credentials and cache settings are persisted in ``~/.uniqc/config.yaml``.
Example configuration structure::
always_ai_hints: false
active_profile: default
default:
originq:
token: xxx
quark:
QUARK_API_KEY: xxx
ibm:
token: xxx
proxy:
http: http://proxy:8080
https: https://proxy:8080
"""
from __future__ import annotations
import os
from pathlib import Path
from typing import Any
import yaml
from uniqc._error_hints import format_enriched_message
# ---------------------------------------------------------------------------
# Top-level symbols (defined here so that uniqc.backend_adapter.config can
# import them and both modules reference the same objects after patching)
# ---------------------------------------------------------------------------
CONFIG_DIR = Path.home() / ".uniqc"
CONFIG_FILE = CONFIG_DIR / "config.yaml"
# Schema version of ``config.yaml``. ``load_config`` automatically migrates
# older files (in place, best-effort persisted) to ``CURRENT_CONFIG_VERSION``;
# files written by a *newer* uniqc are rejected with a clear error.
#
# Migration history:
# 0 -> 1: stamp the ``config_version`` key; the layout itself is unchanged
# (v0 = any config.yaml written before schema versioning existed).
CONFIG_VERSION_KEY = "config_version"
CURRENT_CONFIG_VERSION = 1
DEFAULT_CONFIG: dict[str, Any] = {
CONFIG_VERSION_KEY: CURRENT_CONFIG_VERSION,
"always_ai_hints": False,
"default": {
"originq": {
"token": "",
"available_qubits": [],
"available_topology": [],
"task_group_size": 200,
},
"quark": {
"QUARK_API_KEY": "",
},
"ibm": {
"token": "",
"proxy": {
"http": "",
"https": "",
},
},
"tianyan": {
"login_key": "",
},
"logicalqubit": {
"api_key": "",
"url": "https://cloud.logicalqubit.com",
},
},
}
SUPPORTED_PLATFORMS = ["originq", "quark", "ibm", "tianyan", "logicalqubit"]
META_KEYS = frozenset({"active_profile", "always_ai_hints", "sync", CONFIG_VERSION_KEY})
PLATFORM_REQUIRED_FIELDS = {
"originq": ["token"],
"quark": ["QUARK_API_KEY"],
"ibm": ["token"],
"tianyan": ["login_key"],
"logicalqubit": ["api_key"],
}
PLATFORM_KNOWN_FIELDS = {
"originq": {"token", "task_group_size", "available_qubits"},
"quark": {"QUARK_API_KEY", "token"},
"ibm": {"token", "proxy", "chip_id", "auto_mapping", "circuit_optimize", "task_name", "shots"},
"tianyan": {"login_key", "chip_id", "auto_mapping", "task_name", "shots"},
"logicalqubit": {"api_key", "url", "chip_id", "auto_mapping", "task_name", "shots"},
}
# ---------------------------------------------------------------------------
# Exceptions (re-exported from the central module)
# ---------------------------------------------------------------------------
import contextlib
from uniqc.exceptions import ( # noqa: F401 — re-export for backward compat
ConfigError,
ConfigValidationError,
PlatformNotFoundError,
ProfileNotFoundError,
)
# ---------------------------------------------------------------------------
# Core functions (also used by uniqc.backend_adapter.config via re-export)
# ---------------------------------------------------------------------------
def _ensure_config_dir() -> None:
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
[docs]
def load_config(config_path: str | Path | None = None) -> dict[str, Any]:
path = Path(config_path) if config_path else CONFIG_FILE
if not path.exists():
return DEFAULT_CONFIG.copy()
try:
with open(path, encoding="utf-8") as f:
config = yaml.safe_load(f)
if config is None:
return DEFAULT_CONFIG.copy()
except yaml.YAMLError as e:
raise ConfigError(format_enriched_message(f"Failed to parse YAML configuration: {e}", "config")) from e
except OSError as e:
raise ConfigError(format_enriched_message(f"Failed to read configuration file: {e}", "config")) from e
if isinstance(config, dict):
original_version = get_config_version(config)
config = migrate_config(config)
if get_config_version(config) != original_version:
# Best-effort: persist the migrated file, but never break reads.
with contextlib.suppress(ConfigError):
save_config(config, path)
return config
[docs]
def get_config_version(config: dict[str, Any]) -> int:
"""Return the schema version of a parsed config dict (0 = unversioned legacy)."""
version = config.get(CONFIG_VERSION_KEY, 0)
if isinstance(version, bool) or not isinstance(version, int) or version < 0:
raise ConfigError(
format_enriched_message(
f"Invalid '{CONFIG_VERSION_KEY}' in configuration: {version!r} (expected a non-negative integer)",
"config",
)
)
return version
def _migrate_0_to_1(config: dict[str, Any]) -> dict[str, Any]:
"""v0 (unversioned) → v1: layout unchanged, only the version key is added."""
return config
# Ordered per-step migrations: _MIGRATIONS[v] upgrades a v-config to v+1.
_MIGRATIONS = {0: _migrate_0_to_1}
[docs]
def migrate_config(config: dict[str, Any]) -> dict[str, Any]:
"""Migrate a parsed config dict to ``CURRENT_CONFIG_VERSION``.
Raises ``ConfigError`` if the file was written by a newer uniqc.
"""
version = get_config_version(config)
if version > CURRENT_CONFIG_VERSION:
raise ConfigError(
format_enriched_message(
f"Configuration schema version {version} is newer than this uniqc supports "
f"({CURRENT_CONFIG_VERSION}). Please upgrade uniqc.",
"config",
)
)
while version < CURRENT_CONFIG_VERSION:
config = _MIGRATIONS[version](config)
version += 1
config[CONFIG_VERSION_KEY] = version
return config
[docs]
def save_config(config: dict[str, Any], config_path: str | Path | None = None) -> None:
path = Path(config_path) if config_path else CONFIG_FILE
parent = path.parent
# Ensure the parent directory exists. If we are the ones creating it,
# make it 0o700 so it cannot be entered by other users on a shared host.
# If the user already created it with a looser mode, leave that alone so
# we don't surprise them by locking other tooling out.
newly_created_parent = not parent.exists()
parent.mkdir(parents=True, exist_ok=True)
if newly_created_parent:
try:
os.chmod(parent, 0o700)
except OSError:
# Best-effort: e.g. Windows ignores mode bits >0o100; don't fail.
pass
tmp_path = parent / f".{path.name}.tmp"
try:
fd = os.open(
str(tmp_path),
os.O_CREAT | os.O_WRONLY | os.O_TRUNC,
0o600,
)
try:
with os.fdopen(fd, "w", encoding="utf-8") as f:
yaml.dump(
config,
f,
default_flow_style=False,
allow_unicode=True,
sort_keys=False,
)
except Exception:
# Ensure no half-written temp file is left behind.
with contextlib.suppress(OSError):
os.unlink(tmp_path)
raise
os.replace(tmp_path, path)
# Re-tighten in case ``path`` pre-existed with looser perms (replace
# preserves the *new* file's mode, but be defensive against umask
# quirks or FS-specific behaviours).
try:
os.chmod(path, 0o600)
except OSError:
# Best-effort: Windows / exotic FS may ignore mode bits.
pass
except OSError as e:
raise ConfigError(format_enriched_message(f"Failed to write configuration file: {e}", "config")) from e
[docs]
def validate_config(
config: dict[str, Any] | None = None,
config_path: str | Path | None = None,
) -> list[str]:
errors: list[str] = []
try:
cfg = config if config is not None else load_config(config_path)
except ConfigError as e:
return [str(e)]
if not isinstance(cfg, dict):
return ["Configuration must be a dictionary"]
if not cfg:
return ["Configuration is empty"]
for profile_name, profile_config in cfg.items():
if profile_name in META_KEYS:
continue
if not isinstance(profile_config, dict):
errors.append(f"Profile '{profile_name}' must be a dictionary")
continue
for platform_name in SUPPORTED_PLATFORMS:
if platform_name not in profile_config:
continue
platform_config = profile_config[platform_name]
if not isinstance(platform_config, dict):
errors.append(f"Platform '{platform_name}' in profile '{profile_name}' must be a dictionary")
continue
if platform_name == "quark":
if "QUARK_API_KEY" not in platform_config and "token" not in platform_config:
errors.append(
"Missing required field 'QUARK_API_KEY' for platform "
f"'{platform_name}' in profile '{profile_name}'"
)
else:
required_fields = PLATFORM_REQUIRED_FIELDS.get(platform_name, [])
for field in required_fields:
if field not in platform_config:
errors.append(
f"Missing required field '{field}' for platform "
f"'{platform_name}' in profile '{profile_name}'"
)
# Warn about unknown keys (likely typos)
known = PLATFORM_KNOWN_FIELDS.get(platform_name, set())
unknown = set(platform_config.keys()) - known
for key in sorted(unknown):
errors.append(
f"Warning: unknown field '{key}' for platform '{platform_name}' in profile '{profile_name}'"
)
if platform_name == "ibm" and "proxy" in platform_config:
proxy = platform_config["proxy"]
if not isinstance(proxy, dict):
errors.append(f"Proxy configuration for IBM in profile '{profile_name}' must be a dictionary")
else:
for proxy_type in ["http", "https"]:
if proxy_type in proxy and not isinstance(proxy[proxy_type], str):
errors.append(f"Proxy '{proxy_type}' for IBM in profile '{profile_name}' must be a string")
return errors
[docs]
def create_default_config(config_path: str | Path | None = None) -> None:
path = Path(config_path) if config_path else CONFIG_FILE
if path.exists():
return
_ensure_config_dir()
save_config(DEFAULT_CONFIG, path)
[docs]
def get_active_profile(config_path: str | Path | None = None) -> str:
env_profile = os.environ.get("UNIQC_PROFILE")
if env_profile:
return env_profile
config = load_config(config_path)
if "active_profile" in config:
return config["active_profile"]
return "default"
[docs]
def set_active_profile(
profile: str,
config_path: str | Path | None = None,
) -> None:
config = load_config(config_path)
if profile not in config:
raise ProfileNotFoundError(
format_enriched_message(
f"Profile '{profile}' not found in configuration. "
f"Available profiles: {', '.join(k for k in config if k not in META_KEYS)}",
"config",
)
)
config["active_profile"] = profile
save_config(config, config_path)
[docs]
def get_always_ai_hints(config_path: str | Path | None = None) -> bool:
config = load_config(config_path)
return bool(config.get("always_ai_hints", False))
[docs]
def set_always_ai_hints(
enabled: bool,
config_path: str | Path | None = None,
) -> None:
config = load_config(config_path)
config["always_ai_hints"] = bool(enabled)
save_config(config, config_path)
[docs]
def get_originq_config(profile: str | None = None) -> dict[str, Any]:
if profile is None:
profile = get_active_profile()
return get_platform_config("originq", profile)
[docs]
def get_quark_config(profile: str | None = None) -> dict[str, Any]:
if profile is None:
profile = get_active_profile()
return get_platform_config("quark", profile)
[docs]
def get_ibm_config(profile: str | None = None) -> dict[str, Any]:
if profile is None:
profile = get_active_profile()
return get_platform_config("ibm", profile)
# ---------------------------------------------------------------------------
# Platform-specific credential loaders
# ---------------------------------------------------------------------------
def _load_platform_config(platform: str) -> dict[str, Any]:
profile = get_active_profile()
return get_platform_config(platform, profile)
[docs]
def load_originq_config() -> dict[str, Any]:
config = _load_platform_config("originq")
api_key = config.get("token", "") or None
if api_key:
return {
"api_key": api_key,
"task_group_size": int(config.get("task_group_size", 200) or 200),
"available_qubits": config.get("available_qubits", []),
}
raise ImportError(
format_enriched_message(
"OriginQ Cloud config not found. "
"Run `uniqc config set originq.token <TOKEN>` or edit ~/.uniqc/config.yaml.",
"config",
)
)
[docs]
def load_quark_config() -> dict[str, Any]:
config = _load_platform_config("quark")
api_token = config.get("QUARK_API_KEY", "") or config.get("token", "") or None
if api_token:
return {"api_token": api_token}
raise ImportError(
format_enriched_message(
"QuarkStudio config not found. "
"Run `uniqc config set quark.QUARK_API_KEY <TOKEN>` or edit ~/.uniqc/config.yaml.",
"config",
)
)
[docs]
def load_ibm_config() -> dict[str, Any]:
config = _load_platform_config("ibm")
api_token = config.get("token", "") or None
if api_token:
return {"api_token": api_token}
raise ImportError(
format_enriched_message(
"IBM Quantum config not found. Run `uniqc config set ibm.token <TOKEN>` or edit ~/.uniqc/config.yaml.",
"config",
)
)
[docs]
def load_tianyan_config() -> dict[str, Any]:
config = _load_platform_config("tianyan")
login_key = config.get("login_key", "") or None
if login_key:
return {"login_key": login_key}
raise ImportError(
format_enriched_message(
"Tianyan config not found. "
"Run `uniqc config set tianyan.login_key <KEY>` or edit ~/.uniqc/config.yaml.",
"config",
)
)
[docs]
def load_logicalqubit_config() -> dict[str, Any]:
config = _load_platform_config("logicalqubit")
api_key = config.get("api_key", "") or None
if api_key:
return {
"api_key": api_key,
"url": config.get("url", "") or "https://cloud.logicalqubit.com",
}
raise ImportError(
format_enriched_message(
"LogicalQubit config not found. "
"Run `uniqc config set logicalqubit.api_key <KEY>` or edit ~/.uniqc/config.yaml.",
"config",
)
)
[docs]
def load_dummy_config() -> dict[str, Any]:
try:
config = _load_platform_config("originq")
except Exception:
config = {}
return {
"available_qubits": config.get("available_qubits", []),
"available_topology": config.get("available_topology", []),
"task_group_size": int(config.get("task_group_size", 200) or 200),
}