Michael Chihlas
02fc47c832
Shared helper used by invite, accept-invite, and role-change endpoints (integrated in T8). Counts active users by role against role-specific seat limit on subscription (engineer → seat_limit, l1_tech → l1_seat_limit). None limit = unlimited. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
64 lines
1.8 KiB
Python
64 lines
1.8 KiB
Python
from typing import Literal
|
|
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.account import Account
|
|
from app.models.subscription import Subscription
|
|
from app.models.user import User
|
|
from app.schemas.seat_enforcement import SeatCheckResult
|
|
|
|
|
|
Role = Literal['engineer', 'l1_tech']
|
|
|
|
|
|
def _limit_for_role(subscription: Subscription, role: Role) -> int | None:
|
|
if role == 'engineer':
|
|
return subscription.seat_limit
|
|
if role == 'l1_tech':
|
|
return subscription.l1_seat_limit
|
|
raise ValueError(f"Unknown role: {role}")
|
|
|
|
|
|
async def check_seat_available(
|
|
account: Account,
|
|
subscription: Subscription,
|
|
role: Role,
|
|
db: AsyncSession,
|
|
) -> SeatCheckResult:
|
|
"""
|
|
Count active users with the given role in the account, compare against
|
|
the role-specific seat limit on the subscription. Returns availability.
|
|
|
|
None limit = unlimited (returns available=True).
|
|
"""
|
|
limit = _limit_for_role(subscription, role)
|
|
|
|
stmt = (
|
|
select(func.count(User.id))
|
|
.where(User.account_id == account.id)
|
|
.where(User.account_role == role)
|
|
.where(User.is_active.is_(True))
|
|
)
|
|
current = (await db.execute(stmt)).scalar_one()
|
|
|
|
if limit is None:
|
|
return SeatCheckResult(available=True, current=current, limit=None, role=role)
|
|
return SeatCheckResult(
|
|
available=current < limit,
|
|
current=current,
|
|
limit=limit,
|
|
role=role,
|
|
)
|
|
|
|
|
|
async def get_seat_usage(
|
|
account: Account,
|
|
subscription: Subscription,
|
|
db: AsyncSession,
|
|
) -> tuple[SeatCheckResult, SeatCheckResult]:
|
|
"""Return (engineer, l1_tech) seat-usage tuple for the seat-counter widget."""
|
|
eng = await check_seat_available(account, subscription, 'engineer', db)
|
|
l1 = await check_seat_available(account, subscription, 'l1_tech', db)
|
|
return eng, l1
|