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
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
|
import re
from typing import Optional
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from app.models.user import User
from app.schemas.user import UserCreate, UserRead
from app.utils.db import get_async_session
from app.utils.hash_cfg import hash_password
from app.utils.logger_cfg import logger
router = APIRouter(tags=["auth"])
@router.post("/register", response_model=UserRead)
async def register_user(
user: UserCreate, session: AsyncSession = Depends(get_async_session)
):
logger.debug("Register request received")
email: Optional[str] = user.email.strip() if user.email else None
logger.debug("Normalized email value: {}", email)
logger.info(
"Registration attempt | username={} email={}",
user.username,
email,
)
logger.debug("Validating password complexity")
if not (
re.search(r"[A-Za-z]", user.password)
and re.search(r"\d", user.password)
and re.search(r"[^\w\s]", user.password)
):
logger.warning(
"Registration failed | password complexity requirement not met | username={}",
user.username,
)
logger.debug("Checking if username already exists")
result = await session.execute(select(User).where(User.username == user.username))
existing_username = result.scalars().first()
if existing_username:
logger.warning(
"Registration failed | username already exists | username={}",
user.username,
)
raise HTTPException(
status_code=400,
detail={"field": "username", "message": "Никнейм уже занят."},
)
if email:
logger.debug("Checking if email already exists")
result = await session.execute(select(User).where(User.email == email))
existing_email = result.scalars().first()
if existing_email:
logger.warning(
"Registration failed | email already exists | email={}",
email,
)
raise HTTPException(
status_code=400,
detail={"field": "email", "message": "Адрес уже занят."},
)
logger.debug("Starting password hashing")
hashed_password = hash_password(user.password)
logger.debug("Password hashing completed")
logger.debug("Creating new user")
new_user = User(
username=user.username,
email=email,
password=hashed_password,
)
logger.debug("User model created | username={}", user.username)
logger.debug("Adding user to session")
session.add(new_user)
logger.debug("Preparing to commit database transaction")
await session.commit()
logger.debug("Transaction committed successfully")
logger.debug("Refreshing user instance from database")
await session.refresh(new_user)
logger.debug(
"User instance refreshed | id={} username={}",
new_user.id,
new_user.username,
)
logger.success(
"User successfully registered | id={} username={} email={}",
new_user.id,
new_user.username,
new_user.email,
)
return new_user
|
