Авторизация пользователя при помощи Starlette + Vue.js

Вступление

Задача — создать пример авторизации пользователя с использованием фреймворков Starlette (https://www.starlette.io/) и Vue.js *, который был бы максимально комфортным разработчикам Django для «миграции» в асинхронный стек.

Почему Starlette? В первую очередь скорость. Starlette ультимативно быстр, и в тестах уступает только BlackSheep (https://pypi.org/project/blacksheep/). Во вторых Starlette весьма прост и писать на нем в силу его продуманности легко и приятно.

В качестве ORM мы будем использовать Tortoise ORM (со моделями и выборками «аля Django ORM»).

В качестве сессионного механизма мы будем использовать JWT.

* Описание фронтенда на Vue.js не входит в данную заметку.

Структура проекта

apps/user/models.py — модель пользователя
apps/user/urls.py — роутер
apps/user/views.py — регистрация и логин
.env — наши переменные
settings.py — общие настройки проекта
app.py — точка входа
middleware.py — промежуточный слой для работы с JWT

Файл с переменными .env

Объявим здесь переменные, которые нам в дальнейшем понадобятся для работы:

DEBUG=True DATABASE_URL=postgres://user:123456@localhost/svue_backend_db ALLOWED_HOSTS=127.0.0.1, localhost, local SECRET_KEY=AGe-lJvQslHjNdqOa2_Wwy9JB3GE3d8GzMfC418I6jc JWT_PREFIX=Bearer JWT_ALGORITHM=HS256 

Общие настройки проекта settings.py

config = Config(".env") DEBUG = config("DEBUG", cast=bool, default=False) DATABASE_URL = config("DATABASE_URL", cast=str) SECRET_KEY = config("SECRET_KEY", cast=Secret) ALLOWED_HOSTS = config("ALLOWED_HOSTS", cast=CommaSeparatedStrings) JWT_PREFIX = config("JWT_PREFIX", cast=str) JWT_ALGORITHM = config("JWT_ALGORITHM", cast=str) 

Для удобства использования вынесем переменные из файла .env в отдельный файл настроек.

Точка входа app.py

middleware = [     Middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]),     Middleware(         AuthenticationMiddleware, backend=JWTAuthenticationBackend(secret_key=str(SECRET_KEY), algorithm=JWT_ALGORITHM, prefix=JWT_PREFIX)     ),  # str(SECRET_KEY) is important     Middleware(SessionMiddleware, secret_key=SECRET_KEY),     Middleware(CustomHeaderMiddleware), ]  routes = [     Mount("/user", routes=user_routes),     Mount("/", routes=main_routes), ]  entry_point = Starlette(debug=DEBUG, routes=routes, middleware=middleware)  tortoise_models = [     "apps.user.models", ]  register_tortoise(entry_point, db_url=DATABASE_URL, modules={"models": tortoise_models}, generate_schemas=True) 

Обратите внимание на порядок следования middleware, и на то что Tortoise ORM мы подключаем в самом конце.

Промежуточный слой для работы с JWT middleware.py

Поскольку Starlette еще достаточно молодой фреймворк, удобной «батарейки» JWT к нему еще не написано. Исправим этот недочет.

class JWTUser(BaseUser):     def __init__(self, username: str, user_id: int, email: str, token: str, **kw) -> None:         self.username = username         self.user_id = user_id         self.email = email         self.token = token      @property     def is_authenticated(self) -> bool:         return True      @property     def display_name(self) -> str:         return self.username      def __str__(self) -> str:         return f"JWT user: username={self.username}, id={self.user_id}, email={self.email}"   class JWTAuthenticationBackend(AuthenticationBackend):     def __init__(self, secret_key: str, algorithm: str = "HS256", prefix: str = "Bearer"):         self.secret_key = secret_key         self.algorithm = algorithm         self.prefix = prefix      @classmethod     def get_token_from_header(cls, authorization: str, prefix: str):          if DEBUG:             sprint_f(f"JWT token from headers: {authorization}", "cyan")  # debug part, do not forget to remove it         try:             scheme, token = authorization.split()         except ValueError:             if DEBUG:                 sprint_f(f"Could not separate Authorization scheme and token", "red")             raise AuthenticationError("Could not separate Authorization scheme and token")         if scheme.lower() != prefix.lower():             if DEBUG:                 sprint_f(f"Authorization scheme {scheme} is not supported", "red")             raise AuthenticationError(f"Authorization scheme {scheme} is not supported")         return token      async def authenticate(self, request):          if "Authorization" not in request.headers:             return None          authorization = request.headers["Authorization"]         token = self.get_token_from_header(authorization=authorization, prefix=self.prefix)          try:             jwt_payload = jwt.decode(token, key=str(self.secret_key), algorithms=self.algorithm)         except jwt.InvalidTokenError:             if DEBUG:                 sprint_f(f"Invalid JWT token", "red")             raise AuthenticationError("Invalid JWT token")         except jwt.ExpiredSignatureError:             if DEBUG:                 sprint_f(f"Expired JWT token", "red")             raise AuthenticationError("Expired JWT token")          if DEBUG:             sprint_f(f"Decoded JWT payload: {jwt_payload}", "green")  # debug part, do not forget to remove it          return (             AuthCredentials(["authenticated"]),             JWTUser(username=jwt_payload["username"], user_id=jwt_payload["user_id"], email=jwt_payload["email"], token=token),         ) 

Модель пользователя apps/user/models.py

Tortoise ORM замечательное решение для тех, кто хочет получить скорость asyncpg (https://github.com/MagicStack/asyncpg), и удобство классического Django ORM. Объявим модель пользователя.

from tortoise.models import Model from tortoise import fields  class User(Model):          id = fields.IntField(pk=True)     username = fields.CharField(max_length=255)          email = fields.CharField(max_length=255)       password = fields.CharField(max_length=255)         creation_date = fields.data.DatetimeField(auto_now_add=True)     last_login_date = fields.data.DatetimeField(null=True, blank=True)      def __str__(self):         return self.username     class Meta:         table = "user_user" 

Как мы видим, все очень просто и похоже на привычные нам модели Django.

Роутер apps/user/urls.py

<code> from starlette.routing import Route from .views import refresh_token from .views import user_login from .views import user_register  routes = [     Route("/register", endpoint=user_register, methods=["POST", "OPTIONS"], name="user__register"),     Route("/login", endpoint=user_login, methods=["POST", "OPTIONS"], name="user__login"),     Route("/refresh-token/", endpoint=refresh_token, methods=["POST", "OPTIONS"], name="user__refresh_token"), ] </code>

Роутер Starlette как мы видим также весьма прост и похож на привычный нам роутер Django.

Регистрация и логин apps/user/views.py

<code> from .models import User from settings import JWT_ALGORITHM from settings import JWT_PREFIX from settings import SECRET_KEY  async def create_token(token_config: dict) -> str:      exp = datetime.utcnow() + timedelta(minutes=token_config["expiration_minutes"])     token = {         "username": token_config["username"],         "user_id": token_config["user_id"],         "email": token_config["email"],         "iat": datetime.utcnow(),         "exp": exp,     }      if "get_expired_token" in token_config:         token["sub"] = "token"     else:         token["sub"] = "refresh_token"      token = jwt.encode(token, str(SECRET_KEY), algorithm=JWT_ALGORITHM)     return token.decode("UTF-8")   async def user_register(request: Request) -> JSONResponse:      try:         payload = await request.json()     except JSONDecodeError:         raise HTTPException(status_code=HTTP_400_BAD_REQUEST, detail="Can't parse json request")      username = payload["username"]     email = payload["email"]     password = pbkdf2_sha256.hash(payload["password"])      user_exist = await User.filter(email=email).first()     if user_exist:         raise HTTPException(status_code=HTTP_400_BAD_REQUEST, detail="Already registred")      new_user = User()     new_user.username = username     new_user.email = email     new_user.password = password     await new_user.save()      token = await create_token({"email": email, "username": username, "user_id": new_user.id, "get_expired_token": 1, "expiration_minutes": 30})     refresh_token = await create_token({"email": email, "username": username, "user_id": new_user.id, "get_refresh_token": 1, "expiration_minutes": 10080})      return JSONResponse({"id": new_user.id, "username": new_user.username, "email": new_user.email, "token": f"{JWT_PREFIX} {token}", "refresh_token": f"{JWT_PREFIX} {refresh_token}",}, status_code=200,)   async def user_login(request: Request) -> JSONResponse:      try:         payload = await request.json()     except JSONDecodeError:         raise HTTPException(status_code=HTTP_400_BAD_REQUEST, detail="Can't parse json request")      email = payload["email"]     password = payload["password"]      user = await User.filter(email=email).first()     if user:         if pbkdf2_sha256.verify(password, user.password):             user.last_login_date = datetime.now()             await user.save()              token = await create_token({"email": user.email, "username": user.username, "user_id": user.id, "get_expired_token": 1, "expiration_minutes": 30})             refresh_token = await create_token({"email": user.email, "username": user.username, "user_id": user.id, "get_refresh_token": 1, "expiration_minutes": 10080})              return JSONResponse({"id": user.id, "username": user.username, "email": user.email, "token": f"{JWT_PREFIX} {token}", "refresh_token": f"{JWT_PREFIX} {refresh_token}",}, status_code=200,)         else:             raise HTTPException(status_code=HTTP_400_BAD_REQUEST, detail=f"Invalid login or password")     else:         raise HTTPException(status_code=HTTP_400_BAD_REQUEST, detail=f"Invalid login or password") 

Несколько замечаний по коду. Во первых все ваши функции должны начинаться с ключевого слова async. Второй момент вызов функции внутри функции обязательно должен сопровождаться ключевым словом await. В остальном все тоже самое как и в привычной нам Django.

Ссылки

Полный код на Github:

Бекенд на Starlette

Фронтенд на Vue.js

Пример работы

Спасибо за внимание Удачных интеграций.