Вступление
Задача — создать пример авторизации пользователя с использованием фреймворков 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
Пример работы
Спасибо за внимание Удачных интеграций.
