41 lines
1.1 KiB
Python
41 lines
1.1 KiB
Python
from sqlalchemy.orm import Session
|
|
|
|
from app.core.security import hash_password, verify_password
|
|
from app.models.user import User
|
|
from app.schemas.auth import RegisterRequest
|
|
|
|
|
|
class EmailAlreadyRegisteredError(Exception):
|
|
pass
|
|
|
|
|
|
def get_user_by_email(db: Session, email: str) -> User | None:
|
|
return db.query(User).filter(User.email == email).first()
|
|
|
|
|
|
def get_user_by_id(db: Session, user_id: int) -> User | None:
|
|
return db.query(User).filter(User.id == user_id).first()
|
|
|
|
|
|
def register_user(db: Session, body: RegisterRequest) -> User:
|
|
if get_user_by_email(db, body.email):
|
|
raise EmailAlreadyRegisteredError()
|
|
user = User(
|
|
email=body.email,
|
|
hashed_password=hash_password(body.password),
|
|
full_name=body.full_name,
|
|
is_active=True,
|
|
)
|
|
db.add(user)
|
|
db.commit()
|
|
db.refresh(user)
|
|
return user
|
|
|
|
|
|
def authenticate_user(db: Session, email: str, password: str) -> User | None:
|
|
user = get_user_by_email(db, email)
|
|
if not user or not user.is_active:
|
|
return None
|
|
if not verify_password(password, user.hashed_password):
|
|
return None
|
|
return user
|