Spaces:
Sleeping
Sleeping
| """ | |
| User Model β ORM + Pydantic Schemas for Authentication | |
| ORM: User (SQLAlchemy table: users) | |
| Schemas: UserCreate, UserLogin, UserResponse, TokenResponse | |
| Architecture Reference: insights/architecture_rule.md | |
| Phase 2: "Implement JWT Authentication and RBAC in auth_controller.py" | |
| """ | |
| import uuid | |
| from datetime import datetime | |
| from typing import Optional | |
| from enum import Enum | |
| from sqlalchemy import String, DateTime, Boolean, func | |
| from sqlalchemy.dialects.postgresql import UUID | |
| from sqlalchemy.orm import Mapped, mapped_column | |
| from pydantic import BaseModel, Field, ConfigDict, EmailStr | |
| from models.base import Base | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # Enums | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| class UserRole(str, Enum): | |
| """Role-Based Access Control roles.""" | |
| admin = "admin" # Full system access | |
| inspector = "inspector" # Ghost Shopper field inspector | |
| viewer = "viewer" # Read-only dashboard access | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # ORM Model | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| class User(Base): | |
| """SQLAlchemy ORM model for the 'users' table.""" | |
| __tablename__ = "users" | |
| id: Mapped[uuid.UUID] = mapped_column( | |
| UUID(as_uuid=True), primary_key=True, default=uuid.uuid4 | |
| ) | |
| email: Mapped[str] = mapped_column( | |
| String(255), unique=True, nullable=False, index=True | |
| ) | |
| hashed_password: Mapped[str] = mapped_column( | |
| String(255), nullable=False | |
| ) | |
| full_name: Mapped[str] = mapped_column( | |
| String(255), nullable=False | |
| ) | |
| role: Mapped[str] = mapped_column( | |
| String(20), nullable=False, default=UserRole.inspector.value | |
| ) | |
| is_active: Mapped[bool] = mapped_column( | |
| Boolean, nullable=False, default=True | |
| ) | |
| fcm_token: Mapped[Optional[str]] = mapped_column( | |
| String(500), nullable=True, | |
| comment="Firebase Cloud Messaging token for push notifications" | |
| ) | |
| last_login: Mapped[Optional[datetime]] = mapped_column( | |
| DateTime, nullable=True | |
| ) | |
| created_at: Mapped[datetime] = mapped_column( | |
| DateTime, server_default=func.now() | |
| ) | |
| updated_at: Mapped[datetime] = mapped_column( | |
| DateTime, server_default=func.now(), onupdate=func.now() | |
| ) | |
| def __repr__(self) -> str: | |
| return f"<User(id={self.id}, email='{self.email}', role='{self.role}')>" | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # Pydantic Schemas | |
| # ββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| class UserCreate(BaseModel): | |
| """Schema for user registration.""" | |
| email: str = Field(..., max_length=255, description="User email address") | |
| password: str = Field(..., min_length=8, max_length=128, description="Plain-text password (will be hashed)") | |
| full_name: str = Field(..., min_length=1, max_length=255, description="Full display name") | |
| role: UserRole = Field(UserRole.inspector, description="User role: admin, inspector, viewer") | |
| class UserLogin(BaseModel): | |
| """Schema for login request.""" | |
| email: str = Field(..., description="Registered email address") | |
| password: str = Field(..., description="Plain-text password") | |
| fcm_token: Optional[str] = Field(None, description="FCM token for push notifications") | |
| class UserResponse(BaseModel): | |
| """Schema for user info in API responses (no password).""" | |
| model_config = ConfigDict(from_attributes=True) | |
| id: uuid.UUID | |
| email: str | |
| full_name: str | |
| role: str | |
| is_active: bool | |
| last_login: Optional[datetime] | |
| created_at: datetime | |
| class TokenResponse(BaseModel): | |
| """Schema for JWT token response.""" | |
| access_token: str | |
| token_type: str = "bearer" | |
| expires_in: int = Field(description="Token lifetime in seconds") | |
| user: UserResponse | |