🏉 ⭐️ 🧑🏾 Benutzerautorisierung mit Starlette + Vue.js 😟 🗳️ 🤘🏿

Einführung

Die Aufgabe besteht darin, ein Beispiel für die Benutzerautorisierung mithilfe der Frameworks Starlette ( https://www.starlette.io/ ) und Vue.js * zu erstellen , die für Django-Entwickler am bequemsten sind, um auf den asynchronen Stapel zu „migrieren“.

Warum Starlette? Vor allem Geschwindigkeit. Starlette ist ultimativ schnell und in Tests nach BlackSheep an zweiter Stelle ( https://pypi.org/project/blacksheep/ ). Zweitens ist Starlette aufgrund seiner Nachdenklichkeit sehr einfach und leicht zu schreiben.

Als ORM verwenden wir Tortoise ORM (mit Modellen und Auswahlmöglichkeiten „ala Django ORM“).

Als Sitzungsmechanismus verwenden wir JWT.

* Die Beschreibung des Frontends auf Vue.js ist in diesem Hinweis nicht enthalten.

Projektstruktur

apps / user / models.py - Benutzermodell
apps / user / urls.py - Router-
Apps / user / views.py - Registrierung und Anmeldung
.env - unsere Variablen
settings.py - allgemeine Projekteinstellungen
app.py - Einstiegspunkt in die
Middleware. py - Middleware für die Arbeit mit JWT

Variable .env-Datei

Hier deklarieren wir die Variablen, die wir in Zukunft benötigen, um zu arbeiten:

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

Allgemeine Einstellungen Projekteinstellungen.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)

Der Einfachheit halber übertragen wir die Variablen aus der ENV-Datei in eine separate Einstellungsdatei.

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

Achten Sie auf die Reihenfolge der Middleware und die Tatsache, dass wir Tortoise ORM ganz am Ende verbinden.

JWT Middleware.py Middleware

Da Starlette noch ein relativ junges Framework ist, wurde die praktische JWT- "Batterie" noch nicht dafür geschrieben. Korrigieren Sie diesen Fehler.

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

Benutzermodell apps / user / models.py

Tortoise ORM ist eine großartige Lösung für diejenigen, die Asyncpg-Geschwindigkeit (https://github.com/MagicStack/asyncpg) und den Komfort des klassischen Django ORM erhalten möchten. Deklarieren Sie das Benutzermodell.

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"

Wie wir sehen können, ist alles sehr einfach und den üblichen Django-Modellen ähnlich.

Router Apps / Benutzer / 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>

Der Starlette-Router ist aus unserer Sicht auch sehr einfach und dem üblichen Django-Router ähnlich.

Registrierungs- und Login-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")

Ein paar Kommentare zum Code. Zunächst sollten alle Ihre Funktionen mit dem Schlüsselwort async beginnen. Der zweite Punkt, der Funktionsaufruf innerhalb der Funktion, muss vom Schlüsselwort await begleitet werden. Ansonsten ist alles wie beim üblichen Django.

Verweise

Vollständiger Code auf Github:

Beckend auf Starlette Frontend

auf Vue.js

Beispielarbeit

Vielen Dank für Ihre Aufmerksamkeit Erfolgreiche Integration.

Benutzerautorisierung mit Starlette + Vue.js