# -*- coding: utf-8 -*- import Tea import json import uuid import platform import socket import time from io import BytesIO from datetime import datetime try: from urllib import urlencode except ImportError: from urllib.parse import urlencode from Tea.converter import TeaConverter from Tea.model import TeaModel from Tea.stream import READABLE class Client(object): """ This is a utility module """ class _ModelEncoder(json.JSONEncoder): def default(self, o): if isinstance(o, TeaModel): return o.to_map() elif isinstance(o, bytes): return TeaConverter.to_str(o) super(Client._ModelEncoder, self).default(o) @staticmethod def __read_part(f, size=1024): while True: part = f.read(size) if part: yield part else: return @staticmethod def __get_default_agent(): return 'AlibabaCloud (%s;%s) Python/%s Core/%s TeaDSL/1' % ( platform.system(), platform.machine(), platform.python_version(), Tea.__version__ ) @staticmethod def to_bytes(val): """ Convert a string(utf8) to bytes @return: the return bytes """ return TeaConverter.to_bytes(val) @staticmethod def to_string(val): """ Convert a bytes to string(utf8) @return: the return string """ return TeaConverter.to_string(val) @staticmethod def parse_json(val): """ Parse it by JSON format @return: the parsed result """ try: return json.loads(val) except ValueError: raise RuntimeError('Failed to parse the value as json format, Value: "%s".' % TeaConverter.to_str(val)) @staticmethod def read_as_bytes(stream): """ Read data from a readable stream, and compose it to a bytes @param stream: the readable stream @return: the bytes result """ if isinstance(stream, TeaConverter.basestring): return TeaConverter.to_bytes(stream) else: b = '' for part in Client.__read_part(stream, 1024): b += part return b @staticmethod def read_as_string(stream): """ Read data from a readable stream, and compose it to a string @param stream: the readable stream @return: the string result """ buff = Client.read_as_bytes(stream) return Client.to_string(buff) @staticmethod def read_as_json(stream): """ Read data from a readable stream, and parse it by JSON format @param stream: the readable stream @return: the parsed result """ return Client.parse_json(Client.read_as_string(stream)) @staticmethod def get_nonce(): """ Generate a nonce string @return: the nonce string """ name = socket.gethostname() + str(uuid.uuid1()) namespace = uuid.NAMESPACE_URL return str(uuid.uuid5(namespace, name)) @staticmethod def get_date_utcstring(): """ Get an UTC format string by current date, e.g. 'Thu, 06 Feb 2020 07:32:54 GMT' @return: the UTC format string """ return datetime.utcnow().strftime('%a, %d %b %Y %H:%M:%S GMT') @staticmethod def default_string(real, default): """ If not set the real, use default value @return: the return string """ return real if real is not None else default @staticmethod def default_number(real, default): """ If not set the real, use default value @return: the return number """ return real if real is not None else default @staticmethod def to_form_string(val): """ Format a map to form string, like a=a%20b%20c @return: the form string """ if not val: return "" keys = sorted(list(val)) dic = [(k, TeaConverter.to_str(val[k])) for k in keys if not isinstance(val[k], READABLE)] return urlencode(dic) @staticmethod def to_jsonstring(val): """ Stringify a value by JSON format @return: the JSON format string """ if isinstance(val, str): return str(val) return json.dumps(val, cls=Client._ModelEncoder) @staticmethod def empty(val): """ Check the string is empty? @return: if string is null or zero length, return true """ return not val @staticmethod def equal_string(val1, val2): """ Check one string equals another one? @return: if equals, return true """ return val1 == val2 @staticmethod def equal_number(val1, val2): """ Check one number equals another one? @return: if equals, return true """ return val1 == val2 @staticmethod def is_unset(value): """ Check one value is unset @return: if unset, return true """ return value is None @staticmethod def stringify_map_value(m): """ Stringify the value of map @return: the new stringified map """ if m is None: return {} dic_result = {} for k, v in m.items(): if v is not None: v = TeaConverter.to_string(v) dic_result[k] = v return dic_result @staticmethod def anyify_map_value(m): """ Anyify the value of map @return: the new anyfied map """ return m @staticmethod def assert_as_boolean(value): """ Assert a value, if it is a boolean, return it, otherwise throws @return: the boolean value """ if not isinstance(value, bool): raise ValueError('%s is not a bool' % TeaConverter.to_str(value)) return value @staticmethod def assert_as_string(value): """ Assert a value, if it is a string, return it, otherwise throws @return: the string value """ if not isinstance(value, TeaConverter.unicode): raise ValueError('%s is not a string' % TeaConverter.to_str(value)) return value @staticmethod def assert_as_bytes(value): """ Assert a value, if it is a bytes, return it, otherwise throws @return: the bytes value """ if not isinstance(value, bytes): raise ValueError('%s is not a bytes' % TeaConverter.to_str(value)) return value @staticmethod def assert_as_number(value): """ Assert a value, if it is a number, return it, otherwise throws @return: the number value """ if not isinstance(value, TeaConverter.number): raise ValueError('%s is not a number' % TeaConverter.to_str(value)) return value @staticmethod def assert_as_map(value): """ Assert a value, if it is a map, return it, otherwise throws @return: the map value """ if not isinstance(value, dict): raise ValueError('%s is not a dict' % TeaConverter.to_str(value)) return value @staticmethod def get_user_agent(user_agent): """ Get user agent, if it userAgent is not null, splice it with defaultUserAgent and return, otherwise return defaultUserAgent @return: the string value """ if user_agent: return '%s %s' % ( Client.__get_default_agent(), user_agent ) return Client.__get_default_agent() @staticmethod def is_2xx(code): """ If the code between 200 and 300, return true, or return false @return: boolean """ return 200 <= code < 300 @staticmethod def is_3xx(code): """ If the code between 300 and 400, return true, or return false @return: boolean """ return 300 <= code < 400 @staticmethod def is_4xx(code): """ If the code between 400 and 500, return true, or return false @return: boolean """ return 400 <= code < 500 @staticmethod def is_5xx(code): """ If the code between 500 and 600, return true, or return false @return: boolean """ return 500 <= code < 600 @staticmethod def validate_model(m): """ Validate model @return: void """ if isinstance(m, TeaModel): m.validate() @staticmethod def to_map(in_): """ Model transforms to map[string]any @return: map[string]any """ if isinstance(in_, TeaModel): return in_.to_map() else: return in_ @staticmethod def sleep(millisecond): """ Suspends the current thread for the specified number of milliseconds. """ time.sleep(millisecond / 1000) @staticmethod def to_array(input): """ Transform input as array. """ if input is None: return [] out = [] for i in input: if isinstance(i, TeaModel): out.append(i.to_map()) else: out.append(i) return out @staticmethod def assert_as_readable(value): """ Assert a value, if it is a readable, return it, otherwise throws @return: the readable value """ if isinstance(value, TeaConverter.basestring): value = BytesIO(TeaConverter.to_bytes(value)) if not isinstance(value, READABLE): raise ValueError('The value is not a readable') return value @staticmethod def assert_as_array(value): """ Assert a value, if it is a array, return it, otherwise throws @return the array value """ if not isinstance(value, list): raise ValueError('The value is not a list') return value