Script to download Workflow files from Bitrise.

bitrise_workflow_downloader.py

import os
import json
from optparse import OptionParser
import http.client

BITRISE_URL = 'api.bitrise.io'
BASE_PATH = '/v0.1'

class BitriseApp:
    def __init__(self, slug, repo_slug):
        self.slug = slug
        self.repo_slug = repo_slug
        self.workflow = None

    def upload(self, client):
        if not self.workflow or not self.workflow.strip():
            raise ValueError('Workflow is empty')
        
        response = client.post_bitrise_yml(self.slug, self.workflow)
        if response.status_code != 200:
            print(f"POST errored with: \"{response.reason}\"")
    
    def read_from_server(self, client):
        self.workflow = client.bitrise_yml(self.slug)

    def save_to_disk(self):
        if not self.workflow or not self.workflow.strip():
            raise ValueError('Workflow is empty')
        
        file_path = os.path.join('workflows', f"{self.repo_slug}_{self.slug}.yml")
        with open(file_path, 'w') as f:
            f.write(self.workflow)
        print(f"Saved to {file_path}")

    def read_from_disk(self):
        file_path = os.path.join('workflows', f"{self.repo_slug}_{self.slug}.yml")
        self.workflow = open(file_path).read()

class BitriseClient:
    def __init__(self, access_token=None):
        self.access_token = access_token or os.getenv('BITRISE_ACCESS_TOKEN')

    def apps(self):
        results = []
        conn = http.client.HTTPSConnection(BITRISE_URL)
        headers = {'Authorization': f'Bearer {self.access_token}'}
        
        while True:
            conn.request("GET", f"{BASE_PATH}/apps", headers=headers)
            response = conn.getresponse()
            content = json.loads(response.read().decode())
            
            for app in content['data']:
                results.append(BitriseApp(app['slug'], app['repo_slug']))
            
            if not content['paging']['next']:
                break
        
        return results

    def bitrise_yml(self, slug):
        conn = http.client.HTTPSConnection(BITRISE_URL)
        headers = {'Authorization': f'Bearer {self.access_token}'}
        
        conn.request("GET", f"{BASE_PATH}/apps/{slug}/bitrise.yml", headers=headers)
        response = conn.getresponse()
        
        if response.status == 200:
            return response.read().decode()
        else:
            return None

    def post_bitrise_yml(self, slug, workflow):
        conn = http.client.HTTPSConnection(BITRISE_URL)
        headers = {'Authorization': f'Bearer {self.access_token}', 'Content-Type': 'application/json'}
        data = json.dumps({'app_config_datastore_yaml': workflow})
        
        conn.request("POST", f"{BASE_PATH}/apps/{slug}/bitrise.yml", body=data, headers=headers)
        response = conn.getresponse()
        return response

# Parse command line options
options = {
    'token': os.getenv('BITRISE_ACCESS_TOKEN'),
    'output_dir': 'workflows',
}

parser = OptionParser(usage="Usage: %prog [options] action [app_name]")
parser.add_option('-o', '--output-dir', help='Output directory. Defaults to "workflows/"')
parser.add_option('-h', '--help', action='help', help='Display this help.')

(options, args) = parser.parse_args()

if not options.output_dir:
    options.output_dir = 'workflows'

action = args[0] if len(args) > 0 else None
app_name = args[1] if len(args) > 1 else None

client = BitriseClient(access_token=options.token)

apps = client.apps()

if action == 'downloadall':
    if app_name and app_name.strip():
        print('Do not specify the app name if for downloadall.')
        exit(-1)
    print('Downloading all Workflows.')
    for app in apps:
        try:
            app.read_from_server(client)
            app.save_to_disk()
        except Exception as e:
            print(f"Failed with {e}")
elif action == 'download':
    app = next((app for app in apps if app.repo_slug == app_name), None)
    if not app:
        print(f"No app found with repo_slug: {app_name}")
        exit(-1)
    print(f"Downloading {app.repo_slug}")
    app.read_from_server(client)
    app.save_to_disk()
elif action == 'upload':
    app = next((app for app in apps if app.repo_slug == app_name), None)
    if not app:
        print(f"No app found with repo_slug: {app_name}")
        exit(-1)
    print(f"Uploading {app.repo_slug}")
    app.read_from_disk()
    app.upload(client)