Coverage for app / db / init_db.py: 100%

1import logging 

2import asyncio 

3from sqlalchemy import select 

4from sqlalchemy.ext.asyncio import AsyncSession 

5 

6from app.models.user import User 

7from app.db.session import AsyncSessionLocal 

8 

9logger = logging.getLogger(__name__) 

10 

11async def init_db(db: AsyncSession) -> None: 

12 """ 

13 Initialize database with base/seed data. 

14 Tables are created via Alembic migrations, so we only handle data here. 

15 """ 

16 logger.info("Starting database initialization...") 

17 

18 # Define your default admin details 

19 admin_email = "admin@example.com" 

20 

21 result = await db.execute(select(User).where(User.email == admin_email)) 

22 user = result.scalar_one_or_none() 

23 

24 if not user: 

25 admin_user = User( 

26 email=admin_email, 

27 supabase_uid="system-admin-id", # Replace with actual Supabase UID if needed 

28 ) 

29 db.add(admin_user) 

30 await db.commit() 

31 logger.info(f"Admin user {admin_email} created.") 

32 else: 

33 logger.info("Admin user already exists. Skipping creation.") 

34 

35 logger.info("Database initialization complete.") 

36 

37 

38async def main(): 

39 """Wrapper to run the async seed function manually.""" 

40 # Ensure our logging is formatted when running standalone 

41 from app.core.logging import setup_logging 

42 setup_logging() 

43 

44 async with AsyncSessionLocal() as session: 

45 await init_db(session) 

46 

47if __name__ == "__main__": # pragma: no cover 

48 asyncio.run(main())