File size: 2,188 Bytes
d99646e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 | from sqlalchemy import func
from sqlalchemy.orm import Session
from app.core.security import create_access_token, hash_password, verify_password
from app.models.department import Department
from app.models.user import User
from app.schemas.auth import LoginRequest, SignupRequest, TokenResponse
def get_user_by_email(db: Session, email: str) -> User | None:
return db.query(User).filter(User.email == email.lower()).first()
def get_department_by_name(db: Session, department_name: str) -> Department | None:
normalized_name = department_name.strip().lower()
return (
db.query(Department)
.filter(func.lower(Department.name) == normalized_name)
.first()
)
def resolve_signup_target(
db: Session,
department_name: str,
) -> tuple[str, int | None]:
normalized_name = department_name.strip()
if normalized_name.lower() == "student":
return "STUDENT", None
department = get_department_by_name(db, normalized_name)
if department is None:
raise ValueError("Selected department was not found")
return department.name, department.id
def create_user(db: Session, payload: SignupRequest) -> User:
existing_user = get_user_by_email(db, payload.email)
if existing_user is not None:
raise ValueError("Email is already registered")
role, department_id = resolve_signup_target(db, payload.department_name)
user = User(
email=payload.email.lower(),
full_name=payload.full_name,
hashed_password=hash_password(payload.password),
role=role,
is_admin=False,
department_id=department_id,
)
db.add(user)
db.commit()
db.refresh(user)
return user
def authenticate_user(db: Session, payload: LoginRequest) -> User:
user = get_user_by_email(db, payload.email)
if user is None or not verify_password(payload.password, user.hashed_password):
raise ValueError("Incorrect email or password")
return user
def build_token_response(user: User) -> TokenResponse:
access_token = create_access_token(subject=str(user.id))
return TokenResponse(
access_token=access_token,
user=user,
)
|