feat(db): add Settings model with risk profiles and alert preferences (#5) - Implements RiskProfile enum, risk parameters, JSON alert_preferences, one-to-one User relationship, CheckConstraints, cascade delete, 43 tests

This commit is contained in:
Andrew Kaszubski 2025-12-26 14:16:42 +11:00
parent 0d09f15bd6
commit 1c6c2fadf1
10 changed files with 1854 additions and 11 deletions

View File

@ -74,6 +74,24 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- Unit tests covering field validation, defaults, constraints, and enum handling
- Integration tests for relationships, cascade delete, and concurrent operations
- Settings model for user preferences and risk management (Issue #5: DB-4)
- Settings model for managing user trading preferences, risk tolerance, and alert configurations [file:tradingagents/api/models/settings.py](tradingagents/api/models/settings.py)
- RiskProfile enum for user risk tolerance profiles (CONSERVATIVE, MODERATE, AGGRESSIVE) [file:tradingagents/api/models/settings.py:70-82](tradingagents/api/models/settings.py)
- One-to-one relationship with User model with cascade delete behavior
- Risk management fields: risk_score (0-10 scale), risk_profile, max_position_pct (0-100), max_portfolio_risk_pct (0-100)
- Investment horizon field in years for long-term portfolio planning
- Alert preferences JSON field for configurable email/SMS/push notifications
- Unique constraint on user_id enforcing one settings record per user
- Check constraints for valid numeric ranges: risk_score 0-10, position/portfolio risk 0-100, horizon >= 0
- Automatic timestamps via TimestampMixin (created_at, updated_at)
- Validators for risk_profile enum normalization and type safety [file:tradingagents/api/models/settings.py:246-279](tradingagents/api/models/settings.py)
- Event listener validation (before_flush) for business rule enforcement [file:tradingagents/api/models/settings.py:284-313](tradingagents/api/models/settings.py)
- User model updated with settings relationship for one-to-one association [file:tradingagents/api/models/user.py](tradingagents/api/models/user.py)
- Database migration 004_add_settings_model.py with comprehensive schema definition [file:migrations/versions/004_add_settings_model.py](migrations/versions/004_add_settings_model.py)
- Migration with proper upgrade and downgrade functions for reversible schema changes
- Comprehensive unit test suite covering field validation, defaults, constraints, and enum handling [file:tests/unit/api/test_settings_model.py](tests/unit/api/test_settings_model.py)
- Integration test suite for relationships, cascade delete, and concurrent operations [file:tests/integration/api/test_settings_integration.py](tests/integration/api/test_settings_integration.py)
- Test fixtures directory with centralized mock data (Issue #51)
- FixtureLoader class for loading JSON fixtures with automatic datetime parsing [file:tests/fixtures/__init__.py](tests/fixtures/__init__.py)
- Stock data fixtures: US market OHLCV, Chinese market OHLCV, standardized data [file:tests/fixtures/stock_data/](tests/fixtures/stock_data/)

View File

@ -0,0 +1,196 @@
"""Add Settings model for user preferences and risk management
Revision ID: 004
Revises: 003
Create Date: 2025-12-26 14:30:00.000000
This migration adds the settings table for managing user trading preferences,
risk profiles, and alert configurations. Each user has exactly one Settings
record (one-to-one relationship).
Table: settings
Columns:
- id: Primary key, auto-increment
- user_id: Foreign key to users.id (cascade delete, unique)
- risk_profile: Risk tolerance profile (ENUM: CONSERVATIVE, MODERATE, AGGRESSIVE)
- risk_score: Numeric risk score 0-10 (NUMERIC 5,2, default: 5.0)
- max_position_pct: Max % of portfolio for single position (NUMERIC 5,2, default: 10.0)
- max_portfolio_risk_pct: Max portfolio-wide risk % (NUMERIC 5,2, default: 2.0)
- investment_horizon_years: Investment time horizon in years (INTEGER, default: 5)
- alert_preferences: JSON config for notifications (TEXT/JSON, default: '{}')
- created_at: Timestamp when created (auto)
- updated_at: Timestamp when last updated (auto)
Constraints:
- UNIQUE (user_id): One settings per user (one-to-one)
- CHECK risk_score >= 0 AND risk_score <= 10
- CHECK max_position_pct >= 0 AND max_position_pct <= 100
- CHECK max_portfolio_risk_pct >= 0 AND max_portfolio_risk_pct <= 100
- CHECK investment_horizon_years >= 0
Indexes:
- ix_settings_user_id: Unique index on user_id (one-to-one lookup)
Relationships:
- settings.user_id -> users.id (CASCADE DELETE)
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = '004'
down_revision: Union[str, None] = '003'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
"""Create settings table with all constraints and indexes.
Creates the settings table for managing user trading preferences
with proper foreign keys, constraints, and indexes.
"""
# Create settings table
op.create_table(
'settings',
# Primary key
sa.Column(
'id',
sa.Integer(),
nullable=False,
primary_key=True,
autoincrement=True
),
# Foreign key to users (cascade delete, unique for one-to-one)
sa.Column(
'user_id',
sa.Integer(),
sa.ForeignKey('users.id', ondelete='CASCADE'),
nullable=False,
unique=True,
comment='User who owns these settings (one-to-one)'
),
# Risk profile enum
sa.Column(
'risk_profile',
sa.String(length=20),
nullable=False,
server_default='MODERATE',
comment='Risk tolerance: CONSERVATIVE, MODERATE, or AGGRESSIVE'
),
# Risk score (0-10 scale with 2 decimal places)
sa.Column(
'risk_score',
sa.Numeric(precision=5, scale=2),
nullable=False,
server_default='5.0',
comment='Numeric risk score from 0 (conservative) to 10 (aggressive)'
),
# Position sizing limits (percentages with 2 decimal places)
sa.Column(
'max_position_pct',
sa.Numeric(precision=5, scale=2),
nullable=False,
server_default='10.0',
comment='Maximum percentage of portfolio for single position (0-100)'
),
sa.Column(
'max_portfolio_risk_pct',
sa.Numeric(precision=5, scale=2),
nullable=False,
server_default='2.0',
comment='Maximum portfolio-wide risk percentage (0-100)'
),
# Investment horizon
sa.Column(
'investment_horizon_years',
sa.Integer(),
nullable=False,
server_default='5',
comment='Investment time horizon in years'
),
# Alert preferences (JSON)
# Use TEXT for SQLite compatibility, PostgreSQL will handle as JSON
sa.Column(
'alert_preferences',
sa.Text(),
nullable=False,
server_default='{}',
comment='JSON configuration for email/SMS/push notifications'
),
# Timestamps (from TimestampMixin)
sa.Column(
'created_at',
sa.DateTime(timezone=True),
nullable=False,
server_default=sa.text('CURRENT_TIMESTAMP'),
comment='Timestamp when settings were created'
),
sa.Column(
'updated_at',
sa.DateTime(timezone=True),
nullable=False,
server_default=sa.text('CURRENT_TIMESTAMP'),
comment='Timestamp when settings were last updated'
),
# Table-level constraints
# Unique constraint: one settings per user
sa.UniqueConstraint(
'user_id',
name='uq_settings_user_id'
),
# Check constraints: valid numeric ranges
sa.CheckConstraint(
'risk_score >= 0 AND risk_score <= 10',
name='ck_settings_risk_score_range'
),
sa.CheckConstraint(
'max_position_pct >= 0 AND max_position_pct <= 100',
name='ck_settings_max_position_pct_range'
),
sa.CheckConstraint(
'max_portfolio_risk_pct >= 0 AND max_portfolio_risk_pct <= 100',
name='ck_settings_max_portfolio_risk_pct_range'
),
sa.CheckConstraint(
'investment_horizon_years >= 0',
name='ck_settings_investment_horizon_positive'
),
)
# Create unique index on user_id for efficient one-to-one lookups
op.create_index(
'ix_settings_user_id',
'settings',
['user_id'],
unique=True
)
def downgrade() -> None:
"""Drop settings table and all associated indexes and constraints.
WARNING: This will permanently delete all settings data!
"""
# Drop index
op.drop_index('ix_settings_user_id', 'settings')
# Drop the table (constraints are dropped automatically)
op.drop_table('settings')

View File

@ -28,6 +28,9 @@ from typing import Dict, Any
from tradingagents.default_config import DEFAULT_CONFIG
# Register plugins from sub-conftest files (pytest 9.0+ requires this at root level)
pytest_plugins = ["tests.api.conftest"]
# ============================================================================
# Environment Variable Fixtures

View File

@ -1,11 +1,8 @@
"""
Shared pytest fixtures for integration API tests.
This module imports fixtures from the main API conftest
to make them available to integration tests.
This module may contain integration-specific overrides or fixtures.
Common fixtures are imported from tests.api.conftest via the root conftest.py.
"""
import pytest
# Import all fixtures from main API conftest
pytest_plugins = ["tests.api.conftest"]

View File

@ -0,0 +1,317 @@
"""Integration tests for Settings model (Issue #5: DB-4).
Integration tests covering:
- Settings creation with related User entity
- Querying settings by user
- Updating settings for a user
- Complex alert preferences scenarios
- Multi-user settings isolation
- Settings deletion and cascade behavior
Follows TDD principles with comprehensive coverage.
Tests written BEFORE implementation (RED phase).
"""
import pytest
from decimal import Decimal
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
# Mark all tests in this module as asyncio
pytestmark = pytest.mark.asyncio
class TestSettingsIntegration:
"""Integration tests for Settings model with User relationship."""
@pytest.mark.asyncio
async def test_create_settings_for_user(self, db_session, test_user):
"""Should create settings for a user and retrieve them."""
try:
from tradingagents.api.models.settings import Settings, RiskProfile
# Create settings
settings = Settings(
user_id=test_user.id,
risk_profile=RiskProfile.MODERATE,
risk_score=Decimal("6.0"),
max_position_pct=Decimal("15.0"),
max_portfolio_risk_pct=Decimal("3.0"),
investment_horizon_years=7,
alert_preferences={
"email": {
"enabled": True,
"address": test_user.email,
"alert_types": ["price_alert", "portfolio_alert"]
}
},
)
db_session.add(settings)
await db_session.commit()
await db_session.refresh(settings)
# Retrieve settings by user_id
result = await db_session.execute(
select(Settings).where(Settings.user_id == test_user.id)
)
retrieved_settings = result.scalar_one()
# Verify
assert retrieved_settings.id == settings.id
assert retrieved_settings.user_id == test_user.id
assert retrieved_settings.risk_profile == RiskProfile.MODERATE
assert retrieved_settings.risk_score == Decimal("6.0")
assert retrieved_settings.alert_preferences["email"]["address"] == test_user.email
except ImportError:
pytest.skip("Settings model not yet implemented (TDD RED phase)")
@pytest.mark.asyncio
async def test_update_user_settings(self, db_session, test_user):
"""Should update existing settings for a user."""
try:
from tradingagents.api.models.settings import Settings, RiskProfile
# Create initial settings
settings = Settings(
user_id=test_user.id,
risk_profile=RiskProfile.CONSERVATIVE,
risk_score=Decimal("3.0"),
)
db_session.add(settings)
await db_session.commit()
await db_session.refresh(settings)
initial_id = settings.id
# Update settings
settings.risk_profile = RiskProfile.AGGRESSIVE
settings.risk_score = Decimal("8.5")
settings.max_position_pct = Decimal("25.0")
settings.alert_preferences = {
"sms": {
"enabled": True,
"phone": "+1234567890",
}
}
await db_session.commit()
await db_session.refresh(settings)
# Verify updates
assert settings.id == initial_id # Same record
assert settings.risk_profile == RiskProfile.AGGRESSIVE
assert settings.risk_score == Decimal("8.5")
assert settings.max_position_pct == Decimal("25.0")
assert settings.alert_preferences["sms"]["phone"] == "+1234567890"
except ImportError:
pytest.skip("Settings model not yet implemented (TDD RED phase)")
@pytest.mark.asyncio
async def test_settings_isolation_between_users(self, db_session, test_user, second_user):
"""Should maintain separate settings for different users."""
try:
from tradingagents.api.models.settings import Settings, RiskProfile
# Create settings for first user
settings1 = Settings(
user_id=test_user.id,
risk_profile=RiskProfile.CONSERVATIVE,
risk_score=Decimal("2.0"),
max_position_pct=Decimal("5.0"),
)
db_session.add(settings1)
# Create settings for second user
settings2 = Settings(
user_id=second_user.id,
risk_profile=RiskProfile.AGGRESSIVE,
risk_score=Decimal("9.0"),
max_position_pct=Decimal("30.0"),
)
db_session.add(settings2)
await db_session.commit()
# Retrieve settings for first user
result1 = await db_session.execute(
select(Settings).where(Settings.user_id == test_user.id)
)
user1_settings = result1.scalar_one()
# Retrieve settings for second user
result2 = await db_session.execute(
select(Settings).where(Settings.user_id == second_user.id)
)
user2_settings = result2.scalar_one()
# Verify isolation
assert user1_settings.id != user2_settings.id
assert user1_settings.risk_profile == RiskProfile.CONSERVATIVE
assert user2_settings.risk_profile == RiskProfile.AGGRESSIVE
assert user1_settings.max_position_pct == Decimal("5.0")
assert user2_settings.max_position_pct == Decimal("30.0")
except ImportError:
pytest.skip("Settings model not yet implemented (TDD RED phase)")
@pytest.mark.asyncio
async def test_complex_alert_preferences_workflow(self, db_session, test_user):
"""Should handle complex alert preferences updates."""
try:
from tradingagents.api.models.settings import Settings
# Start with email alerts only
settings = Settings(
user_id=test_user.id,
alert_preferences={
"email": {
"enabled": True,
"address": "user@example.com",
"alert_types": ["price_alert"]
}
},
)
db_session.add(settings)
await db_session.commit()
await db_session.refresh(settings)
# Add SMS alerts
settings.alert_preferences = {
"email": {
"enabled": True,
"address": "user@example.com",
"alert_types": ["price_alert", "portfolio_alert"]
},
"sms": {
"enabled": True,
"phone": "+1234567890",
"alert_types": ["critical_alert"],
"rate_limit": {"max_per_hour": 5}
}
}
await db_session.commit()
await db_session.refresh(settings)
# Verify complex structure
assert "email" in settings.alert_preferences
assert "sms" in settings.alert_preferences
assert len(settings.alert_preferences["email"]["alert_types"]) == 2
assert settings.alert_preferences["sms"]["rate_limit"]["max_per_hour"] == 5
# Disable email, keep SMS - must reassign entire dict for SQLAlchemy to track change
updated_prefs = dict(settings.alert_preferences)
updated_prefs["email"] = dict(updated_prefs["email"])
updated_prefs["email"]["enabled"] = False
settings.alert_preferences = updated_prefs
await db_session.commit()
await db_session.refresh(settings)
assert settings.alert_preferences["email"]["enabled"] is False
assert settings.alert_preferences["sms"]["enabled"] is True
except ImportError:
pytest.skip("Settings model not yet implemented (TDD RED phase)")
@pytest.mark.asyncio
async def test_query_settings_by_risk_profile(self, db_session, test_user, second_user):
"""Should query settings by risk profile."""
try:
from tradingagents.api.models.settings import Settings, RiskProfile
from tradingagents.api.models import User
from tradingagents.api.services.auth_service import hash_password
# Create third user
user3 = User(
username="user3",
email="user3@example.com",
hashed_password=hash_password("password123"),
)
db_session.add(user3)
await db_session.commit()
await db_session.refresh(user3)
# Create settings with different risk profiles
settings1 = Settings(user_id=test_user.id, risk_profile=RiskProfile.CONSERVATIVE)
settings2 = Settings(user_id=second_user.id, risk_profile=RiskProfile.AGGRESSIVE)
settings3 = Settings(user_id=user3.id, risk_profile=RiskProfile.CONSERVATIVE)
db_session.add_all([settings1, settings2, settings3])
await db_session.commit()
# Query conservative profiles
result = await db_session.execute(
select(Settings).where(Settings.risk_profile == RiskProfile.CONSERVATIVE)
)
conservative_settings = result.scalars().all()
# Verify
assert len(conservative_settings) == 2
assert all(s.risk_profile == RiskProfile.CONSERVATIVE for s in conservative_settings)
# Query aggressive profiles
result = await db_session.execute(
select(Settings).where(Settings.risk_profile == RiskProfile.AGGRESSIVE)
)
aggressive_settings = result.scalars().all()
assert len(aggressive_settings) == 1
assert aggressive_settings[0].user_id == second_user.id
except ImportError:
pytest.skip("Settings model not yet implemented (TDD RED phase)")
@pytest.mark.asyncio
async def test_settings_with_user_deletion(self, db_session):
"""Should handle settings cleanup when user is deleted."""
try:
from tradingagents.api.models.settings import Settings, RiskProfile
from tradingagents.api.models import User
from tradingagents.api.services.auth_service import hash_password
# Create user
user = User(
username="tempuser",
email="temp@example.com",
hashed_password=hash_password("password123"),
)
db_session.add(user)
await db_session.commit()
await db_session.refresh(user)
# Create settings
settings = Settings(
user_id=user.id,
risk_profile=RiskProfile.MODERATE,
)
db_session.add(settings)
await db_session.commit()
settings_id = settings.id
user_id = user.id
# Delete user
await db_session.delete(user)
await db_session.commit()
# Verify settings were cascade deleted
result = await db_session.execute(
select(Settings).where(Settings.id == settings_id)
)
deleted_settings = result.scalar_one_or_none()
assert deleted_settings is None
# Verify user is also deleted
result = await db_session.execute(
select(User).where(User.id == user_id)
)
deleted_user = result.scalar_one_or_none()
assert deleted_user is None
except ImportError:
pytest.skip("Settings model not yet implemented (TDD RED phase)")

View File

@ -1,11 +1,8 @@
"""
Shared pytest fixtures for unit API tests.
This module imports fixtures from the main API conftest
to make them available to unit tests.
This module may contain unit-specific overrides or fixtures.
Common fixtures are imported from tests.api.conftest via the root conftest.py.
"""
import pytest
# Import all fixtures from main API conftest
pytest_plugins = ["tests.api.conftest"]

File diff suppressed because it is too large Load Diff

View File

@ -4,5 +4,6 @@ from tradingagents.api.models.base import Base
from tradingagents.api.models.user import User
from tradingagents.api.models.strategy import Strategy
from tradingagents.api.models.portfolio import Portfolio, PortfolioType
from tradingagents.api.models.settings import Settings, RiskProfile
__all__ = ["Base", "User", "Strategy", "Portfolio", "PortfolioType"]
__all__ = ["Base", "User", "Strategy", "Portfolio", "PortfolioType", "Settings", "RiskProfile"]

View File

@ -0,0 +1,288 @@
"""Settings model for user risk profiles and alert preferences.
This module defines the Settings model for managing user trading preferences,
risk profiles, and alert configurations. Each user has exactly one Settings
record (one-to-one relationship).
Model Fields:
- id: Primary key
- user_id: Foreign key to users table (unique, one-to-one)
- risk_profile: User's risk tolerance (CONSERVATIVE, MODERATE, AGGRESSIVE)
- risk_score: Numeric risk score from 0 (very conservative) to 10 (very aggressive)
- max_position_pct: Maximum percentage of portfolio for single position (0-100)
- max_portfolio_risk_pct: Maximum portfolio-wide risk percentage (0-100)
- investment_horizon_years: Investment time horizon in years (>= 0)
- alert_preferences: JSON configuration for email/SMS/push notifications
- created_at, updated_at: Automatic timestamps
Relationships:
- user: One-to-one relationship with User model
- Cascade delete when user is deleted
Constraints:
- Unique constraint on user_id (one settings per user)
- Check constraint: risk_score >= 0 AND risk_score <= 10
- Check constraint: max_position_pct >= 0 AND max_position_pct <= 100
- Check constraint: max_portfolio_risk_pct >= 0 AND max_portfolio_risk_pct <= 100
- Check constraint: investment_horizon_years >= 0
Follows SQLAlchemy 2.0 patterns with Mapped[] and mapped_column().
"""
from enum import Enum as PyEnum
from typing import Optional, Dict, Any
from decimal import Decimal
from sqlalchemy import (
String,
Integer,
Numeric,
ForeignKey,
Index,
UniqueConstraint,
CheckConstraint,
Enum,
event,
JSON,
)
from sqlalchemy.orm import Mapped, mapped_column, relationship, validates, Session
from tradingagents.api.models.base import Base, TimestampMixin
class RiskProfile(str, PyEnum):
"""Enum for user risk tolerance profiles.
CONSERVATIVE: Low risk tolerance, focus on capital preservation
MODERATE: Balanced risk/reward approach (default)
AGGRESSIVE: High risk tolerance, focus on growth
"""
CONSERVATIVE = "CONSERVATIVE"
MODERATE = "MODERATE"
AGGRESSIVE = "AGGRESSIVE"
class Settings(Base, TimestampMixin):
"""Settings model for user preferences and risk management.
A settings record configures a user's trading preferences including
risk tolerance, position sizing limits, and alert configurations.
Each user has exactly one settings record (one-to-one relationship).
Attributes:
id: Primary key, auto-increment
user_id: Foreign key to users.id (cascade delete, unique)
risk_profile: Risk tolerance profile (CONSERVATIVE, MODERATE, AGGRESSIVE)
risk_score: Numeric risk score 0-10 (Decimal 5,2)
max_position_pct: Max % of portfolio for single position (Decimal 5,2)
max_portfolio_risk_pct: Max portfolio-wide risk % (Decimal 5,2)
investment_horizon_years: Investment time horizon in years
alert_preferences: JSON config for notifications (email, SMS, push)
user: Relationship to User model
created_at: Timestamp when created (auto)
updated_at: Timestamp when last updated (auto)
Constraints:
- user_id must be unique (one-to-one with User)
- risk_score must be between 0 and 10 (inclusive)
- max_position_pct must be between 0 and 100 (inclusive)
- max_portfolio_risk_pct must be between 0 and 100 (inclusive)
- investment_horizon_years must be >= 0
Example:
>>> from decimal import Decimal
>>> settings = Settings(
... user_id=1,
... risk_profile=RiskProfile.MODERATE,
... risk_score=Decimal("5.0"),
... max_position_pct=Decimal("10.0"),
... max_portfolio_risk_pct=Decimal("2.0"),
... investment_horizon_years=5,
... alert_preferences={
... "email": {
... "enabled": True,
... "address": "user@example.com",
... "alert_types": ["price_alert", "portfolio_alert"]
... }
... }
... )
>>> session.add(settings)
>>> await session.commit()
"""
__tablename__ = "settings"
# Primary key
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
# Foreign key to user (cascade delete, unique for one-to-one)
user_id: Mapped[int] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"),
nullable=False,
unique=True,
index=True,
comment="User who owns these settings (one-to-one)"
)
# Risk profile (enum)
risk_profile: Mapped[RiskProfile] = mapped_column(
Enum(RiskProfile, native_enum=False, length=20),
nullable=False,
default=RiskProfile.MODERATE,
comment="Risk tolerance: CONSERVATIVE, MODERATE, or AGGRESSIVE"
)
# Risk score (0-10 scale with 2 decimal places)
risk_score: Mapped[Decimal] = mapped_column(
Numeric(precision=5, scale=2),
nullable=False,
default=Decimal("5.0"),
comment="Numeric risk score from 0 (conservative) to 10 (aggressive)"
)
# Position sizing limits (percentages with 2 decimal places)
max_position_pct: Mapped[Decimal] = mapped_column(
Numeric(precision=5, scale=2),
nullable=False,
default=Decimal("10.0"),
comment="Maximum percentage of portfolio for single position (0-100)"
)
max_portfolio_risk_pct: Mapped[Decimal] = mapped_column(
Numeric(precision=5, scale=2),
nullable=False,
default=Decimal("2.0"),
comment="Maximum portfolio-wide risk percentage (0-100)"
)
# Investment horizon
investment_horizon_years: Mapped[int] = mapped_column(
Integer,
nullable=False,
default=5,
comment="Investment time horizon in years"
)
# Alert preferences (JSON)
alert_preferences: Mapped[Dict[str, Any]] = mapped_column(
JSON,
nullable=False,
default=dict,
comment="JSON configuration for email/SMS/push notifications"
)
# Relationships
user: Mapped["User"] = relationship(
"User",
back_populates="settings"
)
# Table-level constraints and indexes
__table_args__ = (
# Unique constraint: one settings per user
UniqueConstraint(
"user_id",
name="uq_settings_user_id"
),
# Check constraints: valid numeric ranges
CheckConstraint(
"risk_score >= 0 AND risk_score <= 10",
name="ck_settings_risk_score_range"
),
CheckConstraint(
"max_position_pct >= 0 AND max_position_pct <= 100",
name="ck_settings_max_position_pct_range"
),
CheckConstraint(
"max_portfolio_risk_pct >= 0 AND max_portfolio_risk_pct <= 100",
name="ck_settings_max_portfolio_risk_pct_range"
),
CheckConstraint(
"investment_horizon_years >= 0",
name="ck_settings_investment_horizon_positive"
),
# Note: Index on user_id is auto-created by unique=True parameter above
)
@validates("risk_profile")
def validate_risk_profile(self, key: str, value) -> RiskProfile:
"""Validate and convert risk profile to RiskProfile enum.
Args:
key: Field name (risk_profile)
value: Risk profile value (str or RiskProfile)
Returns:
RiskProfile enum value
Raises:
ValueError: If value is not a valid risk profile
"""
# If already a RiskProfile, return it
if isinstance(value, RiskProfile):
return value
# Try to convert string to RiskProfile
if isinstance(value, str):
try:
return RiskProfile[value.upper()]
except KeyError:
raise ValueError(
f"Invalid risk profile '{value}'. "
f"Must be one of: {', '.join([p.value for p in RiskProfile])}"
)
# Invalid type
raise ValueError(
f"Risk profile must be string or RiskProfile enum, got {type(value)}"
)
def __repr__(self) -> str:
"""String representation of Settings.
Returns:
String showing settings ID, user ID, and risk profile
"""
return (
f"<Settings(id={self.id}, "
f"user_id={self.user_id}, "
f"risk_profile={self.risk_profile.value}, "
f"risk_score={self.risk_score})>"
)
# Event listener for before_flush validation
# This ensures business rules are validated before database commit
@event.listens_for(Session, "before_flush")
def validate_settings_before_flush(session, flush_context, instances):
"""Validate Settings objects before flushing to database.
This event listener checks business rules that cannot be enforced
by database constraints (data normalization, complex business logic).
Database-enforced constraints (CheckConstraints) will raise IntegrityError
from the database itself.
Args:
session: SQLAlchemy session
flush_context: Flush context
instances: Instances being flushed
Raises:
ValueError: If validation fails for business logic violations
"""
for obj in session.new | session.dirty:
if isinstance(obj, Settings):
# Ensure alert_preferences is never None (should default to empty dict)
if obj.alert_preferences is None:
obj.alert_preferences = {}
# Note: Numeric range validations (risk_score, max_position_pct, etc.)
# are handled by database CheckConstraints and will raise IntegrityError
# if violated. We don't duplicate those checks here.
# Note: Nested JSON mutations (e.g., modifying settings.alert_preferences["key"]["nested"] = value)
# are not automatically tracked by SQLAlchemy. Users should either:
# 1. Reassign the entire dict: settings.alert_preferences = {...}
# 2. Use flag_modified(settings, "alert_preferences") explicitly
# 3. Use a custom MutableDict implementation for nested tracking

View File

@ -24,6 +24,7 @@ class User(Base, TimestampMixin):
is_verified: Whether user email is verified
strategies: Related Strategy objects owned by this user
portfolios: Related Portfolio objects owned by this user
settings: Related Settings object for this user (one-to-one)
"""
__tablename__ = "users"
@ -80,5 +81,13 @@ class User(Base, TimestampMixin):
cascade="all, delete-orphan"
)
# Relationship to settings (Issue #5: DB-4) - one-to-one
settings: Mapped[Optional["Settings"]] = relationship(
"Settings",
back_populates="user",
cascade="all, delete-orphan",
uselist=False
)
def __repr__(self) -> str:
return f"<User(id={self.id}, username='{self.username}', email='{self.email}')>"