Spaces:
Sleeping
Sleeping
| """ | |
| System User schemas for request/response models. | |
| """ | |
| from datetime import datetime | |
| from typing import Optional, List, Dict, Any | |
| from uuid import UUID | |
| from pydantic import BaseModel, Field, EmailStr, validator | |
| from app.system_users.models.model import UserStatus, UserRole | |
| class LoginRequest(BaseModel): | |
| """Login request schema.""" | |
| email_or_phone: str = Field(..., description="Email address or phone number", min_length=3, max_length=100) | |
| password: str = Field(..., description="User password", min_length=6, max_length=100) | |
| remember_me: bool = Field(default=False, description="Keep session active for longer period") | |
| class LoginResponse(BaseModel): | |
| """Login response schema.""" | |
| access_token: str = Field(..., description="JWT access token") | |
| token_type: str = Field(default="bearer", description="Token type") | |
| expires_in: int = Field(..., description="Token expiry time in seconds") | |
| user_info: "UserInfoResponse" = Field(..., description="Basic user information") | |
| class UserInfoResponse(BaseModel): | |
| """User information response schema.""" | |
| user_id: str = Field(..., description="Unique user identifier") | |
| username: str = Field(..., description="Username") | |
| email: str = Field(..., description="Email address") | |
| merchant_id: str = Field(..., description="Merchant ID") | |
| merchant_type: Optional[str] = Field(None, description="Merchant type") | |
| full_name: Optional[str] = Field(None, description="Full name") | |
| role: str = Field(..., description="User role") | |
| phone: Optional[str] = Field(None, description="Phone number") | |
| permissions: Dict[str, List[str]] = Field(default_factory=dict, description="User permissions") | |
| status: str = Field(..., description="Account status") | |
| last_login_at: Optional[datetime] = Field(None, description="Last login timestamp") | |
| profile_picture_url: Optional[str] = Field(None, description="Profile picture URL") | |
| staff_id: Optional[str] = Field(None, description="Staff ID (for POS staff users)") | |
| class CreateUserRequest(BaseModel): | |
| """Create user request schema.""" | |
| username: str = Field(..., description="Unique username", min_length=3, max_length=30) | |
| email: EmailStr = Field(..., description="Email address") | |
| merchant_id: str = Field(..., description="Merchant ID for the user") | |
| merchant_type: Optional[str] = Field(None, description="Merchant type (ncnf, cnf, distributor, retail)") | |
| password: str = Field(..., description="Password", min_length=6, max_length=100) | |
| first_name: str = Field(..., description="First name", min_length=1, max_length=50) | |
| last_name: Optional[str] = Field(None, description="Last name", max_length=50) | |
| full_name: Optional[str] = Field(None, description="Full name (auto-generated if not provided)") | |
| phone: Optional[str] = Field(None, description="Phone number") | |
| role: str = Field(default="user", alias="role_id", description="User role") | |
| permissions: Dict[str, List[str]] = Field(default_factory=dict, description="Additional permissions") | |
| status: UserStatus = Field(default=UserStatus.ACTIVE, description="Initial user status") | |
| metadata: Optional[Dict[str, Any]] = Field(None, description="Additional metadata (employee_user_id, employee_code, etc.)") | |
| class Config: | |
| populate_by_name = True | |
| def validate_username(cls, v): | |
| if not v.replace('_', '').replace('.', '').isalnum(): | |
| raise ValueError('Username can only contain alphanumeric characters, underscores, and periods') | |
| return v.lower() | |
| class UpdateUserRequest(BaseModel): | |
| """Update user request schema.""" | |
| first_name: Optional[str] = Field(None, description="First name", min_length=1, max_length=50) | |
| last_name: Optional[str] = Field(None, description="Last name", max_length=50) | |
| phone: Optional[str] = Field(None, description="Phone number") | |
| role: Optional[str] = Field(None, description="User role") | |
| merchant_type: Optional[str] = Field(None, description="Merchant type") | |
| permissions: Optional[Dict[str, List[str]]] = Field(None, description="User permissions") | |
| status: Optional[UserStatus] = Field(None, description="Account status") | |
| timezone: Optional[str] = Field(None, description="User timezone") | |
| language: Optional[str] = Field(None, description="Preferred language") | |
| class ChangePasswordRequest(BaseModel): | |
| """Change password request schema.""" | |
| current_password: str = Field(..., description="Current password") | |
| new_password: str = Field(..., description="New password", min_length=6, max_length=100) | |
| class ForgotPasswordRequest(BaseModel): | |
| """Forgot password request schema.""" | |
| email: EmailStr = Field(..., description="Email address to receive reset link") | |
| class VerifyForgotPasswordOTPRequest(BaseModel): | |
| """Verify OTP for forgot-password flow.""" | |
| email: EmailStr = Field(..., description="Email address") | |
| otp: str = Field(..., description="6-digit OTP received via WhatsApp/SMS", min_length=6, max_length=6) | |
| class ResetPasswordWithOTPRequest(BaseModel): | |
| """Reset password using OTP (forgot-password flow).""" | |
| email: EmailStr = Field(..., description="Email address") | |
| otp: str = Field(..., description="6-digit OTP received via WhatsApp/SMS", min_length=6, max_length=6) | |
| new_password: str = Field(..., description="New password", min_length=6, max_length=100) | |
| class ResetPasswordRequest(BaseModel): | |
| """Reset password with token request schema.""" | |
| token: str = Field(..., description="Password reset token", min_length=20) | |
| new_password: str = Field(..., description="New password", min_length=6, max_length=100) | |
| class VerifyResetTokenRequest(BaseModel): | |
| """Verify reset token request schema.""" | |
| token: str = Field(..., description="Password reset token to verify") | |
| class UserListResponse(BaseModel): | |
| """User list response schema.""" | |
| users: List[UserInfoResponse] = Field(..., description="List of users") | |
| total_count: int = Field(..., description="Total number of users") | |
| page: int = Field(..., description="Current page number") | |
| page_size: int = Field(..., description="Number of items per page") | |
| class TokenRefreshRequest(BaseModel): | |
| """Token refresh request schema.""" | |
| refresh_token: str = Field(..., description="Refresh token") | |
| class LogoutRequest(BaseModel): | |
| """Logout request schema.""" | |
| logout_from_all_devices: bool = Field(default=False, description="Logout from all devices") | |
| class UserListRequest(BaseModel): | |
| """User list request schema with projection support.""" | |
| filters: Optional[Dict[str, Any]] = Field(default_factory=dict, description="Filter criteria") | |
| skip: int = Field(default=0, description="Number of records to skip", ge=0) | |
| limit: int = Field(default=100, description="Maximum number of records to return", ge=1, le=1000) | |
| projection_list: Optional[List[str]] = Field( | |
| None, | |
| description="List of fields to include in response (user_id, username, email, merchant_id, merchant_type, first_name, last_name, role, status, etc.)" | |
| ) | |
| # Specific filters | |
| status_filter: Optional[UserStatus] = Field(None, description="Filter by user status") | |
| role_filter: Optional[str] = Field(None, description="Filter by user role") | |
| merchant_id_filter: Optional[str] = Field(None, description="Filter by merchant ID") | |
| merchant_type_filter: Optional[str] = Field(None, description="Filter by merchant type") | |
| # Response models | |
| class StandardResponse(BaseModel): | |
| """Standard API response.""" | |
| success: bool = Field(..., description="Operation success status") | |
| message: str = Field(..., description="Response message") | |
| data: Optional[dict] = Field(None, description="Response data") | |
| # Update forward references | |
| LoginResponse.model_rebuild() | |