Introducci贸n
La tarea es crear un ejemplo de autorizaci贸n de usuario utilizando los marcos Starlette ( https://www.starlette.io/ ) y Vue.js *, que ser铆an los m谩s c贸modos para los desarrolladores de Django para "migrar" a la pila asincr贸nica.驴Por qu茅 Starlette? En primer lugar, la velocidad. Starlette es un ultim谩tum r谩pido, y solo superado por BlackSheep en las pruebas ( https://pypi.org/project/blacksheep/ ). En segundo lugar, Starlette es muy simple y f谩cil de escribir debido a su consideraci贸n.Como ORM usaremos Tortoise ORM (con modelos y selecciones "ala Django ORM").Como mecanismo de sesi贸n, utilizaremos JWT.* La descripci贸n de la interfaz en Vue.js no est谩 incluida en esta nota.Estructura del proyecto
apps / user / models.py - modelo de usuarioapps / user / urls.py - enrutadorapps / user / views.py - registro e inicio de sesi贸n.env - nuestras variablessettings.py - configuraci贸n general del proyectoapp.py - punto de entrada demiddleware. py - middleware para trabajar con JWTArchivo .env variable
Aqu铆 declaramos las variables que necesitaremos en el futuro para funcionar: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
Configuraci贸n general project 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)
Por conveniencia, transferiremos las variables del archivo .env a un archivo de configuraci贸n separado.Punto de entrada 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)
),
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)
Preste atenci贸n al orden del middleware y al hecho de que conectamos Tortoise ORM al final.JWT middleware.py middleware
Dado que Starlette sigue siendo un marco bastante joven, a煤n no se ha escrito la conveniente "bater铆a" de JWT. Corrija este defecto.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")
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")
return (
AuthCredentials(["authenticated"]),
JWTUser(username=jwt_payload["username"], user_id=jwt_payload["user_id"], email=jwt_payload["email"], token=token),
)
Modelo de usuario apps / user / models.py
Tortoise ORM es una gran soluci贸n para aquellos que desean obtener la velocidad de asyncpg (https://github.com/MagicStack/asyncpg) y la conveniencia del cl谩sico Django ORM. Declarar el modelo de usuario.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"
Como podemos ver, todo es muy simple y similar a los modelos habituales de Django.Router 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>
El enrutador Starlette, como lo vemos, tambi茅n es muy simple y similar al enrutador Django habitual.Registro y aplicaciones de inicio de sesi贸n / 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")
Algunos comentarios sobre el c贸digo. Primero, todas sus funciones deben comenzar con la palabra clave as铆ncrona. El segundo punto, la llamada a la funci贸n dentro de la funci贸n debe ir acompa帽ada de la palabra clave wait. De lo contrario, todo es igual que en el Django habitual.Referencias
C贸digo completo en Github:Beckend en Starlette Frontenden Vue.jsEjemplo de trabajoGracias por su atenci贸n Integraciones exitosas.