140 lines
4.6 KiB
Python
140 lines
4.6 KiB
Python
from fastapi import APIRouter, Depends, Form, HTTPException, Request, status
|
||
from fastapi.responses import HTMLResponse, RedirectResponse
|
||
from fastapi.templating import Jinja2Templates
|
||
from sqlalchemy.orm import Session as DbSession
|
||
|
||
from app.auth import (
|
||
SESSION_COOKIE,
|
||
create_user_session,
|
||
get_current_user,
|
||
get_current_user_optional,
|
||
hash_password,
|
||
verify_password,
|
||
)
|
||
from app.database import get_db
|
||
from app.models import Session, User
|
||
|
||
router = APIRouter(tags=["web"])
|
||
templates = Jinja2Templates(directory="templates")
|
||
|
||
|
||
@router.get("/", response_class=HTMLResponse)
|
||
def index(request: Request, user: User | None = Depends(get_current_user_optional)):
|
||
if user:
|
||
return RedirectResponse("/cabinet", status_code=status.HTTP_302_FOUND)
|
||
return RedirectResponse("/login", status_code=status.HTTP_302_FOUND)
|
||
|
||
|
||
@router.get("/register", response_class=HTMLResponse)
|
||
def register_page(request: Request, user: User | None = Depends(get_current_user_optional)):
|
||
if user:
|
||
return RedirectResponse("/cabinet", status_code=status.HTTP_302_FOUND)
|
||
return templates.TemplateResponse(request, "register.html", {"error": None})
|
||
|
||
|
||
@router.post("/register")
|
||
def register(
|
||
request: Request,
|
||
email: str = Form(...),
|
||
password: str = Form(...),
|
||
db: DbSession = Depends(get_db),
|
||
):
|
||
email = email.strip().lower()
|
||
if len(password) < 6:
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"register.html",
|
||
{"error": "Пароль должен быть не менее 6 символов"},
|
||
status_code=400,
|
||
)
|
||
existing = db.query(User).filter(User.email == email).first()
|
||
if existing:
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"register.html",
|
||
{"error": "Пользователь с такой почтой уже существует"},
|
||
status_code=400,
|
||
)
|
||
user = User(email=email, password_hash=hash_password(password))
|
||
db.add(user)
|
||
db.commit()
|
||
db.refresh(user)
|
||
session = create_user_session(db, user)
|
||
response = RedirectResponse("/cabinet", status_code=status.HTTP_302_FOUND)
|
||
response.set_cookie(
|
||
key=SESSION_COOKIE,
|
||
value=session.token,
|
||
httponly=True,
|
||
samesite="lax",
|
||
max_age=86400,
|
||
)
|
||
return response
|
||
|
||
|
||
@router.get("/login", response_class=HTMLResponse)
|
||
def login_page(request: Request, user: User | None = Depends(get_current_user_optional)):
|
||
if user:
|
||
return RedirectResponse("/cabinet", status_code=status.HTTP_302_FOUND)
|
||
return templates.TemplateResponse(request, "login.html", {"error": None})
|
||
|
||
|
||
@router.post("/login")
|
||
def login(
|
||
request: Request,
|
||
email: str = Form(...),
|
||
password: str = Form(...),
|
||
db: DbSession = Depends(get_db),
|
||
):
|
||
email = email.strip().lower()
|
||
user = db.query(User).filter(User.email == email).first()
|
||
if not user or not verify_password(password, user.password_hash):
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"login.html",
|
||
{"error": "Неверная почта или пароль"},
|
||
status_code=400,
|
||
)
|
||
session = create_user_session(db, user)
|
||
response = RedirectResponse("/cabinet", status_code=status.HTTP_302_FOUND)
|
||
response.set_cookie(
|
||
key=SESSION_COOKIE,
|
||
value=session.token,
|
||
httponly=True,
|
||
samesite="lax",
|
||
max_age=86400,
|
||
)
|
||
return response
|
||
|
||
|
||
@router.post("/logout")
|
||
def logout(request: Request, db: DbSession = Depends(get_db)):
|
||
token = request.cookies.get(SESSION_COOKIE)
|
||
if token:
|
||
session = db.query(Session).filter(Session.token == token).first()
|
||
if session:
|
||
session.is_active = False
|
||
db.commit()
|
||
response = RedirectResponse("/login", status_code=status.HTTP_302_FOUND)
|
||
response.delete_cookie(SESSION_COOKIE)
|
||
return response
|
||
|
||
|
||
@router.get("/cabinet", response_class=HTMLResponse)
|
||
def cabinet(request: Request, user: User = Depends(get_current_user), db: DbSession = Depends(get_db)):
|
||
sessions = (
|
||
db.query(Session)
|
||
.filter(Session.user_id == user.id, Session.is_active.is_(True))
|
||
.order_by(Session.created_at.desc())
|
||
.all()
|
||
)
|
||
external_data = list(reversed(user.external_data[-10:])) if user.external_data else []
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"cabinet.html",
|
||
{
|
||
"user": user,
|
||
"sessions": sessions,
|
||
"external_data": external_data,
|
||
},
|
||
)
|