"""Seed script: Ensure default admin users exist (SSO-only, Azure AD). Run once after deployment to guarantee these accounts have admin role. If the user already exists (e.g. created via SSO with viewer role), their role is upgraded to admin. If they don't exist yet, they are pre-created so that SSO login will find the existing admin record rather than auto-provisioning them as viewer. """ import asyncio import sys from pathlib import Path _seed_dir = Path(__file__).resolve().parent _backend_dir = _seed_dir.parent / "backend" if _backend_dir.is_dir(): sys.path.insert(0, str(_backend_dir)) else: sys.path.insert(0, str(_seed_dir.parent)) from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from app.config import settings from app.models.user import User, UserRole, UserStatus # These accounts will always have admin role. # They log in via Azure AD SSO — no password needed. DEFAULT_ADMINS = [ {"email": "daveporter@oliver.agency", "name": "Dave Porter"}, {"email": "vadymsamoilenko@oliver.agency", "name": "Vadym Samoilenko"}, ] async def seed_default_admins() -> None: engine = create_async_engine(settings.DATABASE_URL, pool_pre_ping=True) factory = async_sessionmaker(bind=engine, class_=AsyncSession, expire_on_commit=False) async with factory() as db: for admin_data in DEFAULT_ADMINS: result = await db.execute( select(User).where(User.email == admin_data["email"]) ) user = result.scalar_one_or_none() if user: if user.role != UserRole.admin: user.role = UserRole.admin print(f"Upgraded '{admin_data['email']}' to admin (was {user.role.value})") else: print(f"'{admin_data['email']}' is already admin — no change") else: user = User( email=admin_data["email"], name=admin_data["name"], password_hash=None, role=UserRole.admin, status=UserStatus.active, auth_provider="azure_ad", ) db.add(user) print(f"Created admin '{admin_data['email']}'") await db.commit() await engine.dispose() print("\nDone. These accounts will have admin role on next SSO login.") if __name__ == "__main__": asyncio.run(seed_default_admins())