Merge pull request #14 from hussainwali74/schemas_improvement

refactor session schema

Author: igorbenav

crudadmin/session/schemas.py

@@ -1,24 +1,65 @@
 from datetime import UTC, datetime
+from enum import Enum
 from typing import Any, Optional
 from uuid import uuid4
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, field_validator
 
 from ..core.schemas.timestamp import TimestampSchema
 
 
-class SessionData(BaseModel):
-    """Common base data for any user session."""
+class DeviceType(str, Enum):
+    """Device type enumeration"""
 
-    user_id: int
-    session_id: str = Field(default_factory=lambda: str(uuid4()))
-    ip_address: str
-    user_agent: str
-    device_info: dict[str, Any] = Field(default_factory=dict)
-    created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
-    last_activity: datetime = Field(default_factory=lambda: datetime.now(UTC))
-    is_active: bool = True
-    metadata: dict[str, Any] = Field(default_factory=dict)
+    MOBILE = "mobile"
+    TABLET = "tablet"
+    DESKTOP = "desktop"
+    UNKNOWN = "unknown"
+
+
+class BaseSession(BaseModel):
+    """Common base fields for all session types."""
+
+    user_id: int = Field(
+        ..., description="The ID of the user associated with the session."
+    )
+
+    session_id: str = Field(
+        default_factory=lambda: str(uuid4()), description="Unique session identifier"
+    )
+    ip_address: str = Field(..., description="Client IP address")
+    user_agent: str = Field(..., description="Client user agent string")
+    device_info: dict[str, Any] = Field(
+        default_factory=dict, description="Additional device information"
+    )
+    is_active: bool = Field(default=True, description="Whether session is active")
+
+    @field_validator("ip_address")
+    @classmethod
+    def validate_ip_address(cls, ip: str) -> str:
+        """Validate IP address format."""
+        import ipaddress
+
+        try:
+            ipaddress.ip_address(ip)
+        except ValueError as err:
+            raise ValueError("Invalid IP Address format") from err
+        return ip
+
+
+class SessionData(BaseSession):
+    """Common session data for any user session."""
+
+    metadata: dict[str, Any] = Field(
+        default_factory=dict, description="Additional session metadata"
+    )
+    created_at: datetime = Field(
+        default_factory=lambda: datetime.now(UTC),
+        description="Session creation timestamp",
+    )
+    last_activity: datetime = Field(
+        default_factory=lambda: datetime.now(UTC), description="Last activity timestamp"
+    )
 
 
 class SessionCreate(SessionData):
@@ -38,9 +79,9 @@ class SessionUpdate(BaseModel):
 class UserAgentInfo(BaseModel):
     """User agent information parsed from the User-Agent header."""
 
-    browser: str
-    browser_version: str
-    os: str
+    browser: str = Field(..., description="Browser name")
+    browser_version: str = Field(..., description="Browser version")
+    os: str = Field(..., description="Operating System")
     device: str
     is_mobile: bool
     is_tablet: bool
@@ -57,46 +98,32 @@ class CSRFToken(BaseModel):
     expires_at: datetime
 
 
-class AdminSessionBase(BaseModel):
-    """Base schema for AdminSession."""
-
-    user_id: int
-    session_id: str
-    ip_address: str
-    user_agent: str
-    device_info: dict[str, Any] = Field(default_factory=dict)
-    session_metadata: dict[str, Any] = Field(default_factory=dict)
-    is_active: bool = True
-
-
-class AdminSession(TimestampSchema, AdminSessionBase):
+class AdminSession(TimestampSchema, BaseSession):
     """Full AdminSession schema with all fields."""
 
     id: int
-    created_at: datetime
-    last_activity: datetime
+    session_metadata: dict[str, Any] = Field(
+        default_factory=dict, description="admin specific session metadata"
+    )
+    # remove redundant timestamp fields - let TimestampSchema handle them
 
 
-class AdminSessionRead(BaseModel):
+class AdminSessionRead(BaseSession):
     """Schema for reading AdminSession data."""
 
     id: int
-    user_id: int
-    session_id: str
-    ip_address: str
-    user_agent: str
-    device_info: dict[str, Any]
     session_metadata: dict[str, Any]
     created_at: datetime
     last_activity: datetime
     is_active: bool
 
 
-class AdminSessionCreate(AdminSessionBase):
+class AdminSessionCreate(BaseSession):
     """Schema for creating AdminSession in database."""
 
     created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
     last_activity: datetime = Field(default_factory=lambda: datetime.now(UTC))
+    session_metadata: dict[str, Any] = Field(default_factory=dict)
 
 
 class AdminSessionUpdate(BaseModel):
@@ -119,7 +146,6 @@ class AdminSessionUpdateInternal(AdminSessionUpdate):
     "SessionUpdate",
     "UserAgentInfo",
     "CSRFToken",
-    "AdminSessionBase",
     "AdminSession",
     "AdminSessionRead",
     "AdminSessionCreate",