pengantar
Tugasnya adalah membuat contoh otorisasi pengguna menggunakan Starlette ( https://www.starlette.io/ ) dan kerangka kerja Vue.js *, yang akan menjadi yang paling nyaman bagi pengembang Django untuk "bermigrasi" ke tumpukan asinkron.Kenapa Starlette? Pertama-tama, kecepatan. Starlette adalah ultimatum yang cepat, dan yang kedua setelah BlackSheep dalam pengujian ( https://pypi.org/project/blacksheep/ ). Kedua, Starlette sangat sederhana dan mudah untuk menulisnya karena perhatiannya.Sebagai ORM kita akan menggunakan Tortoise ORM (dengan model dan pilihan "ala Django ORM").Sebagai mekanisme sesi, kita akan menggunakan JWT.* Deskripsi frontend pada Vue.js tidak termasuk dalam catatan ini.Struktur proyek
apps / user / models.py - modelaplikasi pengguna / user / urls.py -aplikasi router / user / views.py - pendaftaran dan login.env -pengaturan variabelkami.py - pengaturan proyek umum app.py - titik masukmiddleware. py - middleware untuk bekerja dengan JWTFile .env variabel
Di sini kita mendeklarasikan variabel yang kita perlukan di masa depan untuk bekerja: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
Pengaturan umum pengaturan proyek.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)
Untuk kenyamanan, kami akan mentransfer variabel dari file .env ke file pengaturan terpisah.Titik masuk 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)
Perhatikan urutan middleware, dan fakta bahwa kami menghubungkan Tortoise ORM di bagian paling akhir.JWT middleware.py middleware
Karena Starlette masih merupakan kerangka kerja yang cukup muda, "baterai" JWT yang nyaman belum ditulis untuk itu. Perbaiki cacat ini.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),
)
Aplikasi model pengguna / user / models.py
Tortoise ORM adalah solusi hebat bagi mereka yang ingin mendapatkan kecepatan asyncpg (https://github.com/MagicStack/asyncpg) dan kenyamanan ORM Django klasik. Nyatakan model pengguna.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"
Seperti yang bisa kita lihat, semuanya sangat sederhana dan mirip dengan model Django yang biasa.Aplikasi router / pengguna / 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>
Router Starlette, seperti yang kita lihat, juga sangat sederhana dan mirip dengan router Django yang biasa.Pendaftaran dan masuk aplikasi / pengguna / 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")
Beberapa komentar pada kode. Pertama, semua fungsi Anda harus dimulai dengan kata kunci async. Poin kedua, pemanggilan fungsi di dalam fungsi harus disertai dengan kata kunci yang menunggu. Kalau tidak, semuanya sama seperti di Django biasa.Referensi
Kode lengkap pada Github:Beckend pada Starlette Frontendpada Vue.jsContoh kerjaTerima kasih atas perhatian Anda Integrasi yang berhasil.