Created
January 29, 2026 09:56
-
-
Save masecla22/9e4753cdd877cc90b264e8b980460a12 to your computer and use it in GitHub Desktop.
Small Python script to move Artifacts from one Nexus Sonatype repo to another for migration
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import requests | |
| import shutil | |
| import os | |
| import sys | |
| import json | |
| from urllib.parse import urlparse, urlunparse | |
| # This script migrates artifacts from one Nexus repository to another. | |
| # It supports Maven, Raw, Yum, and PyPI repositories by utilizing Nexus's REST API. | |
| # I dont know if it supports other repository types. | |
| # (might need ) | |
| # =================CONFIGURATION================= | |
| # Source Nexus Details | |
| SOURCE_URL = "https://example.nexus.com" | |
| SOURCE_REPO = "some_repo" | |
| SOURCE_AUTH = ("admin", "") | |
| # Target Nexus Details | |
| TARGET_URL = "https://target.nexus.com" | |
| TARGET_REPO = "some_other_repo" # Same name as source | |
| TARGET_AUTH = ("admin", "") | |
| # Temporary directory for downloads | |
| TEMP_DIR = "./nexus_migration_temp" | |
| # =============================================== | |
| def construct_api_url(base, endpoint): | |
| """Safely joins a base URL and an API endpoint.""" | |
| parsed = urlparse(base) | |
| # Ensure path ends with slash for join, or just append explicitly | |
| path = parsed.path.rstrip('/') + endpoint | |
| return urlunparse(parsed._replace(path=path)) | |
| def construct_content_url(base, repo, asset_path): | |
| """Constructs the repository content URL for PUT requests.""" | |
| # Pattern: {Base}/repository/{RepoName}/{AssetPath} | |
| parsed = urlparse(base) | |
| path = f"{parsed.path.rstrip('/')}/repository/{repo}/{asset_path.lstrip('/')}" | |
| return urlunparse(parsed._replace(path=path)) | |
| def get_session(auth): | |
| s = requests.Session() | |
| s.auth = auth | |
| s.verify = True # Set False to ignore SSL errors | |
| return s | |
| def fetch_assets_generator(session, base_url, repo_name): | |
| """Generates assets from the source, handling pagination.""" | |
| api_url = construct_api_url(base_url, "/service/rest/v1/assets") | |
| params = {"repository": repo_name} | |
| print(f"[*] Querying API: {api_url}") | |
| while True: | |
| try: | |
| resp = session.get(api_url, params=params) | |
| # DEBUG: Handle non-JSON responses | |
| if resp.status_code != 200: | |
| print(f"[!] HTTP Error {resp.status_code}") | |
| print(f"[!] Response Preview: {resp.text[:200]}") | |
| sys.exit(1) | |
| try: | |
| data = resp.json() | |
| except json.decoder.JSONDecodeError: | |
| print(f"[!] FATAL: Response was not JSON. Is the URL correct?") | |
| print(f"[!] Raw Response: {resp.text[:500]}...") | |
| sys.exit(1) | |
| items = data.get('items', []) | |
| if not items: | |
| print("[*] No items found on this page.") | |
| for item in items: | |
| yield item | |
| token = data.get('continuationToken') | |
| if not token: | |
| break | |
| params['continuationToken'] = token | |
| except requests.exceptions.RequestException as e: | |
| print(f"[!] Network Error: {e}") | |
| sys.exit(1) | |
| def migrate(): | |
| if not os.path.exists(TEMP_DIR): | |
| os.makedirs(TEMP_DIR) | |
| src_session = get_session(SOURCE_AUTH) | |
| tgt_session = get_session(TARGET_AUTH) | |
| print(f"=== Starting Migration: {SOURCE_REPO} -> {TARGET_REPO} ===") | |
| for asset in fetch_assets_generator(src_session, SOURCE_URL, SOURCE_REPO): | |
| path = asset['path'] | |
| download_url = asset['downloadUrl'] | |
| # Checksum verification (optional, can use sha1 or md5 from asset metadata) | |
| checksum = asset.get('checksum', {}).get('sha1') | |
| print(f"Processing: {path}") | |
| local_path = os.path.join(TEMP_DIR, os.path.basename(path)) | |
| # 1. Download | |
| try: | |
| with src_session.get(download_url, stream=True) as r: | |
| r.raise_for_status() | |
| with open(local_path, 'wb') as f: | |
| shutil.copyfileobj(r.raw, f) | |
| except Exception as e: | |
| print(f" [!] Download failed: {e}") | |
| continue | |
| # 2. Upload (Using PUT to /repository/{repo}/{path}) | |
| # This is generic and works for Maven, Raw, Yum, PyPI | |
| upload_url = construct_content_url(TARGET_URL, TARGET_REPO, path) | |
| try: | |
| with open(local_path, 'rb') as f: | |
| # Basic PUT. Nexus usually auto-detects MIME, but you can add headers if needed. | |
| up_resp = tgt_session.put(upload_url, data=f) | |
| if up_resp.status_code in [200, 201]: | |
| print(f" [+] Upload Success") | |
| else: | |
| print(f" [-] Upload Failed: {up_resp.status_code} - {up_resp.text}") | |
| except Exception as e: | |
| print(f" [!] Upload Exception: {e}") | |
| finally: | |
| if os.path.exists(local_path): | |
| os.remove(local_path) | |
| try: | |
| os.rmdir(TEMP_DIR) | |
| except: | |
| pass | |
| print("=== Migration Complete ===") | |
| if __name__ == "__main__": | |
| migrate() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment