leocrm-bot
3ab4925783
- 10 models: tenants, users, user_tenants, roles, sessions, audit_log, deletion_log, notifications, password_reset_tokens, api_tokens - Session-based auth (Redis + PostgreSQL audit trail) - Multi-tenant with ORM-level filtering + PostgreSQL RLS (set_config) - RBAC with roles/permissions + field-level permissions - CSRF protection via Origin header validation - Auth rate limiting (Redis counters with TTL) - CORS with explicit origins (no wildcard) - Health endpoint (no auth required) - Notification service + audit log middleware - 29 tests, 26 ACs, all passing - Coverage: 62% (infrastructure modules pending coverage in later tasks)
97 lines
2.5 KiB
Python
97 lines
2.5 KiB
Python
"""Tenant management service."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from typing import Any
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.tenant import Tenant
|
|
from app.models.user import User, UserTenant
|
|
|
|
|
|
class TenantService:
|
|
"""Handles tenant CRUD and user-tenant membership."""
|
|
|
|
async def list_tenants_for_user(
|
|
self,
|
|
db: AsyncSession,
|
|
user_id: uuid.UUID,
|
|
) -> list[dict[str, Any]]:
|
|
"""List all tenants a user belongs to."""
|
|
q = (
|
|
select(Tenant, UserTenant.is_default)
|
|
.join(UserTenant, UserTenant.tenant_id == Tenant.id)
|
|
.where(UserTenant.user_id == user_id)
|
|
)
|
|
result = await db.execute(q)
|
|
rows = result.all()
|
|
return [
|
|
{
|
|
"id": str(t.id),
|
|
"name": t.name,
|
|
"slug": t.slug,
|
|
"is_default": is_default,
|
|
}
|
|
for t, is_default in rows
|
|
]
|
|
|
|
async def create_tenant(
|
|
self,
|
|
db: AsyncSession,
|
|
name: str,
|
|
slug: str,
|
|
) -> Tenant:
|
|
"""Create a new tenant."""
|
|
tenant = Tenant(name=name, slug=slug)
|
|
db.add(tenant)
|
|
await db.flush()
|
|
return tenant
|
|
|
|
async def get_tenant(self, db: AsyncSession, tenant_id: uuid.UUID) -> Tenant | None:
|
|
"""Get a tenant by ID."""
|
|
q = select(Tenant).where(Tenant.id == tenant_id)
|
|
result = await db.execute(q)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def list_tenant_users(
|
|
self,
|
|
db: AsyncSession,
|
|
tenant_id: uuid.UUID,
|
|
) -> list[dict[str, Any]]:
|
|
"""List users in a tenant."""
|
|
q = select(User).where(User.tenant_id == tenant_id)
|
|
result = await db.execute(q)
|
|
users = result.scalars().all()
|
|
return [
|
|
{
|
|
"id": str(u.id),
|
|
"email": u.email,
|
|
"name": u.name,
|
|
"role": u.role,
|
|
"is_active": u.is_active,
|
|
}
|
|
for u in users
|
|
]
|
|
|
|
async def assign_user_to_tenant(
|
|
self,
|
|
db: AsyncSession,
|
|
tenant_id: uuid.UUID,
|
|
user_id: uuid.UUID,
|
|
) -> UserTenant:
|
|
"""Assign a user to a tenant."""
|
|
ut = UserTenant(
|
|
user_id=user_id,
|
|
tenant_id=tenant_id,
|
|
is_default=False,
|
|
)
|
|
db.add(ut)
|
|
await db.flush()
|
|
return ut
|
|
|
|
|
|
tenant_service = TenantService()
|