94 lines
2.6 KiB
Python
94 lines
2.6 KiB
Python
"""Feature gate module for VoIdea tariff-based access control.
|
|
|
|
Usage:
|
|
features = await get_user_features(user, db)
|
|
if not features.get("has_voice"):
|
|
raise HTTPException(403, "Feature not available on your plan")
|
|
"""
|
|
|
|
from typing import Any
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.config import get_settings
|
|
from app.models.tariff import TariffPlan, UserSubscription
|
|
from app.models.user import User
|
|
|
|
settings = get_settings()
|
|
|
|
DEFAULT_FREE_FEATURES: dict[str, Any] = {
|
|
"limit_ideas": None, # None = unlimited
|
|
"limit_sessions_per_day": None,
|
|
"limit_agents": None,
|
|
"limit_team_members": None,
|
|
"limit_storage_mb": 100,
|
|
"has_voice": True,
|
|
"has_voice_commands": True,
|
|
"has_drive": True,
|
|
"has_advanced_analytics": True,
|
|
"has_api_access": True,
|
|
"has_priority_support": True,
|
|
"has_custom_branding": True,
|
|
"_promo": True,
|
|
}
|
|
|
|
|
|
async def get_user_features(
|
|
user: User, db: AsyncSession
|
|
) -> dict[str, Any]:
|
|
"""Get feature flags and limits for a user based on their tariff plan.
|
|
|
|
When tariffs_enabled=False (promo mode), returns all features as available.
|
|
"""
|
|
if not settings.tariffs_enabled:
|
|
return dict(DEFAULT_FREE_FEATURES)
|
|
|
|
result = await db.execute(
|
|
select(UserSubscription).where(UserSubscription.user_id == user.id)
|
|
)
|
|
sub = result.scalar_one_or_none()
|
|
|
|
if sub is None or sub.status != "active":
|
|
return _get_plan_features(db, settings.tariffs_free_code)
|
|
|
|
result = await db.execute(
|
|
select(TariffPlan).where(TariffPlan.id == sub.plan_id)
|
|
)
|
|
plan = result.scalar_one_or_none()
|
|
|
|
if plan is None or not plan.is_active:
|
|
return _get_plan_features(db, settings.tariffs_free_code)
|
|
|
|
return plan.features or dict(DEFAULT_FREE_FEATURES)
|
|
|
|
|
|
async def _get_plan_features(
|
|
db: AsyncSession, code: str
|
|
) -> dict[str, Any]:
|
|
"""Get features dict for a plan by code, falling back to defaults."""
|
|
result = await db.execute(
|
|
select(TariffPlan).where(
|
|
TariffPlan.code == code, TariffPlan.is_active.is_(True)
|
|
)
|
|
)
|
|
plan = result.scalar_one_or_none()
|
|
if plan and plan.features:
|
|
return plan.features
|
|
return dict(DEFAULT_FREE_FEATURES)
|
|
|
|
|
|
def check_feature(features: dict[str, Any], key: str) -> bool:
|
|
"""Check if a boolean feature is enabled."""
|
|
return bool(features.get(key, False))
|
|
|
|
|
|
def check_limit(
|
|
features: dict[str, Any], key: str, current: int = 0
|
|
) -> bool:
|
|
"""Check if a numeric limit is not exceeded (None = unlimited)."""
|
|
limit = features.get(key)
|
|
if limit is None:
|
|
return True
|
|
return current < int(limit)
|