File size: 3,312 Bytes
939c0c0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 | """
Seed database script to bootstrap the system with initial demo data.
Supports both PostgreSQL (local) and SQLite (Hugging Face / local) databases.
"""
from __future__ import annotations
import argparse
import asyncio
import sys
import os
from sqlalchemy import select
# Adjust path to import app modules
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "backend")))
from app.core.config import settings
from app.core.database import AsyncSessionLocal, engine
from app.core.security import UserRole, hash_password
from app.models.tenant import Tenant
from app.models.user import User
async def seed_database(hf_mode: bool = False):
"""Seed the database with a default tenant and administrator account."""
print("Database seeding started...")
# Override settings if hf_mode is specified
if hf_mode:
print("HF Mode enabled: seeding SQLite database...")
async with AsyncSessionLocal() as session:
try:
# 1. Ensure default tenant exists
tenant_slug = "acme"
result = await session.execute(select(Tenant).where(Tenant.slug == tenant_slug))
tenant = result.scalar_one_or_none()
if not tenant:
print("Creating default tenant 'Acme Corp'...")
tenant = Tenant(
name="Acme Corp",
slug=tenant_slug,
description="Default demo organization workspace.",
is_active=True,
plan="free"
)
session.add(tenant)
await session.flush()
else:
print(f"Default tenant 'Acme Corp' already exists (ID: {tenant.id})")
# 2. Ensure default admin user exists
admin_email = settings.FIRST_ADMIN_EMAIL or "admin@company.com"
result = await session.execute(select(User).where(User.email == admin_email))
admin_user = result.scalar_one_or_none()
if not admin_user:
print(f"Creating default admin user '{admin_email}'...")
admin_password = settings.FIRST_ADMIN_PASSWORD or "Admin@123!"
admin_user = User(
email=admin_email,
hashed_password=hash_password(admin_password),
full_name="System Admin",
role=UserRole.ADMIN.value,
tenant_id=tenant.id,
is_active=True,
is_verified=True
)
session.add(admin_user)
else:
print(f"Default admin user '{admin_email}' already exists.")
await session.commit()
print("Database seeding completed successfully! 🎉")
except Exception as e:
print(f"Error seeding database: {e}")
await session.rollback()
raise
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Seed database for Enterprise AI Copilot")
parser.add_argument("--hf-mode", action="store_true", help="Run in Hugging Face SQLite mode")
args = parser.parse_args()
# Run async function
asyncio.run(seed_database(hf_mode=args.hf_mode))
|