👩🏼‍🎨 👛 🎱 Autorisation utilisateur à l'aide de Starlette + Vue.js ⚙️ 🚴🏾 🎅🏿

introduction

La tâche consiste à créer un exemple d'autorisation utilisateur à l'aide des frameworks Starlette ( https://www.starlette.io/ ) et Vue.js *, qui seraient les plus confortables pour les développeurs Django pour «migrer» vers la pile asynchrone.

Pourquoi Starlette? Tout d'abord, la vitesse. Starlette est ultra rapide, et dans les tests, elle est deuxième derrière BlackSheep ( https://pypi.org/project/blacksheep/ ). Deuxièmement, Starlette est très simple et facile à écrire dessus en raison de sa prévenance.

En tant qu'ORM, nous utiliserons Tortoise ORM (avec des modèles et des sélections «ala Django ORM»).

En tant que mécanisme de session, nous utiliserons JWT.

* La description du frontend sur Vue.js n'est pas incluse dans cette note.

Structure du projet

apps / user / models.py - modèle utilisateur
apps / user / urls.py - routeur
apps / user / views.py - enregistrement et connexion
.env - nos variables
settings.py - paramètres généraux du projet
app.py - point d'entrée du
middleware. py - middleware pour travailler avec JWT

Fichier .env variable

Ici, nous déclarons les variables dont nous aurons besoin à l'avenir pour fonctionner:

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

Paramètres généraux du projet 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)

Pour plus de commodité, nous transférerons les variables du fichier .env vers un fichier de paramètres séparé.

Point d'entrée 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)

Faites attention à l'ordre des middlewares et au fait que nous connectons Tortoise ORM à la toute fin.

JWT middleware.py middleware

Depuis Starlette est encore un cadre assez jeune, la "batterie" JWT pratique n'a pas encore été écrite pour cela. Corrigez ce défaut.

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),
        )

Modèle utilisateur apps / user / models.py

Tortoise ORM est une excellente solution pour ceux qui veulent obtenir une vitesse asyncpg (https://github.com/MagicStack/asyncpg) et la commodité du Django ORM classique. Déclarez le modèle utilisateur.

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"

Comme nous pouvons le voir, tout est très simple et similaire aux modèles Django habituels.

Applications de routeur / utilisateur / 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>

Le routeur Starlette, comme nous le voyons, est également très simple et similaire au routeur Django habituel.

Enregistrement et connexion des applications / utilisateur / 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")

Quelques commentaires sur le code. Tout d'abord, toutes vos fonctions doivent commencer par le mot-clé async. Le deuxième point, l'appel de fonction à l'intérieur de la fonction doit être accompagné du mot-clé wait. Sinon, tout est le même que dans le Django habituel.

Références

Code complet sur Github:

Beckend sur Starlette Frontend

sur Vue.js

Exemple de travail

Merci pour votre attention Intégrations réussies.

Autorisation utilisateur à l'aide de Starlette + Vue.js