Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Release (deploy) new versions of android app (apk file) to cafebazaar automatically. Great to be used on CI/CD pipelines
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}')
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment