style: apply Black code formatting

Author: tanzilahmed0

backend/api/auth.py

@@ -7,7 +7,7 @@
 from fastapi.security import HTTPBearer
 
 from models.response_schemas import ApiResponse, AuthResponse, LoginRequest, User
-from models.user import UserPublic
+from models.user import UserInDB
 from services.auth_service import AuthService
 
 # Configure logging
@@ -38,17 +38,14 @@ async def login_with_google(request: LoginRequest) -> ApiResponse[AuthResponse]:
             request.google_token.strip()
         )
 
-        # Convert UserInDB to UserPublic for API response
-        public_user = UserPublic.from_db_user(user)
-
-        # Convert to response format expected by frontend
+        # Convert UserInDB to the response model directly
         user_response = User(
-            id=public_user.id,
-            email=public_user.email,
-            name=public_user.name,
-            avatar_url=public_user.avatar_url,
-            created_at=public_user.created_at,
-            last_sign_in_at=public_user.last_sign_in_at,
+            id=str(user.id),
+            email=user.email,
+            name=user.name,
+            avatar_url=user.avatar_url,
+            created_at=user.created_at.isoformat(),
+            last_sign_in_at=user.updated_at.isoformat(),  # Using updated_at for last sign-in
         )
 
         auth_response = AuthResponse(
@@ -71,6 +68,9 @@ async def login_with_google(request: LoginRequest) -> ApiResponse[AuthResponse]:
             ),
         )
 
+    except HTTPException:
+        # Re-raise HTTPException without modification
+        raise
     except ValueError as e:
         logger.error(f"Google OAuth validation error: {str(e)}")
         raise HTTPException(status_code=401, detail=f"Invalid Google token: {str(e)}")
@@ -88,15 +88,15 @@ async def get_current_user(
         logger.info("Received current user request")
 
         user = auth_service.get_current_user(token)
-        public_user = UserPublic.from_db_user(user)
 
+        # Convert UserInDB to the response model directly
         user_response = User(
-            id=public_user.id,
-            email=public_user.email,
-            name=public_user.name,
-            avatar_url=public_user.avatar_url,
-            created_at=public_user.created_at,
-            last_sign_in_at=public_user.last_sign_in_at,
+            id=str(user.id),
+            email=user.email,
+            name=user.name,
+            avatar_url=user.avatar_url,
+            created_at=user.created_at.isoformat(),
+            last_sign_in_at=user.updated_at.isoformat(),  # Using updated_at for last sign-in
         )
 
         logger.info(f"Current user request successful for: {user.email}")
@@ -164,14 +164,13 @@ async def refresh_token(request: dict) -> ApiResponse[AuthResponse]:
         )
 
         # Convert to response format
-        public_user = UserPublic.from_db_user(user)
         user_response = User(
-            id=public_user.id,
-            email=public_user.email,
-            name=public_user.name,
-            avatar_url=public_user.avatar_url,
-            created_at=public_user.created_at,
-            last_sign_in_at=public_user.last_sign_in_at,
+            id=str(user.id),
+            email=user.email,
+            name=user.name,
+            avatar_url=user.avatar_url,
+            created_at=user.created_at.isoformat(),
+            last_sign_in_at=user.updated_at.isoformat(),
         )
 
         auth_response = AuthResponse(
@@ -186,6 +185,9 @@ async def refresh_token(request: dict) -> ApiResponse[AuthResponse]:
             success=True, data=auth_response, message="Token refreshed successfully"
         )
 
+    except HTTPException:
+        # Re-raise HTTPException without modification
+        raise
     except jwt.InvalidTokenError as e:
         logger.warning(f"Invalid refresh token: {str(e)}")
         raise HTTPException(
@@ -219,6 +221,9 @@ async def auth_health_check() -> ApiResponse[dict]:
                 detail=f"Authentication service is unhealthy: {health_data.get('error', 'Unknown error')}",
             )
 
+    except HTTPException:
+        # Re-raise HTTPException without modification
+        raise
     except Exception as e:
         logger.error(f"Auth health check error: {str(e)}")
         raise HTTPException(status_code=500, detail=f"Health check failed: {str(e)}")

backend/api/health.py

@@ -4,7 +4,7 @@
 
 from fastapi import APIRouter
 
