app/core/user.py - master - webprojectleonid/backend

app/core/user.py
65 строк2 KB
Leonid-poc
first commit
20 окт 2025, 08:38
20 окт 2025, 08:38f9afa07
from datetime import datetime, timedelta
from fastapi import Depends, HTTPException
from jose import jwt, JWTError
from passlib.context import CryptContext
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession
from dotenv import load_dotenv
import os

from app.db import get_db
from app.models import User

load_dotenv()

SESSION_SECRET = os.getenv("SESSION_SECRET")
ALGORITHM = os.getenv("ALGORITHM")
# Для хэширования паролей
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# OAuth2PasswordBearer указывает endpoint для входа
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/login")

def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password):
    return pwd_context.hash(password)

# Альтернативный способ через ORM (рекомендуемый):
async def get_user_by_email_orm(db: AsyncSession, email: str):
    result = await db.execute(
        # Используем SQLAlchemy ORM запрос
        User.__table__.select().where(User.email == email)
    )
    user_row = result.first()
    if user_row:
        return User(**user_row._mapping)
    return None

async def authenticate_user(db: AsyncSession, email: str, password: str):
    user = await get_user_by_email_orm(db, email)
    if not user or not verify_password(password, user.hashed_password):
        return False
    return user

def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SESSION_SECRET, algorithm=ALGORITHM)
    return encoded_jwt

async def get_current_user(token: str = Depends(oauth2_scheme), db: AsyncSession = Depends(get_db)) -> User:
    credentials_exception = HTTPException(status_code=401, detail="Не валидны данные", headers={"WWW-Authenticate": "Bearer"})
    try:
        payload = jwt.decode(token, SESSION_SECRET, algorithms=[ALGORITHM])
        email: str = payload.get("sub")
        if email is None:
            raise credentials_exception
        user = await get_user_by_email_orm(db, email)
    except JWTError:
        raise credentials_exception
    if user is None:
        raise credentials_exception
    return user