Spaces:
Runtime error
Runtime error
| import sys | |
| import os | |
| import asyncio | |
| from sqlalchemy import select | |
| # Add parent directory to path | |
| current_dir = os.path.dirname(os.path.abspath(__file__)) | |
| parent_dir = os.path.dirname(current_dir) | |
| sys.path.append(parent_dir) | |
| from app.core.database import AsyncSessionLocal | |
| from app.models.tenant import Tenant | |
| from app.models.user import User | |
| async def seed_tenant(): | |
| async with AsyncSessionLocal() as db: | |
| print("🌱 Seeding Tenant...") | |
| # Check if tenant exists | |
| result = await db.execute(select(Tenant).where(Tenant.name == "Default Health System")) | |
| tenant = result.scalars().first() | |
| if not tenant: | |
| tenant = Tenant( | |
| name="Default Health System", | |
| industry="healthcare", | |
| settings={"strict_mode": True, "compliance": "HIPAA"} | |
| ) | |
| db.add(tenant) | |
| await db.commit() | |
| await db.refresh(tenant) | |
| print(f"✅ Created Tenant: {tenant.name} (ID: {tenant.id})") | |
| else: | |
| print(f"ℹ️ Tenant already exists: {tenant.name} (ID: {tenant.id})") | |
| # Check if Admin User exists | |
| result = await db.execute(select(User).where(User.email == "admin@example.com")) | |
| user = result.scalars().first() | |
| if not user: | |
| # Create dummy admin | |
| user = User( | |
| email="admin@example.com", | |
| name="System Admin", | |
| hashed_password="hashed_secret_password", # Dummy | |
| tenant_id=tenant.id, | |
| is_active=True | |
| ) | |
| db.add(user) | |
| await db.commit() | |
| print(f"✅ Created Admin User linked to Tenant {tenant.id}") | |
| else: | |
| # Link to tenant if not linked | |
| if not user.tenant_id: | |
| user.tenant_id = tenant.id | |
| await db.commit() | |
| print(f"✅ Linked existing Admin User to Tenant {tenant.id}") | |
| else: | |
| print(f"ℹ️ Admin User already exists and linked") | |
| if __name__ == "__main__": | |
| asyncio.run(seed_tenant()) | |