import re import traceback from datetime import datetime from flask import current_app from typing import List class OnlineHost: def __init__(self, id: str, url: str): self.id = id self.url = url # I decided to just use dict everywhere instead because I have to use dict to read it from json # class SSHHostKey: # def __init__(self, key_type=None, content=None, sha256=None): # self.key_type = key_type # self.content = content # self.sha256 = sha256 class VirtualMachine: def __init__(self, id, host, ipv4=None, ipv6=None, state="unknown", ssh_host_keys: List[dict] = list()): self.id = id self.host = host self.ipv4 = ipv4 self.ipv6 = ipv6 self.state = state self.ssh_host_keys = ssh_host_keys class VirtualizationInterface: def capacity_avaliable(self, additional_ram_bytes: int) -> bool: pass def get(self, id: str) -> VirtualMachine: pass def get_all_by_id(self) -> dict: pass def create(self, email: str, id: str, template_image_file_name: str, vcpus: int, memory: int, ssh_authorized_keys: list): pass def destroy(self, email: str, id: str): pass def vm_state_command(self, email: str, id: str, command: str): pass def net_set_dhcp(self, email: str, host_id: str, network_name: str, macs: list, remove_ipv4: str, add_ipv4: str): pass def validate_capsul_id(id): if not re.match(r"^(cvm|capsul)-[a-z0-9]{10}$", id): raise ValueError(f"vm id \"{id}\" must match \"^capsul-[a-z0-9]{{10}}$\"") def authorized_as_hub(headers): if headers.get('Authorization'): auth_header_value = headers.get('Authorization').replace("Bearer ", "") return auth_header_value == current_app.config["HUB_TOKEN"] return False def get_account_balance(vms, payments, as_of): vm_cost_dollars = 0.0 for vm in vms: vm_months = get_vm_months_float(vm, as_of) vm_cost_dollars += vm_months * float(vm["dollars_per_month"]) payment_dollars_total = float( sum(map(lambda x: 0 if x["invalidated"] else x["dollars"], payments)) ) return payment_dollars_total - vm_cost_dollars average_number_of_days_in_a_month = 30.44 def get_vm_months_float(vm, as_of): end_datetime = vm["deleted"] if vm["deleted"] else as_of days = float((end_datetime - vm["created"]).total_seconds())/float(60*60*24) # y / m / d date_when_minimum_run_time_was_changed_to_one_hour = datetime(2022, 2, 7) if vm["created"] > date_when_minimum_run_time_was_changed_to_one_hour: one_hour_in_days = float(1)/float(24) if days < one_hour_in_days: days = one_hour_in_days else: if days < 1: days = float(1) return days / average_number_of_days_in_a_month def my_exec_info_message(exec_info): traceback_result = traceback.format_tb(exec_info[2]) if isinstance(traceback_result, list): traceback_result = "\n".join(traceback_result) return f"{exec_info[0].__module__}.{exec_info[0].__name__}: {exec_info[1]}: {traceback_result}"