Enhance user management, improve API structure, add database optimizations, and update Docker setup

- Introduced endpoints for user management, including CRUD operations, pagination, and password management.
- Added new schema validations for user updates, password strength, pagination, and standardized error responses.
- Integrated custom exception handling for a consistent API error experience.
- Refined CORS settings: restricted methods and allowed headers, added header exposure, and preflight caching.
- Optimized database: added indexes on `is_active` and `is_superuser` fields, updated column types, enforced constraints, and set defaults.
- Updated `Dockerfile` to improve security by using a non-root user and adding a health check for the application.
- Enhanced tests for database initialization, user operations, and exception handling to ensure better coverage.
This commit is contained in:
Felipe Cardoso
2025-10-30 15:43:52 +01:00
parent d83959963b
commit 2c600290a1
16 changed files with 1511 additions and 100 deletions

View File

@@ -0,0 +1,139 @@
"""
Common schemas used across the API for pagination, responses, etc.
"""
from typing import Generic, TypeVar, List, Optional
from pydantic import BaseModel, Field
from math import ceil
T = TypeVar('T')
class PaginationParams(BaseModel):
"""Parameters for pagination."""
page: int = Field(
default=1,
ge=1,
description="Page number (1-indexed)"
)
limit: int = Field(
default=20,
ge=1,
le=100,
description="Number of items per page (max 100)"
)
@property
def offset(self) -> int:
"""Calculate the offset for database queries."""
return (self.page - 1) * self.limit
@property
def skip(self) -> int:
"""Alias for offset (compatibility with existing code)."""
return self.offset
model_config = {
"json_schema_extra": {
"example": {
"page": 1,
"limit": 20
}
}
}
class PaginationMeta(BaseModel):
"""Metadata for paginated responses."""
total: int = Field(..., description="Total number of items")
page: int = Field(..., description="Current page number")
page_size: int = Field(..., description="Number of items in current page")
total_pages: int = Field(..., description="Total number of pages")
has_next: bool = Field(..., description="Whether there is a next page")
has_prev: bool = Field(..., description="Whether there is a previous page")
model_config = {
"json_schema_extra": {
"example": {
"total": 150,
"page": 1,
"page_size": 20,
"total_pages": 8,
"has_next": True,
"has_prev": False
}
}
}
class PaginatedResponse(BaseModel, Generic[T]):
"""Generic paginated response wrapper."""
data: List[T] = Field(..., description="List of items")
pagination: PaginationMeta = Field(..., description="Pagination metadata")
model_config = {
"json_schema_extra": {
"example": {
"data": [
{"id": "123", "name": "Example Item"}
],
"pagination": {
"total": 150,
"page": 1,
"page_size": 20,
"total_pages": 8,
"has_next": True,
"has_prev": False
}
}
}
}
class MessageResponse(BaseModel):
"""Simple message response."""
success: bool = Field(default=True, description="Operation success status")
message: str = Field(..., description="Human-readable message")
model_config = {
"json_schema_extra": {
"example": {
"success": True,
"message": "Operation completed successfully"
}
}
}
def create_pagination_meta(
total: int,
page: int,
limit: int,
items_count: int
) -> PaginationMeta:
"""
Helper function to create pagination metadata.
Args:
total: Total number of items
page: Current page number
limit: Items per page
items_count: Number of items in current page
Returns:
PaginationMeta object with calculated values
"""
total_pages = ceil(total / limit) if limit > 0 else 0
return PaginationMeta(
total=total,
page=page,
page_size=items_count,
total_pages=total_pages,
has_next=page < total_pages,
has_prev=page > 1
)

View File

@@ -0,0 +1,85 @@
"""
Error schemas for standardized API error responses.
"""
from enum import Enum
from typing import List, Optional
from pydantic import BaseModel, Field
class ErrorCode(str, Enum):
"""Standard error codes for the API."""
# Authentication errors (AUTH_xxx)
INVALID_CREDENTIALS = "AUTH_001"
TOKEN_EXPIRED = "AUTH_002"
TOKEN_INVALID = "AUTH_003"
INSUFFICIENT_PERMISSIONS = "AUTH_004"
USER_INACTIVE = "AUTH_005"
AUTHENTICATION_REQUIRED = "AUTH_006"
# User errors (USER_xxx)
USER_NOT_FOUND = "USER_001"
USER_ALREADY_EXISTS = "USER_002"
USER_CREATION_FAILED = "USER_003"
USER_UPDATE_FAILED = "USER_004"
USER_DELETION_FAILED = "USER_005"
# Validation errors (VAL_xxx)
VALIDATION_ERROR = "VAL_001"
INVALID_PASSWORD = "VAL_002"
INVALID_EMAIL = "VAL_003"
INVALID_PHONE_NUMBER = "VAL_004"
INVALID_UUID = "VAL_005"
INVALID_INPUT = "VAL_006"
# Database errors (DB_xxx)
DATABASE_ERROR = "DB_001"
DUPLICATE_ENTRY = "DB_002"
FOREIGN_KEY_VIOLATION = "DB_003"
RECORD_NOT_FOUND = "DB_004"
# Generic errors (SYS_xxx)
INTERNAL_ERROR = "SYS_001"
NOT_FOUND = "SYS_002"
METHOD_NOT_ALLOWED = "SYS_003"
RATE_LIMIT_EXCEEDED = "SYS_004"
class ErrorDetail(BaseModel):
"""Detailed information about a single error."""
code: ErrorCode = Field(..., description="Machine-readable error code")
message: str = Field(..., description="Human-readable error message")
field: Optional[str] = Field(None, description="Field name if error is field-specific")
model_config = {
"json_schema_extra": {
"example": {
"code": "VAL_002",
"message": "Password must be at least 8 characters long",
"field": "password"
}
}
}
class ErrorResponse(BaseModel):
"""Standardized error response format."""
success: bool = Field(default=False, description="Always false for error responses")
errors: List[ErrorDetail] = Field(..., description="List of errors that occurred")
model_config = {
"json_schema_extra": {
"example": {
"success": False,
"errors": [
{
"code": "AUTH_001",
"message": "Invalid email or password",
"field": None
}
]
}
}
}

View File

@@ -123,7 +123,26 @@ class TokenData(BaseModel):
is_superuser: bool = False
class PasswordChange(BaseModel):
"""Schema for changing password (requires current password)."""
current_password: str
new_password: str
@field_validator('new_password')
@classmethod
def password_strength(cls, v: str) -> str:
"""Basic password strength validation"""
if len(v) < 8:
raise ValueError('Password must be at least 8 characters')
if not any(char.isdigit() for char in v):
raise ValueError('Password must contain at least one digit')
if not any(char.isupper() for char in v):
raise ValueError('Password must contain at least one uppercase letter')
return v
class PasswordReset(BaseModel):
"""Schema for resetting password (via email token)."""
token: str
new_password: str