Release (deploy) new versions of android app (apk file) to cafebazaar automatically. Great to be used on CI/CD pipelines

cafebazaar_rollout.py

import hashlib
import json
import os

import requests
import urllib3


class CafeBazaarClient:

    def __init__(self, package_name, username, password):
        self.url = 'https://pishkhan.cafebazaar.ir/api'
        self.package_name = package_name
        self.username = username
        self.password = password
        self._access_token = None
        self._static_parameters = {'lang': 'fa'}

    @property
    def _authentication_headers(self):
        self._ensure_authentication()
        # TODO: Check expiration
        return {'authorization': f'Bearer {self._access_token}'}

    def _ensure_authentication(self):
        if self._access_token is None:
            response = requests.post(
                f'{self.url}/sessions/',
                params=self._static_parameters,
                data={
                    'email': self.username,
                    'password': self.password,
                    'recaptcha_response_field': '',  # TODO:
                }
            )
            result = response.json()
            self._access_token = result.get('access_token')
            if (not 200 <= response.status_code < 300) or (not self._access_token):
                raise Exception(f'Error in login: {result}')

    def get_app_list(self):
        result = requests.get(
            f'{self.url}/apps/',
            params=self._static_parameters,
            headers=self._authentication_headers
        ).json()
        return result.get('apps')

    def post_new_release(self):
        result = requests.post(
            f'{self.url}/apps/{self.package_name}/releases/',
            params=self._static_parameters,
            headers=self._authentication_headers
        ).json()
        release = result.get('release')
        if not release:
            raise Exception(f'Error occurred in release creation: {result}')
        return release

    def delete_release(self, release_id):
        result = requests.delete(
            f'{self.url}/apps/{self.package_name}/releases/{release_id}',
            params=self._static_parameters,
            headers=self._authentication_headers
        ).json()
        return result

    def get_releases(self, committed=1):
        """
        :param committed: 1 means submitted, 0 means draft
        :return: list of all release objects
        """
        result = requests.get(
            f'{self.url}/apps/{self.package_name}/releases/',
            params={'committed': committed, **self._static_parameters},
            headers=self._authentication_headers
        ).json()
        return result.get('releases')

    def upload_apk(self, apk_path):
        with open(apk_path, 'rb') as apk_file:
            file_size = os.path.getsize(apk_file.name)
            file_name = os.path.basename(apk_file.name)
            file_hash = hashlib.md5(apk_file.read()).hexdigest()

        """
            Sample response
            {
                chunk_size: 100000
                file_id: "00000000-0000-0000-0000-00000000"
                file_name: "xxx.apk"
                offset: 99999999
                type: "success
            }
        """

        # We make a blank file id, after that we'll be able to upload the file:
        result = requests.get(
            f'{self.url}/upload/file/',
            params={
                'file_hash': file_hash,
                'size': file_size,
                'file_name': file_name,
                **self._static_parameters
            },
            headers=self._authentication_headers
        ).json()
        file_id = result.get('file_id')
        offset = result.get('offset')
        chunk_size = result.get('chunk_size')

        # Upload the file in multiple parts
        with open(apk_path, 'rb') as apk_file:

            chunk_http = urllib3.PoolManager()
            while True:
                current_chunk_size = chunk_size if offset + chunk_size < file_size else file_size - offset
                chunk_data = apk_file.read(current_chunk_size)
                chunk_hash = hashlib.md5(chunk_data).hexdigest()

                # For unknown reasons, we can not use the `requests` python library, so we use urllib3
                chunk_response = chunk_http.request(
                    'PATCH',
                    f'{self.url}/upload/file/{file_id}?offset={offset}'
                    f'&chunk_size={current_chunk_size}'
                    f'&size={file_size}'
                    f'&chunk_hash={chunk_hash}'
                    f'&lang=fa',
                    headers={'Content-Type': 'application/octet-stream', **self._authentication_headers},
                    body=chunk_data
                )
                chunk_result = json.loads(chunk_response.data.decode())
                if not 200 <= chunk_response.status < 300:
                    raise Exception(f'Error in uploading chunk: {chunk_result}')

                print(f'Chunk uploaded with size {current_chunk_size} from {offset}')
                offset += current_chunk_size
                if offset >= file_size:
                    break

            print(f'File uploaded: {chunk_result}')

        response = requests.post(
            f'{self.url}/apps/{self.package_name}/releases/{release_id}/packages/',
            params=self._static_parameters,
            headers=self._authentication_headers,
            json={'apk_id': chunk_result.get('apk').get('id')}
        )
        result = response.json()
        if not 200 <= response.status_code < 300:
            raise Exception(f'Release packaging error: {result}')
        return result

    def set_auto_publish(self, release_id, auto_publish: bool):
        """
        Default auto publish is False or any release.
        :param release_id: The release number, something like 1234556
        :param auto_publish: True means auto publish
        :return:
        """
        result = requests.patch(
            f'{self.url}/apps/{self.package_name}/releases/{release_id}',
            params=self._static_parameters,
            headers=self._authentication_headers,
            json={'auto_publish': auto_publish}
        ).json()
        return result

    def rollout(self, release_id, changelog_en: str, changelog_fa: str, developer_note: str = '',
                staged_rollout_percentage: int = 100):
        """
        Rollout the release. (The final step for a release_id)
        :param release_id: The release number, something like 1234556
        :param changelog_en: Changelog in english, sample: <p>English changelog</p>
        :param changelog_fa: Like the English changelog
        :param developer_note: A note to the reviewer, describing why permissions are required and etc.
        :param staged_rollout_percentage: 100 means the full rollout.
        :return:
        """
        response = requests.post(
            f'{self.url}/apps/{self.package_name}/releases/{release_id}/v2/commit/',  # Do not remove the trailing slash
            params=self._static_parameters,
            headers=self._authentication_headers,
            data={
                'changelog_en': changelog_en,
                'changelog_fa': changelog_fa,
                'developer_note': developer_note,
                'staged_rollout_percentage': staged_rollout_percentage,
            }
        )
        result = response.json()
        if not 200 <= response.status_code < 300:
            raise Exception(f'Release commit error: {result}')
        return result


if __name__ == '__main__':
    import argparse
    from pathlib import Path

    parser = argparse.ArgumentParser()
    parser.add_argument('package_name', help='The package name, like com.example.myapp', type=str)
    parser.add_argument('-U', '--username', type=str, help='The email of your account, like sth@gmail.com')
    parser.add_argument('-P', '--password', type=str, help='The password of your account')
    parser.add_argument('-a', '--apk-path', type=str, help='The apk file path')
    parser.add_argument('--changelog-en-path', type=str, help='The EN changelog file path')
    parser.add_argument('--changelog-fa-path', type=str, help='The FA changelog file path')
    args = parser.parse_args()

    client = CafeBazaarClient(args.package_name, args.username, args.password)
    print(f'\nDeleting draft releases...')
    deletes = [client.delete_release(release_id=release['id']) for release in client.get_releases(committed=0)]
    print(f'Deleted {len(deletes)} releases successfully.')

    print(f'\nSubmitting a new release...')
    release_id = client.post_new_release().get('id')
    print(f'A new release submitted successfully: {release_id}')

    print(f'\nUploading the apk...')
    client.upload_apk(args.apk_path)
    print(f'Apk uploaded successfully!')

    print(f'\nRolling out the release...')
    client.set_auto_publish(release_id, True)
    rollout_response = client.rollout(
        release_id,
        changelog_en=Path(args.changelog_en_path).read_text(),
        changelog_fa=Path(args.changelog_fa_path).read_text(),
    )
    print(f'Rollout finished: {rollout_response}')