-from services.database_service import db_service
+from services.database_service import get_db_service
 from services.redis_service import redis_service
 from services.storage_service import storage_service
 
@@ -44,7 +44,7 @@ async def health_check() -> Dict[str, Any]:
         }
 
     # Check all services in production
-    database_health = db_service.health_check()
+    database_health = get_db_service().health_check()
     redis_health = redis_service.health_check()
     storage_health = storage_service.health_check()
 

backend/models/base.py

@@ -0,0 +1,3 @@
+from sqlalchemy.orm import declarative_base
+
+Base = declarative_base()

backend/models/user.py

@@ -1,146 +1,139 @@
 import uuid
 from datetime import datetime
-from typing import Optional
+from typing import Optional, List
 
 from pydantic import BaseModel, EmailStr, Field, field_validator
-from sqlalchemy import Boolean, Column, DateTime, String, Text
-from sqlalchemy.dialects.postgresql import UUID
-from sqlalchemy.orm import declarative_base, relationship
-
-Base = declarative_base()
+from sqlalchemy import Boolean, Column, DateTime, String, Text, func, TypeDecorator
+from sqlalchemy.dialects.postgresql import UUID as PG_UUID
+from sqlalchemy.orm import Mapped, mapped_column, relationship, declarative_base
+
+from models.base import Base
+
+
+class UUID(TypeDecorator):
+    """
+    Platform-independent UUID type.
+
+    Uses PostgreSQL's UUID type, otherwise uses
+    CHAR(32), storing as string.
+    """
+
+    impl = PG_UUID
+    cache_ok = True
+
+    def load_dialect_impl(self, dialect):
+        if dialect.name == "postgresql":
+            return dialect.type_descriptor(PG_UUID())
+        else:
+            return dialect.type_descriptor(String(32))
+
+    def process_bind_param(self, value, dialect):
+        if value is None:
+            return value
+        elif dialect.name == "postgresql":
+            return str(value)
+        else:
+            if not isinstance(value, uuid.UUID):
+                return "%.32x" % uuid.UUID(value).int
+            else:
+                # hexstring
+                return "%.32x" % value.int
+
+    def process_result_value(self, value, dialect):
+        if value is None:
+            return value
+        else:
+            if not isinstance(value, uuid.UUID):
+                value = uuid.UUID(value)
+            return value
 
 
 class UserTable(Base):
     """SQLAlchemy User table model for PostgreSQL"""
 
     __tablename__ = "users"
 
-    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+    id: Mapped[uuid.UUID] = mapped_column(UUID, primary_key=True, default=uuid.uuid4)
     email = Column(String(255), unique=True, nullable=False, index=True)
-    name = Column(String(255), nullable=False)
+    name: Mapped[str] = mapped_column(String(255), nullable=False)
     avatar_url = Column(Text, nullable=True)
     google_id = Column(String(255), unique=True, nullable=True, index=True)
-    is_active = Column(Boolean, default=True, nullable=False)
-    is_verified = Column(Boolean, default=False, nullable=False)
-    created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
-    updated_at = Column(
-        DateTime, default=datetime.utcnow, onupdate=datetime.utcnow, nullable=False
+    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
+    is_verified: Mapped[bool] = mapped_column(Boolean, default=False)
+
+    # Timestamps
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now()
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), onupdate=func.now()
     )
-    last_sign_in_at = Column(DateTime, nullable=True)
 
     # Relationships
-    # projects = relationship(
-    #     "ProjectTable", back_populates="user", cascade="all, delete"
+    # projects: Mapped[List["ProjectTable"]] = relationship(
+    #     back_populates="user", cascade="all, delete-orphan"
     # )
-    # chat_messages = relationship(
-    #     "ChatMessageTable", back_populates="user", cascade="all, delete"
+    # chat_messages: Mapped[List["ChatMessageTable"]] = relationship(
+    #     back_populates="user", cascade="all, delete-orphan"
     # )
 
     def __repr__(self):
-        return f"<User(id={self.id}, email={self.email}, name={self.name})>"
-
+        return f"<User(id={self.id}, email='{self.email}', name='{self.name}')>"
 
-class UserCreate(BaseModel):
-    """Pydantic model for creating a user"""
 
-    email: EmailStr = Field(..., description="User email address")
-    name: str = Field(..., min_length=1, max_length=255, description="User full name")
-    avatar_url: Optional[str] = Field(None, description="User avatar URL")
-    google_id: Optional[str] = Field(None, description="Google OAuth ID")
+# Pydantic models for API validation and serialization
 
-    @field_validator("name")
-    @classmethod
-    def validate_name(cls, v):
-        if not v or not v.strip():
-            raise ValueError("Name cannot be empty or just whitespace")
-        return v.strip()
 
-    @field_validator("avatar_url")
-    @classmethod
-    def validate_avatar_url(cls, v):
-        if v and not v.startswith(("http://", "https://")):
-            raise ValueError("Avatar URL must be a valid HTTP/HTTPS URL")
-        return v
+class UserBase(BaseModel):
+    email: EmailStr
+    name: Optional[str] = None
+    avatar_url: Optional[str] = None
+    is_active: bool = True
+    is_verified: bool = False
 
+    class Config:
+        from_attributes = True
 
-class UserUpdate(BaseModel):
-    """Pydantic model for updating a user"""
 
-    name: Optional[str] = Field(None, min_length=1, max_length=255)
-    avatar_url: Optional[str] = Field(None)
-    is_active: Optional[bool] = Field(None)
-    is_verified: Optional[bool] = Field(None)
-    last_sign_in_at: Optional[datetime] = Field(None)
+class UserCreate(UserBase):
+    google_id: str
+    name: str  # Make name required for UserCreate
 
-    @field_validator("name")
+    @field_validator("name", "google_id")
     @classmethod
-    def validate_name(cls, v):
-        if v is not None and (not v or not v.strip()):
-            raise ValueError("Name cannot be empty or just whitespace")
-        return v.strip() if v else v
+    def validate_non_empty(cls, v):
+        if not v or not v.strip():
+            raise ValueError("Field cannot be empty")
+        return v.strip()
 
-    @field_validator("avatar_url")
-    @classmethod
-    def validate_avatar_url(cls, v):
-        if v and not v.startswith(("http://", "https://")):
-            raise ValueError("Avatar URL must be a valid HTTP/HTTPS URL")
-        return v
 
+class UserUpdate(BaseModel):
+    name: Optional[str] = None
+    avatar_url: Optional[str] = None
 
-class UserInDB(BaseModel):
-    """Pydantic model for user data from database"""
 
+class UserInDB(UserBase):
     id: uuid.UUID
-    email: str
-    name: str
-    avatar_url: Optional[str] = None
-    google_id: Optional[str] = None
-    is_active: bool
-    is_verified: bool
     created_at: datetime
     updated_at: datetime
-    last_sign_in_at: Optional[datetime] = None
-
-    model_config = {"from_attributes": True}
-
 
-class UserPublic(BaseModel):
-    """Pydantic model for public user data (API responses)"""
-
-    id: str
-    email: str
-    name: str
-    avatar_url: Optional[str] = None
-    created_at: str
-    last_sign_in_at: Optional[str] = None
-
-    @classmethod
-    def from_db_user(cls, user: UserInDB) -> "UserPublic":
-        """Convert database user to public user model"""
-        return cls(
-            id=str(user.id),
-            email=user.email,
-            name=user.name,
-            avatar_url=user.avatar_url,
-            created_at=user.created_at.isoformat() + "Z",
-            last_sign_in_at=(
-                user.last_sign_in_at.isoformat() + "Z" if user.last_sign_in_at else None
-            ),
-        )
+    class Config:
+        from_attributes = True
 
 
 class GoogleOAuthData(BaseModel):
-    """Pydantic model for Google OAuth data"""
-
     google_id: str
     email: EmailStr
     name: str
     avatar_url: Optional[str] = None
-    email_verified: bool = False
+    email_verified: bool = True
 
-    @field_validator("google_id")
+    @field_validator("name", "google_id", "email")
     @classmethod
-    def validate_google_id(cls, v):
+    def strip_whitespace(cls, v):
         if not v or not v.strip():
-            raise ValueError("Google ID cannot be empty")
+            raise ValueError("Field cannot be empty")
         return v.strip()
+
+    class Config:
+        from_attributes = True