import asyncio from .api_errors import * from .models import AccessToken import re def _make_invalid_argument_type_error(name, value, except_type): related = {"param_name": name, "excepted_type": except_type, "value": value} raise Exception(API_ERROR_INVALID_ARGUMENT_TYPE, related) def _make_invalid_argument_value_error(name, value, err_code, message): related = {"param_name": name, "value": value, "error_code": err_code, "message": message} raise Exception(API_ERROR_INVALID_ARGUMENT_VALUE, related) class ApiParam: def __init__(self, name, description, required=True, default=None): self.name = name self.description = description self.required = required self.default = default def validate(self, value): """ Валидация параметра, в случае ошибки нужно выбросить исключение, соответствующее ошибке в случае успеха нужно вернуть объект, который и будет передан в метод в качестве параметра """ raise Exception(API_ERROR_INTERNAL_ERROR, f"param {self.name} have invalid definition (defined as super class)") def get_doc(self): return self.description def get_name(self): return self.name def is_required(self): return self.required def get_type_name(self): return f"{type(self)}" def to_json(self): return { "name": self.get_name(), "type": self.get_type_name(), "description": self.get_doc(), "required": self.is_required() } def __str__(self): return f"{type(self)}: name={self.name}" # Специальный класс параметра, нужен для получения доступа к raw request class ApiRequestParam(ApiParam): def __init__(self, name="request", description=None, **kwargs): super().__init__(name=name, description=description, **kwargs) def validate(self, value): return value def get_type_name(self): return "__internal_request" def get_doc(self): return None def to_json(self): return None class ApiParamStr(ApiParam): def __init__(self, regex=None, max_length=None, min_length=None, **kwargs): super().__init__(**kwargs) self.regex = None if regex is None else re.compile(regex) self.max_length = max_length self.min_length = min_length def validate(self, value): if value is None: if self.required: raise Exception(API_ERROR_MISSING_ARGUMENT, self.name) else: # вернуть значение по умолчанию без дополнительной проверки return self.default if self.regex is not None: if not self.regex.match(value): _make_invalid_argument_value_error(self.name, value, "invalid", f"expected string like this python regex: '{self.regex.pattern}'") if self.max_length is not None: if len(value) > self.max_length: _make_invalid_argument_value_error(self.name, value, "long", f"too long string. max size is '{self.max_length}' char(s)") if self.min_length is not None: if len(value) < self.min_length: _make_invalid_argument_value_error(self.name, value, "short", f"too short string. min size is '{self.max_length}' char(s)") return value def get_type_name(self): return "String" class ApiParamInt(ApiParam): def __init__(self, value_min=None, value_max=None, **kwargs): super().__init__(**kwargs) self.value_max = value_max self.value_min = value_min def _check_min_max(self, value): if self.value_min is not None: if self.value_min > value: _make_invalid_argument_value_error(self.name, value, 'out_of_range', f'the minimum value for this param is {self.value_min}') if self.value_max is not None: if self.value_max < value: _make_invalid_argument_value_error(self.name, value, 'out_of_range', f'the maximum value for this param is {self.value_max}') def validate(self, value): if value is None: if self.required: raise Exception(API_ERROR_MISSING_ARGUMENT, self.name) else: # вернуть значение по умолчанию без дополнительной проверки return self.default try: value = int(value) except Exception: _make_invalid_argument_type_error(self.name, value, self.get_type_name()) self._check_min_max(value) return value def get_doc(self): return super().get_doc().replace('{min}', str(self.value_min)).replace('{max}', str(self.value_max)) def get_type_name(self): return "Int" class ApiParamEnum(ApiParam): def __init__(self, choices, **kwargs): super().__init__(**kwargs) if not callable(choices): self.choices = [] for c in choices: self.choices.append( (c[0], c[1]) ) else: if asyncio.iscoroutinefunction(choices): loop = asyncio.get_event_loop() coroutine = choices() loop.run_until_complete(coroutine) else: self.choices = choices() async def validate(self, value): if value is None: if self.required: raise Exception(API_ERROR_MISSING_ARGUMENT, self.name) else: # вернуть значение по умолчанию без дополнительной проверки return self.default for choice in self.choices: cmp = choice[0] if type(choice[0]) is str else str(choice[0]) if cmp == value: return choice[0] _make_invalid_argument_value_error(self.name, value, 'out_of_range', f'expected value one of {[c[0] for c in self.choices]}') def get_doc(self): doc = super().get_doc() ch = "" return doc.replace('{choices}', ch) def get_type_name(self): return "Enum" class ApiParamBoolean(ApiParam): def __init__(self, **kwargs): super().__init__( **kwargs) def validate(self, value): if value is None: if self.required: raise Exception(API_ERROR_MISSING_ARGUMENT, self.name) else: # вернуть значение по умолчанию без дополнительной проверки return self.default if value == 'true' or value == '1' or value == 'y': return True elif value == 'false' or value == '0' or value == 'n': return False else: _make_invalid_argument_value_error(self.name, value, "invalid", "expected (true|y|1) or (false|n|0)") def get_type_name(self): return "Boolean" class ApiParamFloat(ApiParamInt): def validate(self, value): if value is None: if self.required: raise Exception(API_ERROR_MISSING_ARGUMENT, self.name) else: # вернуть значение по умолчанию без дополнительной проверки return self.default try: value = float(value) except Exception: _make_invalid_argument_type_error(self.name, value, self.get_type_name()) self._check_min_max(value) return value def get_doc(self): return super().get_doc().replace('{min}', str(self.value_min)).replace('{max}', str(self.value_max)) def get_type_name(self): return "Float" class ApiParamAccessToken(ApiParam): def __init__(self, regex=None, name="access_token", description="Токен, выданный методом account.auth", **kwargs): super().__init__(name=name, description=description, **kwargs) self.regex = (lambda: None if regex is None else re.compile(regex)) async def validate(self, value): if value is None: if self.required: raise Exception(API_ERROR_MISSING_ARGUMENT, self.name) else: # вернуть None, потому что параметра нет return None return await AccessToken.get_by_token(value) def get_type_name(self): return "AccessToken" class ApiParamPassword(ApiParamStr): def __init__(self, name="password", description="Пароль пользователя", **kwargs): super().__init__(name=name, description=description, regex="^[ -~а-яА-Я]{6,}$", **kwargs) class ApiParamPhone(ApiParamStr): def __init__(self, name="phone", description="Телефон в формате [[+]7]1112223333 " "(в квадратных скобках необязательная часть)", **kwargs): super().__init__(name=name, description=description, regex="^((\\+7)|(7)|)[0-9]{10}$", **kwargs) def validate(self, value): value = super(ApiParamPhone, self).validate(value) if value is not None: # Гоша попросил запилить фичу, чтобы принимались номера: # +79991112233 # 79991112233 # 9991112233 if re.match("^[0-9]{10}$", value) is not None: return f"+7{value}" elif re.match("^7[0-9]{10}$", value) is not None: return f"+{value}" return value class ApiParamVerifyCode(ApiParamInt): def __init__(self, name="code", description="Код верификации (требуется если клиенту будет отправлена " "одна из ошибок верификации)", **kwargs): super().__init__(name=name, required=False, value_min=0, value_max=9999, description=description, **kwargs)