Currently seems to work only on Python 2.7 due to swagger-codegen using the reserved keyword async.
It also needs a patch on six.py (info here).
One must move the second parameter "unicode_escape" from the unicoide() call in six.py:L653 (see on github).
| import time | |
| from os import environ as env | |
| from sys import stdout | |
| from wodby import Configuration as WodbyConfiguration, InstanceApi, ApplicationApi, TaskApi, Instance, Task, \ | |
| InstanceType, ResponseTask | |
| from wodby import ApiClient as WodbyApiClient | |
| from wodby.rest import ApiException | |
| def wodby_client(): | |
| if 'WODBY_API_KEY' not in env: | |
| raise RuntimeError("Missing WODBY_API_KEY in environment.") | |
| conf = WodbyConfiguration() | |
| conf.api_key['X-API-KEY'] = env['WODBY_API_KEY'] | |
| return WodbyApiClient(configuration=conf) | |
| def main(): | |
| client = wodby_client() | |
| app_api = ApplicationApi(client) | |
| instance_api = InstanceApi(client) | |
| task_api = TaskApi(client) | |
| server_uuid = 'your-server-uuid-here' | |
| server_instances = instance_api.get_instances(server_id=server_uuid) | |
| apps = {} | |
| inst = None # type: Instance | |
| for inst in server_instances: | |
| if inst.type is None: | |
| print("Missing UUID") | |
| continue | |
| if inst.type is InstanceType.PROD: | |
| print(u"Prod instance, skipping: {}".format(inst.id)) | |
| continue | |
| app_id = inst.app_id | |
| if app_id not in apps: | |
| apps[app_id] = app_api.get_app(app_id) | |
| task = None | |
| try: | |
| task = instance_api.upgrade_instance(inst.id) # type: ResponseTask | |
| print(u'Upgrading {} - {}'.format(apps[app_id].title, inst.title)) | |
| except ApiException as e: | |
| if e.status == 400: | |
| from json import loads | |
| data = loads(e.body) | |
| if data['error']['attributes']['errors'][0] == u'There is no new version available': | |
| print(u'OK: {} - {}'.format(apps[app_id].title, inst.title)) | |
| # Check if an error occurred | |
| if task is None: | |
| continue | |
| task = task.task # type: Task | |
| task_id = task.id | |
| interval = 10.0 | |
| waited = 0 | |
| while task.status != "done": | |
| task = task_api.get_task(task_id) # type: Task | |
| if task.status == "failed": | |
| print("Failed upgrade task.") | |
| break | |
| elif task.status == "canceled": | |
| print("Canceled: Upgrade task.") | |
| elif waited > 300: | |
| print("Expired 5 minutes TTL. Last task status was: {} (ID: {})".format(task.status, task.id)) | |
| else: | |
| # Try to wait a few seconds and check again. | |
| stdout.write('.') | |
| stdout.flush() | |
| time.sleep(interval) | |
| waited += interval | |
| # Slowly increase wait time by 20% on each wait. | |
| interval += (interval/5) | |
| stdout.write(': Finished in circa {} seconds!\n'.format(waited)) | |
| if __name__ == '__main__': | |
| main() |