181 lines
4.7 KiB
Python
181 lines
4.7 KiB
Python
"""
|
|
User models and validation for the MyDocManager application.
|
|
|
|
Contains Pydantic models for user creation, updates, database representation,
|
|
and API responses with proper validation and type safety.
|
|
"""
|
|
|
|
import re
|
|
from datetime import datetime
|
|
from typing import Optional, Any
|
|
from bson import ObjectId
|
|
from pydantic import BaseModel, Field, field_validator, EmailStr
|
|
from pydantic_core import core_schema
|
|
|
|
from app.models.auth import UserRole
|
|
|
|
|
|
class PyObjectId(ObjectId):
|
|
"""Custom ObjectId type for Pydantic v2 compatibility."""
|
|
|
|
@classmethod
|
|
def __get_pydantic_core_schema__(
|
|
cls, source_type: Any, handler
|
|
) -> core_schema.CoreSchema:
|
|
return core_schema.json_or_python_schema(
|
|
json_schema=core_schema.str_schema(),
|
|
python_schema=core_schema.union_schema([
|
|
core_schema.is_instance_schema(ObjectId),
|
|
core_schema.chain_schema([
|
|
core_schema.str_schema(),
|
|
core_schema.no_info_plain_validator_function(cls.validate),
|
|
])
|
|
]),
|
|
serialization=core_schema.plain_serializer_function_ser_schema(
|
|
lambda x: str(x)
|
|
),
|
|
)
|
|
|
|
@classmethod
|
|
def validate(cls, v):
|
|
if not ObjectId.is_valid(v):
|
|
raise ValueError("Invalid ObjectId")
|
|
return ObjectId(v)
|
|
|
|
|
|
def validate_password_strength(password: str) -> str:
|
|
"""
|
|
Validate password meets security requirements.
|
|
|
|
Requirements:
|
|
- At least 8 characters long
|
|
- Contains at least one uppercase letter
|
|
- Contains at least one lowercase letter
|
|
- Contains at least one digit
|
|
- Contains at least one special character
|
|
|
|
Args:
|
|
password: The password string to validate
|
|
|
|
Returns:
|
|
str: The validated password
|
|
|
|
Raises:
|
|
ValueError: If password doesn't meet requirements
|
|
"""
|
|
if len(password) < 8:
|
|
raise ValueError("Password must be at least 8 characters long")
|
|
|
|
if not re.search(r'[A-Z]', password):
|
|
raise ValueError("Password must contain at least one uppercase letter")
|
|
|
|
if not re.search(r'[a-z]', password):
|
|
raise ValueError("Password must contain at least one lowercase letter")
|
|
|
|
if not re.search(r'\d', password):
|
|
raise ValueError("Password must contain at least one digit")
|
|
|
|
if not re.search(r'[!@#$%^&*()_+\-=\[\]{};:"\\|,.<>\/?]', password):
|
|
raise ValueError("Password must contain at least one special character")
|
|
|
|
return password
|
|
|
|
|
|
def validate_username_not_empty(username: str) -> str:
|
|
"""
|
|
Validate username is not empty or whitespace only.
|
|
|
|
Args:
|
|
username: The username string to validate
|
|
|
|
Returns:
|
|
str: The validated username
|
|
|
|
Raises:
|
|
ValueError: If username is empty or whitespace only
|
|
"""
|
|
if not username or not username.strip():
|
|
raise ValueError("Username cannot be empty or whitespace only")
|
|
|
|
return username.strip()
|
|
|
|
|
|
class UserCreate(BaseModel):
|
|
"""Model for creating a new user."""
|
|
|
|
username: str
|
|
email: EmailStr
|
|
password: str
|
|
role: UserRole = UserRole.USER
|
|
|
|
@field_validator('username')
|
|
@classmethod
|
|
def validate_username(cls, v):
|
|
return validate_username_not_empty(v)
|
|
|
|
@field_validator('password')
|
|
@classmethod
|
|
def validate_password(cls, v):
|
|
return validate_password_strength(v)
|
|
|
|
|
|
class UserUpdate(BaseModel):
|
|
"""Model for updating an existing user."""
|
|
|
|
username: Optional[str] = None
|
|
email: Optional[EmailStr] = None
|
|
password: Optional[str] = None
|
|
role: Optional[UserRole] = None
|
|
is_active: Optional[bool] = None
|
|
|
|
@field_validator('username')
|
|
@classmethod
|
|
def validate_username(cls, v):
|
|
if v is not None:
|
|
return validate_username_not_empty(v)
|
|
return v
|
|
|
|
@field_validator('password')
|
|
@classmethod
|
|
def validate_password(cls, v):
|
|
if v is not None:
|
|
return validate_password_strength(v)
|
|
return v
|
|
|
|
|
|
class UserInDB(BaseModel):
|
|
"""Model for user data stored in database."""
|
|
|
|
id: PyObjectId = Field(default_factory=PyObjectId, alias="_id")
|
|
username: str
|
|
email: str
|
|
hashed_password: str
|
|
role: UserRole
|
|
is_active: bool = True
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
|
|
model_config = {
|
|
"populate_by_name": True,
|
|
"arbitrary_types_allowed": True,
|
|
"json_encoders": {ObjectId: str}
|
|
}
|
|
|
|
|
|
class UserResponse(BaseModel):
|
|
"""Model for user data in API responses (excludes password_hash)."""
|
|
|
|
id: PyObjectId = Field(alias="_id")
|
|
username: str
|
|
email: str
|
|
role: UserRole
|
|
is_active: bool
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
|
|
model_config = {
|
|
"populate_by_name": True,
|
|
"arbitrary_types_allowed": True,
|
|
"json_encoders": {ObjectId: str}
|
|
}
|