import random
from django.core.exceptions import ValidationError
from django.http import HttpResponse, HttpResponseBadRequest
from .api_utils import *
from .api_params import *
from .models import *
from .api_media_utils import *
import time
def _make_model_validation_errors(validation_error: ValidationError, api_err=API_ERROR_OBJECT_VALIDATION):
errors = {}
for field_name in validation_error.error_dict:
err_list = validation_error.error_dict[field_name]
print(err_list)
obj = []
for err in err_list:
obj.append(str(err.code))
errors[field_name] = obj
return make_error_object(Exception(api_err, errors))
class ApiAccount:
@staticmethod
def __check_phone_code(phone, code):
if code is None:
res, err_code = PhoneVerificationService.send_verify(phone)
if not res:
if err_code in API_ERROR_VERIFICATION:
raise Exception(API_ERROR_VERIFICATION[err_code])
else:
raise Exception(API_ERROR_VERIFY_UNKNOWN)
raise Exception(API_ERROR_NEED_VERIFY)
else:
res, err_code = PhoneVerificationService.check_code(phone, code)
if not res:
if err_code in API_ERROR_VERIFICATION:
raise Exception(API_ERROR_VERIFICATION[err_code])
else:
raise Exception(API_ERROR_VERIFY_UNKNOWN)
return True
@staticmethod
def __make_user_json(user: Account, self_using=False):
obj = {
"id": user.id,
"name": user.name,
"surname": user.surname,
"phone": user.phone,
"email": user.email,
"about": user.about,
"city": {"code": user.city, "name": CITIES_CHOICES[user.city]} if user.city is not None else None,
"register_datetime": int(time.mktime(user.register_datetime.timetuple())),
"role": user.role,
}
if hasattr(user, 'accountavatar'):
obj["avatar"] = user.accountavatar.photo.id if user.accountavatar.photo is not None else None
obj["profile_background"] = \
user.accountavatar.profile_background.id if user.accountavatar.photo is not None else None
else:
obj["avatar"] = None
obj["profile_background"] = None
if user.role == Account.ROLE_EXECUTOR:
obj |= {
"executor_type": user.executoraccount.executor_type,
"executor_inn": user.executoraccount.inn,
"executor_info": user.executoraccount.additional_info
}
return obj
@staticmethod
@api_method("account.register",
doc="Регистрация нового пользователя",
params=[
ApiParamEnum(name="role", choices=Account.ROLE_CHOICES, description="Роль пользователя: {choices}"),
ApiParamPhone(),
ApiParamVerifyCode(),
ApiParamPassword(required=False,
description=f"Пароль пользователя, требуется для завершения регистрации. "
f"Если код верификации будет принят, но пароль не передан"
f"метод вернет ошибку {API_ERROR_MISSING_ARGUMENT}"),
], returns="Аналогично методу account.auth в случае успеха")
async def register(role, phone, password, code):
user = Account.create_user(
phone=phone,
password=(password if password is not None else ""),
role=role
)
try:
await sync_to_async(user.full_clean)()
except ValidationError as validation_error:
# traceback.print_exc()
return _make_model_validation_errors(validation_error, API_ERROR_USER_REGISTER)
if ApiAccount.__check_phone_code(user.phone, code):
if password is None:
raise Exception(API_ERROR_MISSING_ARGUMENT, "password")
user = await Account.objects.acreate(phone=phone, password=password, role=role)
if role == Account.ROLE_EXECUTOR:
await ExecutorAccount.objects.acreate(account=user)
# удаляем код, типа завершение транзакции
PhoneVerificationService.check_code(phone, code, auto_pop=True)
try:
token = await AccessToken.create_token(user)
return api_make_response({"access_token": token.access_token})
except Exception as ex:
# если вдруг токен нельзя создать
user.delete()
raise ex
@staticmethod
@api_method("account.auth",
doc="Аутентификация пользователя",
params=[
ApiParamPhone(name="login", description="Логин пользователя"),
ApiParamPassword(),
],
returns="В случае правильных логина и пароля access_token. "
"В противном случае объект ошибки.")
async def auth(login, password):
return api_make_response({"access_token": (await AccessToken.auth(login, password)).access_token})
@staticmethod
@api_method("account.deauth",
doc="Удаление токена, дальшейшие вызовы API с этим токеном вернут ошибку невалидного токена",
params=[
ApiParamAccessToken(),
], returns="В случае успеха стандартный код успеха")
async def deauth(access_token):
await AccessToken.deauth(access_token.access_token)
return api_make_response({})
@staticmethod
@api_method("account.delete",
doc='Удаление аккаунта пользователя БЕЗ ВОЗМОЖНОСТИ ВОССТАНОВЛЕНИЯ. '
'Так же будут удалены ВСЕ связанные с аккаунтом данные.',
params=[ApiParamAccessToken()],
returns="Стандартный ответ успеха, в случае успеха")
async def delete(access_token):
user = access_token.owner
await sync_to_async(user.delete)()
return api_make_response({})
@staticmethod
@api_method("account.get",
doc="Получение информации о пользователе",
params=[
ApiParamAccessToken(),
ApiParamInt(name="user_id", required=False, value_min=0,
description="ID пользователя, аккаунт которого нужно вернуть. "
"Если не указывать, вернет аккаунт владельца.")
],
returns="Поля пользователя (name, surname, email, phone и прочие).")
async def get(access_token, user_id):
if user_id is None:
user = access_token.owner
else:
user = await access_token.owner.get_by_id(user_id)
if user is None:
return make_error_object(Exception(API_ERROR_NOT_FOUND, {"user": user_id}))
return api_make_response(ApiAccount.__make_user_json(user))
@staticmethod
@api_method("account.edit",
doc="Редактирование основной информации о пользователе. "
"Будут изменены только те данные, которые будут переданы в запросе, остальные не будут изменены.",
params=[
ApiParamAccessToken(),
ApiParamStr(name="name", description="Имя пользователя",
min_length=2, max_length=60, required=False),
ApiParamStr(name="surname", description="Фамилия пользователя",
min_length=2, max_length=60, required=False),
ApiParamStr(name="about", description="Текст о себе. максимум - 1000 символов",
max_length=1000, required=False),
ApiParamEnum(name="executor_type", choices=ExecutorAccount.EXECUTOR_TYPE_CHOICES, required=False,
description="Тип исполнителя (только для роли исполнитель): {choices}"),
ApiParamStr(name="executor_inn", required=False, regex="^(\\d{12}|\\d{10})$",
description="ИНН исполнителя (только для роли исполнитель): "
"12 цифр если исполнитель - физ.лицо и 10 цифр если это юр. лицо"),
ApiParamEnum(name="city", description="Город, в котором находится ползователь: {choices}",
required=False, choices=CITIES_CHOICES)
],
returns="Вернет основную информацию о пользователе, иначе ошибки")
async def edit(access_token, name, surname, about, executor_type, executor_inn, city):
user = access_token.owner
executor_need_save, need_save = False, False
if name is not None:
user.name = name
need_save = True
if surname is not None:
user.surname = surname
need_save = True
if about is not None:
user.about = about
need_save = True
if city is not None:
user.city = city
need_save = True
if user.role == Account.ROLE_EXECUTOR:
print("Executor account detected")
if executor_type is not None:
print("Executor account type detected")
user.executoraccount.executor_type = executor_type
executor_need_save = True
if executor_inn is not None:
print("Executor account inn detected")
t = executor_type
if t is None:
t = user.executoraccount.executor_type
if t is None:
raise Exception(API_ERROR_MISSING_ARGUMENT, "executor_type")
if t == ExecutorAccount.EXECUTOR_TYPE_SELF_EMPLOYED:
if re.match("^\\d{12}$", executor_inn) is None:
raise Exception(API_ERROR_INVALID_ARGUMENT_VALUE, "executor_inn must be defined as 12 digits")
if t == ExecutorAccount.EXECUTOR_TYPE_LEGAL_ENTITY:
if re.match("^\\d{10}$", executor_inn) is None:
raise Exception(API_ERROR_INVALID_ARGUMENT_VALUE, "executor_inn must be defined as 10 digits")
user.executoraccount.inn = executor_inn
executor_need_save = True
if need_save:
try:
await sync_to_async(user.full_clean)()
except ValidationError as ve:
return _make_model_validation_errors(ve, API_ERROR_USER_MODIFY)
await sync_to_async(user.save)()
if executor_need_save:
try:
await sync_to_async(user.executoraccount.full_clean)()
except ValidationError as ve:
return _make_model_validation_errors(ve, API_ERROR_USER_MODIFY)
await sync_to_async(user.executoraccount.save)()
return api_make_response(ApiAccount.__make_user_json(user))
@staticmethod
@api_method("account.changePhone",
doc="Смена пароля. Для подтверждения требуется старый пароль.",
params=[
ApiParamAccessToken(),
ApiParamPassword(description="Пароль пользователя, нужен для подтверждения владельца аккаунта"),
ApiParamPhone(description="Новый телефон пользователя"),
ApiParamVerifyCode(description="Код подтверждения операции, придет на новый телефон")
],
returns="Вернет стандартный объект успеха")
async def change_phone(access_token, password, phone, code):
user = access_token.owner
if not user.check_password(password):
raise Exception(API_ERROR_INVALID_PASSWORD)
# чекаем телефон
user.phone = phone
try:
await sync_to_async(user.full_clean)()
except ValidationError as ve:
return _make_model_validation_errors(ve, API_ERROR_USER_MODIFY)
if ApiAccount.__check_phone_code(user.phone, code):
await Account.objects.filter(id=user.id).aupdate(phone=phone)
PhoneVerificationService.check_code(user.phone, code, auto_pop=True)
return api_make_response({})
@staticmethod
@api_method("account.resetPassword",
doc="Смена пароля. Для подтверждения действия требуется код с телефона. Так же в случае успешного "
"восстановления удалит все существующие сессии пользователя.",
params=[
ApiParamPhone(description="Телефон, для которого нужно сделать восстановление"),
ApiParamPassword(name="new_password", required=False,
description="Новый пароль, нужен после того, как будет отправлен код ("
"можно передать раньше, тогда он будет проверен сразу)"),
ApiParamVerifyCode(description="Код подтверждения операции, придет на телефон")
],
returns="Вернет стандартный объект успеха")
async def reset_password(new_password, phone, code):
user = await Account.get_by_natural_key(phone)
if new_password is not None:
user.password = new_password
try:
await sync_to_async(user.full_clean)()
except:
traceback.print_exc()
raise Exception(API_ERROR_INVALID_PASSWORD)
if ApiAccount.__check_phone_code(phone, code):
if code is not None and new_password is None:
raise Exception(API_ERROR_MISSING_ARGUMENT, "new_password")
await sync_to_async(user.save)()
await AccessToken.delete_sessions(user)
PhoneVerificationService.check_code(phone, code, auto_pop=True)
return api_make_response({})
class ApiSecurity:
@staticmethod
@api_method("security.listSessions",
doc="Получение списка сессий (кроме текущей)",
params=[
ApiParamAccessToken(),
ApiParamPassword(description="Пароль пользователя, нужен для подтверждения личности")
],
returns="Вернет sessions: [{id: int, name: str, created: unix_timestamp}]")
async def list_sessions(access_token, password):
sessions = await access_token.list_sessions()
if not access_token.owner.check_password(password):
raise Exception(API_ERROR_INVALID_PASSWORD)
return api_make_response({
"current": {
"id": access_token.id,
"created": int(time.mktime(access_token.creation_time.timetuple())),
},
"sessions": [
{
"id": s.id,
"name": f"{s.access_token[:8]}...",
"created": int(time.mktime(s.creation_time.timetuple())),
} for s in sessions
]
})
@staticmethod
@api_method("security.removeOtherSessions",
doc="Получение списка сессий (кроме текущей)",
params=[
ApiParamAccessToken(),
ApiParamPassword(description="Пароль пользователя, нужен для подтверждения личности")
],
returns="Вернет sessions: [{id: int, name: str, created: unix_timestamp}]")
async def remove_other_sessions(access_token, password):
if not access_token.owner.check_password(password):
raise Exception(API_ERROR_INVALID_PASSWORD)
sessions = await access_token.list_sessions()
response = {
"current": {
"id": access_token.id,
"created": int(time.mktime(access_token.creation_time.timetuple())),
},
"removed": [
{
"id": s.id,
"name": f"{s.access_token[:8]}...",
"created": int(time.mktime(s.creation_time.timetuple())),
} for s in sessions
]
}
await access_token.delete_sessions_without_current()
return api_make_response(response)
@staticmethod
@api_method("security.removeSession",
doc="Удалит сессию с указанным id. Не удаляет текущую сессию, даже если указан ее ID "
"(будет возвращена ошибка NOT FOUND)",
params=[
ApiParamAccessToken(),
ApiParamPassword(description="Пароль пользователя, нужен для подтверждения личности"),
ApiParamInt(name="session", description="ID сессии, которую надо деактивировать", value_min=0)
],
returns="Вернет стандартный отъект в случае успеха")
async def remove_session(access_token, password, session):
if not access_token.owner.check_password(password):
raise Exception(API_ERROR_INVALID_PASSWORD)
await access_token.delete_session(session)
return api_make_response({})
@staticmethod
@api_method("security.changePassword",
doc="Смена пароля. Для подтверждения требуется старый пароль.",
params=[
ApiParamAccessToken(),
ApiParamPassword(name="old_password", description="Старый пароль пользователя"),
ApiParamPassword(description="Новый пароль пользователя"),
],
returns="Вернет стандартный объект успеха")
async def change_password(access_token, old_password, password):
user = access_token.owner
if not user.check_password(old_password):
raise Exception(API_ERROR_INVALID_PASSWORD, "old_password")
user.password = password
await sync_to_async(user.save)()
return api_make_response({})
class ApiOrder:
@staticmethod
@api_method("order.getForm",
doc="Получение формы создания заказа в виде полей, которые нужно показать пользователю",
params=[],
returns="JSON объект, содержащий данные формы. Структура пока не определена")
def get_form():
return api_make_response({
"fields": [
# name = models.CharField(max_length=200, verbose_name="Название заказа")
{
"name": "name",
"label": "Название заказа",
"widget": "text",
"attrs": {
"max_len": 200
},
"required": True
},
# square = models.DecimalField(max_digits=7, decimal_places=2, blank=False, verbose_name="Площадь в м²")
{
"name": "square",
"label": "Площадь в м²",
"widget": "decimal",
"attrs": {
"max_digits": 7,
"decimal_places": 2
},
"required": True
},
# work_time = models.CharField(max_length=100, blank=True, verbose_name="Рабочее время")
{
"name": "work_time",
"label": "Рабочее время",
"widget": "text",
"attrs": {
"max_len": 100
},
"required": False
},
# type_of_renovation = models.CharField(max_length=10, choices=TYPE_OF_RENOVATION_CHOICES,
# default=CHOICE_UNDEFINED, blank=True, verbose_name="Тип ремонта")
{
"name": "type_of_renovation",
"label": "Тип ремонта",
"widget": "choice",
"attrs": {
"default": None,
"choices": "Order.TYPE_OF_RENOVATION_CHOICES"
},
"required": False
},
# type_of_room = models.CharField(max_length=10, choices=TYPE_OF_ROOM_CHOICES,
# blank=True, default=CHOICE_UNDEFINED, verbose_name="Тип квартиры")
{
"name": "type_of_room",
"label": "Тип квартиры",
"widget": "radio",
"attrs": {
"default": None,
"choices": "Order.TYPE_OF_ROOM_CHOICES"
},
"required": False
},
# флажок
{
"name": "is_with_warranty",
"label": "С гарантией",
"widget": "checkbox",
"attrs": {
"default": True
},
"required": False
},
]
})
@staticmethod
@api_method("order.create",
doc="Создание заказа",
params=[
ApiParamAccessToken(),
ApiParamStr(name='name', max_length=200, description="Название заказа"),
ApiParamStr(name='description', max_length=1000, description="Описание заказа",
required=False, default=""),
ApiParamFloat(name='square', value_max=99999.99, value_min=1.0,
description='Площадь в м²'),
ApiParamStr(name='work_time', max_length=100, description="Рабочее время",
required=False, default=""),
ApiParamEnum(name='type_of_renovation', choices=Order.TYPE_OF_RENOVATION_CHOICES, required=False,
default=Order.CHOICE_UNDEFINED,
description="Тип ремонта: {choices}"),
ApiParamEnum(name='type_of_house', choices=Order.TYPE_OF_HOUSE_CHOICES, required=False,
default=Order.CHOICE_UNDEFINED,
description="Тип дома: {choices}"),
ApiParamEnum(name='type_of_room', choices=Order.TYPE_OF_ROOM_CHOICES, required=False,
default=Order.CHOICE_UNDEFINED,
description="Тип квартиры: {choices}"),
ApiParamEnum(name='purchase_of_material', choices=Order.PURCHASE_OF_MATERIAL_CHOICES,
required=False, default=Order.CHOICE_UNDEFINED,
description="Закуп материала: {choices}"),
ApiParamEnum(name='type_of_executor', choices=Order.TYPE_OF_EXECUTOR_CHOICES,
required=False, default=Order.CHOICE_UNDEFINED,
description="Тип исполнителя: {choices}"),
# дальше отдельные флаги
ApiParamBoolean(name="is_with_warranty", required=False, default=True,
description="С гарантией"),
ApiParamBoolean(name="is_with_contract", required=False, default=False,
description="Работа по договору"),
ApiParamBoolean(name="is_with_trade", required=False, default=False,
description="Возможен торг"),
ApiParamBoolean(name="is_with_cleaning", required=False, default=False,
description="С уборкой"),
ApiParamBoolean(name="is_with_garbage_removal", required=False, default=False,
description="С вывозом мусора"),
ApiParamBoolean(name="is_require_design", required=False, default=False,
description="Требуется дизайн проект"),
# примерная цена
ApiParamFloat(name='approximate_price', value_max=9999999999.99, value_min=1.0,
description='Примерная цена'),
# date_start = models.DateField(null=True, blank=True, default=None, verbose_name="Дата начала")
# date_end = models.DateField(null=True, blank=True, default=None, verbose_name="Дата окончания")
ApiParamEnum(name="address_city", description="Город: {choices}", choices=CITIES_CHOICES),
ApiParamStr(name='address_text', max_length=70, description="Улица, дом",
required=False, default="")
# email = models.EmailField(null=True, blank=True, verbose_name="Email")
# phone = models.CharField(null=True, blank=True, max_length=16, verbose_name="Телефон")
],
returns="ID созданного заказа, иначе одну из ошибок")
async def create(**kwargs):
access_token = kwargs.pop('access_token')
ApiOrder._check_write_permissions(access_token)
try:
order = await Order.objects.acreate(owner=access_token.owner, **kwargs)
return api_make_response({"order_id": order.id})
except ValidationError as ve:
return _make_model_validation_errors(ve, API_ERROR_USER_MODIFY)
@staticmethod
@api_method("order.setPublished",
doc="Установка статуса публикации заказа",
params=[
ApiParamAccessToken(),
ApiParamInt(name='order_id', value_min=0, description='ID заказа'),
ApiParamBoolean(name='value', description='Значение поля "опубликован"')
],
returns="Обновленный объект заказа")
async def set_published(access_token, order_id, value):
ApiOrder._check_write_permissions(access_token)
query = Order.objects.filter(id=order_id)
order = await query.afirst()
if order.owner_id != access_token.owner.id:
raise Exception(API_ERROR_ACCESS_DENIED, 'edit operation allowed only for owner')
await query.aupdate(published=value)
# вернем в зад обновленный объект
order.published = value
return api_make_response(ApiOrder._order_to_json(order))
ORDER_GET_ORDERING = [
('create_time', 'время создания, сначала новые записи'),
('-create_time', 'время создания, сначала старые записи'),
]
@staticmethod
def _order_to_json(order):
return order.__dict__
@staticmethod
@api_method("order.get",
doc="Получение объектов заказа, применит все фильтры",
params=[
ApiParamAccessToken(),
ApiParamInt(name='order_id', required=False, value_min=0,
description='ID заказа, который нужно вернуть, вернет один заказ, является фильтром'),
ApiParamInt(name='user_id', required=False, value_min=0,
description='Показать заказы для конкретного пользователя'),
ApiParamEnum(name='ordering', choices=ORDER_GET_ORDERING, required=False,
description='Сортировка заказов. Возможные значения: {choices} '
'Не имеет эффекта если ответ состоит из одного заказа.'),
],
returns="Массив заказов, соответсвующий всем указанным фильтрам.")
async def get(access_token, order_id, user_id, ordering):
query = Order.objects
if order_id is not None:
res = await query.aget(id=order_id)
if user_id is not None:
if access_token.owner.id == res.owner_id or (res.published and res.moderated):
return api_make_response([ApiOrder._order_to_json(res)])
else:
raise Exception(API_ERROR_NOT_ALLOWED, 'attempt access to closed order')
if user_id is not None:
user = await access_token.owner.get_by_id(user_id)
if user is None:
raise Exception(API_ERROR_NOT_FOUND, 'user')
if user.role != Account.ROLE_CUSTOMER:
raise Exception(API_ERROR_NOT_ALLOWED, 'target user is not customer')
query = query.filter(owner_id=user_id)
if ordering is not None:
query = query.order_by(ordering)
return api_make_response([ApiOrder._order_to_json(item) async for item in query.all()])
@staticmethod
def _check_write_permissions(access_token):
if not access_token.owner.is_completed():
raise Exception(API_ERROR_NEED_COMPLETED_ACCOUNT)
if access_token.owner.role != Account.ROLE_CUSTOMER:
raise Exception(API_ERROR_NOT_ALLOWED, 'you must be a customer')
class ApiMedia:
@staticmethod
def __filename_to_ext(filename: str):
formats = {
".jpeg": "jpeg",
".png": "png",
".pdf": "pdf"
}
for k in formats:
if filename.endswith(k):
return formats[k]
return None
@staticmethod
def __ext_to_content_type(ext: str):
formats = {
"jpeg": "image/jpeg",
"png": "image/png",
"pdf": "application/pdf"
}
if ext in formats:
return formats[ext]
return "application/binary"
@staticmethod
@api_method("media.upload",
doc="Загрузка медиа на сервер. Вызывать методом POST. Обязательно должен быть прикреплен файл"
" с именем поля 'file'",
params=[
ApiRequestParam(),
ApiParamAccessToken(),
], returns="id медиа, в противном случае ошибку")
async def upload(request, access_token):
if request.method != "POST":
return make_error_object(Exception(API_ERROR_INVALID_REQUEST, "method must be executed http POST method"))
if 'file' not in request.FILES:
return make_error_object(Exception(API_ERROR_INVALID_REQUEST,
"you must attach file with field-name 'file'"))
filename = request.FILES['file'].name
print(filename, type(filename))
# if not access_token.user.is_completed():
# return make_error_object(Exception(API_ERROR_NEED_COMPLETED_ACCOUNT))
ext = ApiMedia.__filename_to_ext(filename)
if ext is None:
return make_error_object(Exception(API_ERROR_INVALID_REQUEST, "unsupported file extension"))
try:
storage_name = Media.generate_storage_name(filename, datetime.now(), access_token.owner)
await sync_to_async(s3_upload_from_buffer)(storage_name, request.FILES['file'].read())
m = await Media.objects.acreate(user=access_token.owner, original_name=filename,
extension=ext, storage_name=storage_name)
return api_make_response({'media_id': m.id})
except Exception:
traceback.print_exc()
return make_error_object(Exception(API_ERROR_INTERNAL_ERROR, "try to upload file after 5 minutes"))
@staticmethod
@api_method("media.get",
doc="Получение медиа",
params=[
ApiRequestParam(),
ApiParamAccessToken(),
ApiParamInt(name="media_id", description="ID медиа",
value_min=0, value_max=1000000000),
], returns="медиа, в противном случае ошибку")
async def get(request, access_token, media_id):
if request.method != "GET":
return make_error_object(Exception(API_ERROR_INVALID_REQUEST, "method must be executed http GET method"))
m = await Media.objects.filter(Q(owner=access_token.user) |
Q(owner__account__photo_id=media_id) |
Q(owner__account__profile_background_id=media_id)).filter(pk=media_id).afirst()
if m is not None:
try:
# Media.get_by_id(access_token.user, m_id)
res = HttpResponse(content=s3_get(m.storage_name))
res.headers["Content-type"] = "image/jpg"
return res
except Exception:
traceback.print_exc()
return make_error_object(Exception(API_ERROR_NOT_FOUND, "object in storage not found"))
@staticmethod
@api_method("media.list",
doc="Получение списка всех загруженных медиа для текущего пользователя",
params=[
ApiParamAccessToken(),
ApiParamInt(name="count", description="Количество объектов, максимум 100, по умолчанию 25",
value_min=1, value_max=100, default=25, required=False),
ApiParamInt(name="offset", description="Смещение списка относительно начала",
value_min=0, value_max=100000, default=0, required=False),
], returns="Список с объектами media.")
async def get_list(access_token, count, offset):
ms = [item async for item in
Media.objects.filter(owner=access_token.user, ).order_by('upload_datetime')[offset:offset + count]
]
res = [{
'id': m.id,
'name': m.original_name,
'upload_time': int(time.mktime(m.upload_datetime.timetuple())),
'extension': m.extension
} for m in ms]
return api_make_response(res)
class DatabaseApi:
# TODO переместить сюда форму заказа, список городов
# def get_order_form
# def get_citys
pass
async def api_call_method(request, method_name, params: dict):
if method_name in api_methods_dict:
return await api_methods_dict[method_name]["func"](__raw_request=request, **params)
else:
return make_error_object(Exception(API_ERROR_METHOD_NOT_FOUND))
def api_get_documentation():
out = []
for m in api_methods_dict:
params = []
for p in api_methods_dict[m]["params"]:
j = p.to_json()
if j is not None:
params.append(j)
out.append({
"name": m,
"doc": api_methods_dict[m]["doc"],
"returns": api_methods_dict[m]["returns"],
"params": params
})
return out