bitbucket API: clone/update all repositories

bitbucket-clone-update-all-repos.py

import json
from pathlib import Path

import requests
import subprocess
import os
import base64
from urllib.parse import urlparse

# Configuration
# Create Account Token
# https://id.atlassian.com/manage-profile/security/api-tokens
# Create Workspace Token
# https://bitbucket.org/<bitbucket-workspace-slug>/workspace/settings/access-tokens
BITBUCKET_API_KEY = "⚙️->Atlassian-Account-Settings->Security>API-Tokens->Create and Manage API Tokens-> Create API Token"  # Replace with your actual API key
BITBUCKET_USERNAME = "<bitbucket-workspace-slug>"  # Replace with your Bitbucket username
BASE_URL = "https://api.bitbucket.org/2.0/repositories"
WORKSPACE = "<bitbucket-workspace-slug>"  # Replace with your Bitbucket workspace slug


def get_all_repositories():
    """Fetch all repositories from Bitbucket using pagination"""
    repositories = []
    url = f"{BASE_URL}/{WORKSPACE}"

    # Prepare authentication
    auth_string = f"{BITBUCKET_USERNAME}:{BITBUCKET_API_KEY}"
    auth_header = {
        "Authorization": f"Basic {base64.b64encode(auth_string.encode()).decode()}"
    }

    while url:
        try:
            response = requests.get(url, headers=auth_header)
            response.raise_for_status()
            data = response.json()

            # Add repositories from current page
            for repo in data.get('values', []):
                repositories.append({
                    'name': repo['name'],
                    'clone_url': next(link['href'] for link in repo['links']['clone'] if link['name'] == 'ssh')
                })

            # Check for next page
            url = data.get('next')
            print(f"Fetched {len(repositories)} repositories so far...")

        except requests.exceptions.RequestException as e:
            print(f"Error fetching repositories: {e}")
            break

    return repositories


def clone_repository(repo_url, repo_name):
    """Clone a single repository to the current directory"""
    try:
        # if repo already exists then do a pull
        if os.path.exists(repo_name):
            print(f"!!! {repo_name} - directory already exists -- doing a git pull instead...")
            subprocess.run(['git', '-C', repo_name, 'pull'], check=True)
            return 'pull'

        # Clone the repository
        print(f"Cloning {repo_name}...")
        subprocess.run(['git', 'clone', repo_url], check=True)
        return 'clone'

    except subprocess.CalledProcessError as e:
        print(f"Error cloning {repo_name}: {e}")
        return str(e)


def main():
    # Get all repositories
    print("Fetching repository list from Bitbucket...")
    repositories = get_all_repositories()

    json.dump(repositories, Path('repos.json').open(encoding='utf-8', mode='w'), indent=1, ensure_ascii=False)

    if not repositories:
        print("No repositories found or error occurred")
        return

    print(f"Found {len(repositories)} repositories")

    # Clone each repository
    stats = []
    for repo in repositories:
        # Inject credentials into clone URL
        # auth_url = f"https://{BITBUCKET_USERNAME}:{BITBUCKET_API_KEY}@{parsed_url.netloc}{parsed_url.path}"

        clone_url = repo['clone_url']
        repo_name = repo['name']
        stats.append( clone_repository(clone_url, repo_name))

    print(f"\n{len(stats)} repositories processed")


if __name__ == "__main__":
    # Create authentication info file (optional, for git credentials)
    # with open('.git-credentials', 'w') as f:
    #     f.write(f"https://{BITBUCKET_USERNAME}:{BITBUCKET_API_KEY}@bitbucket.org")

    # Configure git to use the credentials file
    # subprocess.run(['git', 'config', '--global', 'credential.helper', 'store'])

    main()