Skip to content

Создание нового модуля

Jump to bottom
Konstantin Zagorulko edited this page May 16, 2022 · 2 revisions

При создании нового модуля пользуемся правилами:

Создание

Если создаёте часть ядра для новой сущности, создайте отдельную папку, в папку могут входить файлы models.py, resources.py, utils.py.

  • После создания модели, она добавляется core/models.py. Импортировать модели также стоит из core.models.py.

  • После создания ресурса нужно добавить в конец файла переменную routes, заполните её роутами. Затем следует добавить ресурс в core/routes.py.

Пример:

Нужно сделать роуты для пользователей.

  1. Создаём папку users в core.
  2. Создаём модель UserModel в файле core/users/models.py:
from ..database import db


class UserModel(db.Model):
    __tablename__ = 'users'

    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(50), unique=True, nullable=False)
    password = db.Column(db.String(120), nullable=False)
    
    def jsonify(self):
        return {
            'id': self.id,
            'username': self.username,
        }
  1. Добавляем модель в core/models.py:
from .users.models import UserModel

__all__ = ['UserModel']
  1. Создаём миграцию

В Terminal прописываем:

Если есть make:

make db-migrate
make db-upgrade

Если нет:

alembic revision --autogenerate
alembic upgrade head
  1. Создаём эндпойт или функцию:
from starlette.endpoints import HTTPEndpoint
from starlette.responses import JSONResponse

from ..models import UserModel
from ..utils import make_error


class User(HTTPEndpoint):
    @staticmethod
    async def get(request):
        user_id = request.path_params['user_id']
        user = await UserModel.get(user_id)
        if user:
            return JSONResponse(user.jsonify())
        return make_error(description='User not found', status_code=404)
            
async def ping(request):
    return JSONResponse({'onPing': 'wePong'})
  1. Добавляем его в переменную routes:
routes = [
    Route('/', User),
    Route('/ping', ping, methods=['GET'])
]

обратите внимание на путь эндпойнта '/'. Префикс эндпойнта не нужно указывать при роутинге в вашем ресурсе, он будет указан в дальнейшем.

  1. Добавляем роуты юзера к общим роутам:
from starlette.routing import Mount

from .users.resources import routes as users_routes

routes = [
    Mount('/users', routes=users_routes),
]

__all__ = ['routes']

Затестить запросы можно с помощью постмана

Надо сначала залогиниться, скопировать Access token и вставить его в Authorization -> Bearer Token

image

Декораторы

Сейчас мы используем декораторы: @with_transaction, @staticmethod, @jwt_required, @permissions.required, @validate.

with_transaction

Используется, когда что-то изменяется в базе данных. То есть во всех методах, кроме get.

staticmethod

Используется в гет методах, если вам не нужно обращаться к классу через self. IDE обычно подствечивают метод, который можно сделать статическим.

jwt_required

jwt_required имеет опциональный интрефейс. Можно писать просто

@jwt_required(return_user=False)
async def get_blank_response(request):
    return Response('', status_code=204)

или

@jwt_required
async def get_user_username(request, user):
    return Response(user.username)

как видно, return_user стоит по умолчанию в значении True. И если вам он не нужен, то можете убрать его, передпав атрибут return_user со значением False. Также в этом декораторе имеется опциональный параметр token_type, но его вряд ли придётся использовать.

Permissions

В отличие от прошлых декораторов, этот является не функцией, а классом поэтому его следует объявлять заранее с указанием имени приложения, в котором он будет работать:

from ..utils import Permissions

permissions = Permissions(app_name='users')

Использовать его следует, когда ваш метод должен быть ограничен полномочиями.

@with_transaction
@jwt_required
@permissions.required(action='delete')
async def delete_all_users(request):
    await Users.delete.gino.status()
    return Response('', status_code=204)

также у него есть аргументы: return_user и return_role, которые по умолчанию установлены в False

validate

Декоратор используется для валидации.

Возвращает словарь ключ-значение data или если возникает ошибка валидации, то игнорирует тело функции и просто возвращает ошибку через make_error.

Принимает в себя переменные: schema - это схема проверки, return_request (возвращает реквест, например, чтобы получить path_params) и custom_checks (пока не будем использовать). Поля не включённые в схему проверки, будут отброшены.

Имеет вид:

@validation(schema={
    'email': {
         'type':str,
    }
})
async def post(data):
    print(data['email'])

Есть базовые проверки:

  1. type - проверяет тип данных, можно проверить несколько типов:
@validation(schema={
    'username': {
        'type': str,
    },
    'price': {
        'type': (int, float),
    },
})
async def post(data):
    print(data['username'], data['price'])
  1. Максимальная/минимальная длинна поля
@validation(schema={
    'password':  {
        'min_length': 5,
        'max_length': 50,
    },
})
async def post(data):
    print(data['password'])
  1. Обязательность
@validation(schema={
    'password':  {
        'required': True,
    },
})
async def post(data):
    print(data['password'])

Если поле необязательное - просто не указывайте этот параметр.

  1. Custom check
# Сначала прописывается custom_check (лучше вынести это в utils)

# users/utils.py
from ..models import UserModel


async def is_username_unique(username):
    user = await UserModel.query.where(
        UserModel.username == username
    ).gino.first()
    if user:
        return False
    return True

# users/routes.py
from .utils import is_username_unique


@validate(schema={
    'username': {
        'unique_username': True,
    }, custom_checks= {
        'unique_username': {
            # it works with async functions
            'func': lambda v, *args: is_username_unique(v),
            'message': lambda v, *args: f'Пользователь с `username` `{v}` уже '
            f'существует.'
        },
    }
})
def post(data):
    return make_response({'result': 'Имя пользователя уникально'})

также можно передавать func и без лямбды:

'role': {
   'func': validate_role,
   'message': lambda v, *args: f'Роль с `id` `{v}` не существует.'
}

но для этого аргументы функции должны иметь вид

async def validate_role(role_id, *args):
    role = await RoleModel.get(role_id)
    return bool(role)

Запросы

  • параметры из пути (/users/1) (получить 1)
user_id = request.path_params['user_id']
  • параметр запроса (`/users/1?role=admin) (роль юзера)
role = request.query_params['role']

Ответы

При ответах придерживаемся следующих правил:

  • При создании вернуть id новой сущности (make_response)
  • При обновлении данных, вернуть константу NO_CONTENT
  • При ошибке вызвать функцию utils/make_error, передать описание ошибки и указать status_code (JSONResponse)

Именование

Примеры:

путь: /users/

общий путь, через него обращаются к сущности. Именуем Users.

путь: /users/{user_id:int} через него обращаются к конкретнуму пользователю. Именуем User.

Clone this wiki locally