feat: add account-based subscription model with migrations

Transition from team-based to account-based multi-tenancy (Free/Pro/Team).
Migrations 016-020 create accounts, subscriptions, plan_limits, and
account_invites tables, then migrate existing users and content FKs.

New models: Account, Subscription, PlanLimits, AccountInvite.
Updated models add account_id alongside existing team_id (coexistence
for safe two-PR deployment). Permissions and deps refactored for
account_role instead of is_team_admin.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
chihlasm
2026-02-07 02:38:47 -05:00
parent fb84bd8144
commit 4ccb93ee31
22 changed files with 933 additions and 47 deletions

View File

@@ -1,4 +1,4 @@
from typing import Annotated
from typing import Annotated, Optional
from uuid import UUID
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
@@ -8,6 +8,7 @@ from sqlalchemy import select
from app.core.database import get_db
from app.core.security import decode_token
from app.models.user import User
from app.models.plan_limits import PlanLimits
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")
@@ -90,14 +91,35 @@ async def require_admin(
async def require_engineer_or_admin(
current_user: Annotated[User, Depends(get_current_active_user)]
) -> User:
"""Require engineer, team admin, or super admin role (blocks viewers)."""
"""Require engineer, account owner, or super admin role (blocks viewers)."""
if current_user.is_super_admin:
return current_user
if current_user.is_team_admin and current_user.team_id is not None:
if current_user.account_role in ("owner", "engineer"):
return current_user
if current_user.role not in ("engineer",):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Engineer or admin access required"
)
return current_user
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Engineer or admin access required"
)
async def require_account_owner(
current_user: Annotated[User, Depends(get_current_active_user)]
) -> User:
"""Require account owner or super admin access."""
if current_user.is_super_admin:
return current_user
if current_user.account_role == "owner":
return current_user
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Account owner access required"
)
async def get_plan_limits_for_user(
current_user: Annotated[User, Depends(get_current_active_user)],
db: Annotated[AsyncSession, Depends(get_db)],
) -> Optional[PlanLimits]:
"""Get plan limits for the current user's account."""
from app.core.subscriptions import get_user_plan_limits
return await get_user_plan_limits(current_user.account_id, db)