Get All BitBucket Repositories (App Password)

repo-downloader-bitbucket.py

# Get all bitbucket repositories within an organisation and clone them in parallel
#
# Quick run:
#   docker run -it --rm -v "${PWD}:/src" -v "C:\Projects\:/repos" --env-file .env -w /src python:3.11 bash -c "pip install requests && python ./repo-downloader-bitbucket.py"
#
# 1. Set up an app password on your BitBucket account. Only repo read permissions are necessary.
# 2. Setup your .env file or provide the environment variables listed below. The contents of the .env file are listed below:
#   # If you are using the default docker command above, /repos will work
#   #PROJECT_ROOT=/repos
#   PROJECT_ROOT=
#   BITBUCKET_ORGANISATION=
#   # You can get your BITBUCKET_USER from https://bitbucket.org/account/settings/ under Bitbucket profile settings 
#   BITBUCKET_USER=
#   BITBUCKET_APP_PASS=
#   # Set to one of the available clone methods for your repo - conventionally either git or https
#   # If you are using the docker method above, leave as https to git clone using the app pass
#   # Setting this to ssh will authenticate using your configured git credentials. This won't work 
#   # in Docker unless you do some extra work.
#   #CLONE_METHOD=https
#   CLONE_METHOD=

# 3. Run the script.

import requests
import json
import subprocess
import os
from urllib.parse import urlparse
PROJECT_ROOT = os.environ.get("PROJECT_ROOT")
BITBUCKET_ORGANISATION = os.environ.get("BITBUCKET_ORGANISATION")
BITBUCKET_USER = os.environ.get("BITBUCKET_USER")
BITBUCKET_APP_PASS = os.environ.get("BITBUCKET_APP_PASS")
# Typically git or https. https method can use app password
CLONE_METHOD = os.environ.get("CLONE_METHOD") or "https"

print(f'PROJECT_ROOT: {PROJECT_ROOT}')
print(f'BITBUCKET_ORGANISATION: {BITBUCKET_ORGANISATION}')
print(f'BITBUCKET_USER: {BITBUCKET_USER}')
print(f'BITBUCKET_APP_PASS: {BITBUCKET_APP_PASS}')
print(f'CLONE_METHOD: {CLONE_METHOD}')

reposToGet = []
def handleRepositoriesResponsePage(jResponse):
    for repo in jResponse['values']:
        for linkName, linkValue in repo['links'].items():
            if linkName == "clone":
                availableMethods = []
                addressFound = False
                for availableCloneMethod in linkValue:
                    availableMethods.append(availableCloneMethod['name'])
                    if availableCloneMethod['name'] == CLONE_METHOD:
                        if availableCloneMethod['name'] == "https":
                            repoCloneAddress = availableCloneMethod["href"]
                            repoCloneComponents = repoCloneAddress.split("@")
                            repoToClone = f'{repoCloneComponents[0]}:{BITBUCKET_APP_PASS}@{repoCloneComponents[1]}'
                        else:
                            print(availableCloneMethod)
                            repoToClone = availableCloneMethod["href"]
                        reposToGet.append([repo["slug"],repoToClone])
                        addressFound = True
                        
                if not addressFound:
                    raise Exception(f'Selected clone method not available for {repo["name"]}. Supported methods were [{",".join(availableMethods)}]')
    if 'next' in jResponse:
        return jResponse['next']
    else:
        return False

print(f'Requesting repository list from https://api.bitbucket.org/2.0/repositories/{BITBUCKET_ORGANISATION}')
response = requests.get(f'https://api.bitbucket.org/2.0/repositories/{BITBUCKET_ORGANISATION}', auth=(BITBUCKET_USER,BITBUCKET_APP_PASS), headers={"Accept": "application/json"})
if not response.ok :
    raise Exception(f'Request failed: {response.reason}')
else:
    next = handleRepositoriesResponsePage(response.json())
    while next:
        url = urlparse(next)
        response = requests.get(f'https://api.bitbucket.org/2.0/repositories/{BITBUCKET_ORGANISATION}', auth=(BITBUCKET_USER,BITBUCKET_APP_PASS), params={"page":url.query.split('=')[1]})
        next = handleRepositoriesResponsePage(response.json())
    processes = []
    for slug,repo in reposToGet:
        processes.append([slug,repo,subprocess.Popen(["git","clone","-q",repo,os.path.join(PROJECT_ROOT,slug)])])
    exit_codes = [[slug,repo,process.wait()] for slug,repo,process in processes]
    for slug, repo, exit_code in exit_codes:
        print(f'{slug} ({repo}): Exit {exit_code}')