initial commit
This commit is contained in:
272
api/api_params.py
Executable file
272
api/api_params.py
Executable file
@@ -0,0 +1,272 @@
|
||||
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}"
|
||||
|
||||
|
||||
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 = "<ul>"
|
||||
for c in self.choices:
|
||||
if c[0] != '':
|
||||
ch += f"<li>{c[0]} - {c[1]}</li>"
|
||||
ch += "</ul>"
|
||||
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="<i>Токен</i>, выданный методом <code>account.auth</code>", **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="Телефон в формате <code>[[+]7]1112223333</code> "
|
||||
"<i>(в квадратных скобках необязательная часть)</i>", **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="Код верификации (требуется если клиенту будет отправлена "
|
||||
"одна из ошибок <i>верификации</i>)", **kwargs):
|
||||
super().__init__(name=name, required=False, value_min=0, value_max=9999, description=description, **kwargs)
|
||||
Reference in New Issue
Block a user