all ruff checking and modify Converter class

This commit is contained in:
2024-06-21 12:30:27 +07:00
parent cd3bae23e6
commit e1034fb927
19 changed files with 1067 additions and 995 deletions
+9 -4
View File
@@ -1,10 +1,15 @@
from fastapi import Request, Depends, responses, HTTPException
from fastapi import Depends, HTTPException, Request, responses
from functions.admin import get_log, is_logged_in
async def log(templates, request: Request, log_id: int, logged_in: bool = Depends(is_logged_in.is_logged_in)):
async def log(templates, request: Request, log_id: int,
logged_in: bool = Depends(is_logged_in.is_logged_in)):
if logged_in:
if not get_log.get_log(log_id):
raise HTTPException(status_code=400)
return templates.TemplateResponse("log.html", {"request": request, "log": get_log.get_log(log_id)})
return templates.TemplateResponse("log.html", {"request": request,
"log": get_log.get_log(log_id)})
else:
return responses.RedirectResponse(url=f"/refresh?source=/logs/{log_id}", status_code=303)
return responses.RedirectResponse(url=f"/refresh?source=/logs/{log_id}",
status_code=303)
+13 -7
View File
@@ -1,25 +1,31 @@
from fastapi import Form, Depends, responses, HTTPException
from functions.admin.models import database, user, token
import jwt
from datetime import datetime, timedelta
import jwt
from fastapi import Depends, Form, HTTPException, responses
async def logins(username: str = Form(...), password: str = Form(...), db=Depends(database.get_db)):
from functions.admin.models import database, token, user
async def logins(username: str = Form(...), password: str = Form(...),
db=Depends(database.get_db)):
# username = "Dmitrium12"
response = db.query(user.User).filter(user.User.username == username).first()
if not response or not response.check_password(password):
raise HTTPException(status_code=400, detail="Неправильное имя пользователя или пароль")
raise HTTPException(status_code=400,
detail="Неправильное имя пользователя или пароль")
access_token_expires = datetime.utcnow() + timedelta(minutes=15)
access_token_payload = {"sub": response.username, "exp": access_token_expires}
access_token = jwt.encode(access_token_payload, "secret", algorithm="HS256")
refresh_token_expires = datetime.utcnow() + timedelta(days=7)
refresh_token_payload = {"sub": response.username, "exp": refresh_token_expires}
refresh_token = jwt.encode(refresh_token_payload, "secret", algorithm="HS256")
db_token = token.Token(access_token=access_token, refresh_token=refresh_token, user=response)
db_token = token.Token(access_token=access_token, refresh_token=refresh_token,
user=response)
db.add(db_token)
db.commit()
response = responses.RedirectResponse(url="/logs", status_code=303)
response.set_cookie(key="access_token", value=access_token, expires=int(access_token_expires.timestamp()))
response.set_cookie(key="access_token", value=access_token,
expires=int(access_token_expires.timestamp()))
response.set_cookie(key="refresh_token", value=refresh_token,
expires=int(refresh_token_expires.timestamp()))
return response
+7 -3
View File
@@ -1,8 +1,12 @@
from fastapi import Request, Depends, responses
from fastapi import Depends, Request, responses
from functions.admin import get_logs, is_logged_in
async def logs(templates, request: Request, logged_in: bool = Depends(is_logged_in.is_logged_in)):
async def logs(templates, request: Request,
logged_in: bool = Depends(is_logged_in.is_logged_in)):
if logged_in:
return templates.TemplateResponse("logs.html", {"request": request, "logs": get_logs.get_logs()})
return templates.TemplateResponse("logs.html",
{"request": request, "logs": get_logs.get_logs()})
else:
return responses.RedirectResponse(url="/refresh?source=/logs", status_code=303)
+10 -5
View File
@@ -1,15 +1,20 @@
from fastapi import Request, Depends, Cookie, responses
from functions.admin.models import database, user, token
import jwt
from datetime import datetime, timedelta
async def refresh_access_token(req: Request, refresh_token: str = Cookie(None), db=Depends(database.get_db)):
import jwt
from fastapi import Cookie, Depends, Request, responses
from functions.admin.models import database, token, user
async def refresh_access_token(req: Request, refresh_token: str = Cookie(None),
db=Depends(database.get_db)):
request_args = dict(req.query_params)
try:
refresh_token_payload = jwt.decode(refresh_token, "secret", algorithms=["HS256"])
except jwt.exceptions.DecodeError:
return responses.RedirectResponse(url="/login", status_code=303)
response = db.query(user.User).filter(user.User.username == refresh_token_payload["sub"]).first()
response = db.query(user.User).filter(
user.User.username == refresh_token_payload["sub"]).first()
if not response:
return responses.RedirectResponse(url="/login", status_code=303)
access_token_expires = datetime.utcnow() + timedelta(minutes=15)