Nexus 制品库 API

Sonatype Nexus 托管 Maven、npm、Docker、Raw 等制品。运维场景：查询组件版本、清理 SNAPSHOT、检查仓库配额、同步发布记录到 CMDB。Nexus 3 REST API 基于 /service/rest/v1。

---

环境准备

uv add httpx

export NEXUS_URL="https://nexus.example.com"
export NEXUS_USER="deploy-bot"
export NEXUS_PASS="nexus-password-placeholder"

建议为自动化账号分配 nx-repository-view-*-browse 与必要的 delete 权限，分环境分账号。

---

搜索组件

#!/usr/bin/env python3
"""在 Nexus 中搜索 Maven 组件，列出 group:artifact 的最新版本。"""

import os
from typing import Any

import httpx


class NexusClient:
    def __init__(self, base_url: str, username: str, password: str) -> None:
        self.client = httpx.Client(
            base_url=f"{base_url.rstrip('/')}/service/rest/v1",
            auth=(username, password),
            timeout=30.0,
        )

    def search_components(
        self,
        repository: str,
        group: str | None = None,
        name: str | None = None,
        version: str | None = None,
    ) -> list[dict[str, Any]]:
        params: dict[str, str] = {"repository": repository}
        if group:
            params["group"] = group
        if name:
            params["name"] = name
        if version:
            params["version"] = version

        items: list[dict[str, Any]] = []
        continuation: str | None = None
        while True:
            if continuation:
                params["continuationToken"] = continuation
            resp = self.client.get("/search", params=params)
            resp.raise_for_status()
            data = resp.json()
            items.extend(data.get("items", []))
            continuation = data.get("continuationToken")
            if not continuation:
                break
        return items


def main() -> None:
    nexus = NexusClient(
        os.environ["NEXUS_URL"],
        os.environ["NEXUS_USER"],
        os.environ["NEXUS_PASS"],
    )
    # 示例：搜索 maven-releases 中 com.example 下的组件
    results = nexus.search_components(
        repository="maven-releases",
        group="com.example",
        name="user-service",
    )
    for item in results[:10]:
        print(f"{item['group']}:{item['name']}:{item['version']}  format={item['format']}")


if __name__ == "__main__":
    main()

---

列出仓库与存储信息

def list_repositories(nexus: NexusClient) -> None:
    resp = nexus.client.get("/repositories")
    resp.raise_for_status()
    for repo in resp.json():
        # 类型如 hosted / proxy / group
        print(f"{repo['name']:25s}  format={repo['format']:8s}  type={repo['type']}")


def get_component_assets(nexus: NexusClient, component_id: str) -> list[dict]:
    resp = nexus.client.get(f"/components/{component_id}")
    resp.raise_for_status()
    return resp.json().get("assets", [])

---

删除指定组件（谨慎）

def delete_component(nexus: NexusClient, component_id: str, dry_run: bool = True) -> None:
    """删除组件及其所有 asset；生产务必 dry_run 先行。"""
    msg = f"删除 component_id={component_id}"
    if dry_run:
        print(f"[DRY-RUN] {msg}")
        return
    resp = nexus.client.delete(f"/components/{component_id}")
    resp.raise_for_status()
    print(f"已{msg}")

清理过期 SNAPSHOT：搜索 → 审批 → 批量 delete_component（默认 dry_run=True）。Raw 仓库上传用 POST /components?repository=...，适合运维脚本包分发。

注意

• Nexus 删除不可恢复，脚本默认 dry_run=True
• 大仓库搜索注意分页（continuationToken）
• npm/Maven 代理仓库的组件可能无法直接删除，需清理缓存任务

---

小结

场景	API
搜索制品	GET /search?repository=...
列仓库	GET /repositories
删组件	DELETE /components/{id}

下一章用 Docker SDK 管理本地或远程 Docker 引擎。