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

myket_rollout.py

import hashlib
import uuid

import requests


class MyketClient:
    def __init__(self, package_name, username, password):
        self.url = 'https://developer.myket.ir/api'
        self.resource_url = 'https://resource.myket.ir'
        self.package_name = package_name
        self.username = username
        self.password = password
        self._session_payload = dict()
        self._static_parameters = {'lang': 'fa'}

    @property
    def _token(self):
        return self._session_payload.get('token')

    @property
    def _account_id(self):
        return self._session_payload.get('accountId')

    @property
    def _authentication_headers(self):
        self._ensure_authentication()
        # TODO: Check expiration
        return {'authorization': self._session_payload.get('token')}

    @property
    def _authentication_cookies(self):
        self._ensure_authentication()
        return {
            'myketAccessToken': self._session_payload.get('token'),
            'accountId': self._session_payload.get('accountId'),
            'secureId': self._session_payload.get('secureId'),
        }

    def _ensure_authentication(self):
        """
        {
            'token': 'xxx',
            'accountId': 'xxx',
            'accountKey': 'xxx',
            'role': 'Developer',
            'is2Step': False,
            'result': 'Successful',
            'secureId': 'xxx'
         }
        :return:
        """
        if self._session_payload.get('token') is None:
            response = requests.post(
                f'{self.url}/dev-auth/signin/',
                params=self._static_parameters,
                data={
                    'identifier': self.username,
                    'retry': False,  # TODO:
                    'secret': hashlib.sha1(self.password.encode()).hexdigest(),
                    'verificationCode': '',  # TODO:
                }
            )
            result = response.json()
            if not 200 <= response.status_code < 300:
                raise Exception(f'Login error: {result}')
            self._session_payload = result
            return result

    def get_new_version_constraints(self):
        """
        {
            "allowedAddVersion":true,
            "allowedAddStagedRollout":false
        }
        """
        self._ensure_authentication()
        response = requests.get(
            f'{self.url}/developers/{self._account_id}/applications/{self.package_name}/new-version-constraints',
            params=self._static_parameters,
            headers=self._authentication_headers,
            cookies=self._authentication_cookies,
        )
        result = response.json()
        if not 200 <= response.status_code < 300:
            raise Exception(f'Get new version constraint error: {result}')
        return result

    def _upload_apk(self, apk_path):
        """
        :param apk_path: Apk file path
        :return: Link of the apk
        """
        self._ensure_authentication()
        with open(apk_path, 'rb') as apk_file:
            result = requests.post(
                f'{self.resource_url}/Backload/Filehandler',
                params={'objectContext': uuid.uuid4().__str__()},
                files={f'{apk_file.name}': apk_file}
            ).json()
        return result.get('files')[0].get('url')

    def rollout(self, apk_path, changelog_en: str, changelog_fa: str, staged_rollout_percentage: int = 100):
        self._ensure_authentication()
        apk_link = self._upload_apk(apk_path)
        response = requests.post(
            f'{self.url}/developers/{self._account_id}/applications/{self.package_name}/versions',
            params=self._static_parameters,
            headers=self._authentication_headers,
            cookies=self._authentication_cookies,
            json={
                'apkLink': apk_link,
                'stagedRolloutPercent': staged_rollout_percentage,
                'translationInfos': [
                    {'description': changelog_fa, 'language': 'Fa'},
                    {'description': changelog_en, 'language': 'En'},
                ],
            }
        )
        result = response.json()
        if not 200 <= response.status_code < 300:
            raise Exception(f'Release commit error: {result}')
        return result

    def publish(self):
        self._ensure_authentication()
        response = requests.put(
            f'{self.url}/developers/{self._account_id}/applications/{self.package_name}/confirm',
            params=self._static_parameters,
            headers=self._authentication_headers,
            cookies=self._authentication_cookies,
            json={
                'publishDate': None,
            }
        )
        result = response.json()
        if not 200 <= response.status_code < 300:
            raise Exception(f'Release publish 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 [email protected]')
    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 = MyketClient(args.package_name, args.username, args.password)

    print(f'\nGetting new version constraints...')
    constraints = client.get_new_version_constraints()
    print(f'Done getting new version constraints: {constraints}')

    print(f'\nUploading a new release...')
    rollout_response = client.rollout(
        apk_path=args.apk_path,
        changelog_en=Path(args.changelog_en_path).read_text(),
        changelog_fa=Path(args.changelog_fa_path).read_text(),
        staged_rollout_percentage=100 if constraints.get('allowedAddStagedRollout') is True else 0
    )
    print(f'A new release submitted successfully: {rollout_response}')

    print(f'\nPublishing the submitted release...')
    publish_response = client.publish()
    print(f'The new release published successfully: {publish_response}')

    print(f'Rollout finished!')