48 lines
1.5 KiB
Python
48 lines
1.5 KiB
Python
import re
|
|
from fastapi import HTTPException, BackgroundTasks
|
|
from sqlalchemy import select
|
|
from typing import Optional
|
|
from pydantic import BaseModel, Field, model_validator
|
|
from background_jobs.send_email import send_email
|
|
from database.models import User
|
|
from database.transaction import get_session
|
|
from passlib.hash import bcrypt
|
|
from emails.emails import WelcomeEmail
|
|
|
|
|
|
class RegistrationInput(BaseModel):
|
|
email: str = Field(..., min_length=4)
|
|
name: Optional[str]
|
|
password: str
|
|
|
|
@model_validator(mode="before")
|
|
def validate_email(cls, values):
|
|
email = values.get("email") or ""
|
|
|
|
if email and not re.match(r"(.+)@(.+)\..{2,6}", email):
|
|
raise ValueError("Specified e-mail is not valid!")
|
|
|
|
return values
|
|
|
|
|
|
class RegistrationEndpoint:
|
|
@staticmethod
|
|
async def on_post(user_data: RegistrationInput, backgorund_tasks: BackgroundTasks) -> User:
|
|
query = select(User).filter_by(email=user_data.email)
|
|
async with get_session() as db:
|
|
existing_user = (await db.scalars(query)).first()
|
|
|
|
if existing_user:
|
|
raise HTTPException(status_code=422, detail="User already exists")
|
|
|
|
model = await User.create(db, {
|
|
"name": user_data.name,
|
|
"email": user_data.email,
|
|
"password_hashed": bcrypt.hash(user_data.password),
|
|
"description": ""
|
|
})
|
|
|
|
backgorund_tasks.add_task(send_email, WelcomeEmail(to=user_data.email))
|
|
|
|
return model.as_dict()
|