b9dfb86260
Security hardening release addressing CodeQL and Dependabot alerts: - Fix stack trace exposure in error responses - Add SSRF protection with DNS resolution checking - Implement proper URL hostname validation (replaces substring matching) - Add centralized path sanitization to prevent path traversal - Fix ReDoS vulnerability in email validation regex - Improve HTML sanitization in validation utilities - Fix capability wildcard matching in auth utilities - Update glob dependency to address CVE - Add CodeQL suppression comments for verified false positives 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
126 lines
4.1 KiB
Python
126 lines
4.1 KiB
Python
"""
|
|
GT 2.0 Base Model Classes - Service-Based Architecture
|
|
|
|
Provides Pydantic models for data serialization with the DuckDB service.
|
|
No SQLAlchemy ORM dependency - pure Python/Pydantic models.
|
|
"""
|
|
|
|
from typing import Any, Dict, Optional, List, Type, TypeVar
|
|
from datetime import datetime
|
|
import uuid
|
|
from pydantic import BaseModel, Field, ConfigDict
|
|
|
|
# Generic type for model classes
|
|
T = TypeVar('T', bound='BaseServiceModel')
|
|
|
|
|
|
class BaseServiceModel(BaseModel):
|
|
"""
|
|
Base model for all GT 2.0 entities using service-based architecture.
|
|
|
|
Replaces SQLAlchemy models with Pydantic models + DuckDB service.
|
|
"""
|
|
|
|
# Pydantic v2 configuration
|
|
model_config = ConfigDict(
|
|
from_attributes=True,
|
|
validate_assignment=True,
|
|
arbitrary_types_allowed=True,
|
|
use_enum_values=True
|
|
)
|
|
|
|
# Standard fields for all models
|
|
id: str = Field(default_factory=lambda: str(uuid.uuid4()), description="Unique identifier")
|
|
created_at: datetime = Field(default_factory=datetime.utcnow, description="Creation timestamp")
|
|
updated_at: datetime = Field(default_factory=datetime.utcnow, description="Last update timestamp")
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Convert model to dictionary"""
|
|
return self.model_dump()
|
|
|
|
@classmethod
|
|
def from_dict(cls: Type[T], data: Dict[str, Any]) -> T:
|
|
"""Create model instance from dictionary"""
|
|
return cls(**data)
|
|
|
|
@classmethod
|
|
def from_row(cls: Type[T], row: Dict[str, Any]) -> T:
|
|
"""Create model instance from database row"""
|
|
# Convert database row to model, handling type conversions
|
|
model_data = {}
|
|
|
|
for field_name, field_info in cls.model_fields.items():
|
|
if field_name in row:
|
|
value = row[field_name]
|
|
|
|
# Handle datetime conversion
|
|
if field_info.annotation == datetime and isinstance(value, str):
|
|
try:
|
|
value = datetime.fromisoformat(value)
|
|
except ValueError:
|
|
value = datetime.utcnow()
|
|
|
|
model_data[field_name] = value
|
|
|
|
return cls(**model_data)
|
|
|
|
def update_timestamp(self):
|
|
"""Update the updated_at timestamp"""
|
|
self.updated_at = datetime.utcnow()
|
|
|
|
|
|
class BaseCreateModel(BaseModel):
|
|
"""Base model for creation requests"""
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
|
|
class BaseUpdateModel(BaseModel):
|
|
"""Base model for update requests"""
|
|
model_config = ConfigDict(from_attributes=True)
|
|
updated_at: datetime = Field(default_factory=datetime.utcnow)
|
|
|
|
|
|
class BaseResponseModel(BaseServiceModel):
|
|
"""Base model for API responses"""
|
|
pass
|
|
|
|
|
|
# Legacy compatibility - some files might still import Base
|
|
Base = BaseServiceModel # For backwards compatibility during migration
|
|
|
|
|
|
# Database service integration helpers
|
|
class DatabaseMixin:
|
|
"""Mixin providing database service integration methods"""
|
|
|
|
@classmethod
|
|
async def get_table_name(cls) -> str:
|
|
"""Get the database table name for this model"""
|
|
# Convert CamelCase to snake_case and pluralize
|
|
name = cls.__name__.lower()
|
|
if name.endswith('y'):
|
|
name = name[:-1] + 'ies'
|
|
elif name.endswith('s'):
|
|
name = name + 'es'
|
|
else:
|
|
name = name + 's'
|
|
return name
|
|
|
|
@classmethod
|
|
async def create_sql(cls) -> str:
|
|
"""Generate CREATE TABLE SQL for this model"""
|
|
# This would generate SQL based on Pydantic field types
|
|
# For now, return placeholder - actual schemas are in DuckDB service
|
|
table_name = await cls.get_table_name()
|
|
return f"-- CREATE TABLE {table_name} generated by DuckDB service"
|
|
|
|
async def to_sql_values(self) -> Dict[str, Any]:
|
|
"""Convert model to SQL-safe values"""
|
|
data = self.to_dict()
|
|
|
|
# Convert datetime objects to ISO strings
|
|
for key, value in data.items():
|
|
if isinstance(value, datetime):
|
|
data[key] = value.isoformat()
|
|
|
|
return data |