50 lines
1.3 KiB
Python
50 lines
1.3 KiB
Python
from sqlalchemy import or_, select
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
from app.extensions import db
|
|
|
|
from .models import User
|
|
|
|
|
|
class DuplicateUserError(ValueError):
|
|
pass
|
|
|
|
|
|
def create_user(username: str, email: str, password: str) -> User:
|
|
normalized_username = username.strip()
|
|
normalized_email = email.strip().lower()
|
|
|
|
existing_user = db.session.scalar(
|
|
select(User).where(
|
|
or_(
|
|
User.username == normalized_username,
|
|
User.email == normalized_email,
|
|
)
|
|
)
|
|
)
|
|
if existing_user is not None:
|
|
raise DuplicateUserError("That username or email is already registered.")
|
|
|
|
user = User(username=normalized_username, email=normalized_email)
|
|
user.set_password(password)
|
|
db.session.add(user)
|
|
|
|
try:
|
|
db.session.commit()
|
|
except IntegrityError as exc:
|
|
db.session.rollback()
|
|
raise DuplicateUserError(
|
|
"That username or email is already registered."
|
|
) from exc
|
|
|
|
return user
|
|
|
|
|
|
def authenticate_user(email: str, password: str) -> User | None:
|
|
user = db.session.scalar(
|
|
select(User).where(User.email == email.strip().lower())
|
|
)
|
|
if user is None or not user.check_password(password):
|
|
return None
|
|
return user
|
|
|