Get best of the week

使用Starlette + Vue.js的用户授权

介绍



任务是使用Starlette（https://www.starlette.io/）和Vue.js *框架创建用户授权的示例，这对于Django开发人员来说最适合“迁移”到异步堆栈。

为什么选择星彩？首先，速度。Starlette的最后通fast速度很快，并且在测试中仅次于BlackSheep（https://pypi.org/project/blacksheep/）。其次，由于Starlette的周到性，它非常简单易写。

作为ORM，我们将使用Tortoise ORM（带有模型和选项“ ala Django ORM”）。

作为会话机制，我们将使用JWT。

*本说明中不包含有关Vue.js前端的描述。

项目结构


apps / user / models.py-用户模型
apps / user / urls.py-路由器
apps / user / views.py-注册和登录
.env-我们的变量
settings.py-常规项目设置
app.py-
中间件入口点。 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)

注意中间件的顺序，以及最后连接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),
        )

用户模型应用程序/用户/models.py


对于想要获得asyncpg速度（https://github.com/MagicStack/asyncpg）和经典Django ORM便利性的用户来说，Tortoise 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模型相似。

路由器应用程序/用户/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路由器。

注册和登录应用程序/用户/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

上的完整代码：Vue.js上的Starlette Frontend

上的Beckend。

示例工作

感谢您的关注成功的集成。

More articles:


All Articles