# app/core/init_db.py import logging from typing import Optional, List, Dict from uuid import UUID from sqlalchemy.orm import Session from app.core.config import settings from app.crud.user import user_crud as user_crud from app.crud.event import event_crud from app.crud.gift import gift_category_crud, event_gift_category_crud from app.schemas.users import UserCreate from app.schemas.gifts import GiftCategoryCreate, EventGiftCategoryCreate from app.core.database import engine logger = logging.getLogger(__name__) def init_db(db: Session) -> Optional[UserCreate]: """ Initialize database with first superuser if settings are configured and user doesn't exist. """ # Use default values if not set in environment variables superuser_email = settings.FIRST_SUPERUSER_EMAIL or "admin@example.com" superuser_password = settings.FIRST_SUPERUSER_PASSWORD or "admin123" if not settings.FIRST_SUPERUSER_EMAIL or not settings.FIRST_SUPERUSER_PASSWORD: logger.warning("First superuser credentials not configured in settings, using defaults") try: # Check if superuser already exists existing_user = user_crud.get_by_email( db, email=superuser_email ) if existing_user: logger.info("Superuser already exists, skipping creation") user = existing_user else: # Create superuser if doesn't exist user_in = UserCreate( email=superuser_email, password=superuser_password, first_name="Admin", # Required by your schema last_name="User", # Required by your schema is_superuser=True ) user = user_crud.create(db, obj_in=user_in) logger.info(f"Created first superuser: {user.email}") # Initialize default gift categories init_default_categories(db, user.id) return user except Exception as e: logger.error(f"Error creating superuser: {e}") raise def init_default_categories(db: Session, user_id: UUID) -> Dict[str, UUID]: """ Initialize default gift categories and associate them with all existing events. Args: db: Database session user_id: ID of the user creating the categories Returns: Dictionary mapping category names to their UUIDs """ # Default categories to create default_categories = [ {"name": "misc", "description": "Miscellaneous items", "icon": "gift", "color": "#6c757d"}, {"name": "toys", "description": "Toys and games", "icon": "toy", "color": "#28a745"}, {"name": "tech", "description": "Technology and gadgets", "icon": "devices", "color": "#007bff"}, {"name": "clothes", "description": "Clothing and accessories", "icon": "clothing", "color": "#dc3545"} ] # Dictionary to store category IDs by name category_ids = {} # Create categories if they don't exist for idx, category_data in enumerate(default_categories): # Check if category already exists existing_category = db.query(gift_category_crud.model).filter( gift_category_crud.model.name == category_data["name"] ).first() if existing_category: logger.info(f"Category '{category_data['name']}' already exists, skipping creation") category = existing_category else: # Create category category_in = GiftCategoryCreate( **category_data, created_by=user_id ) category = gift_category_crud.create(db, obj_in=category_in) logger.info(f"Created default category: {category.name}") category_ids[category.name] = category.id # Get all existing events events = event_crud.get_multi(db) # Associate categories with events for event in events: for idx, (category_name, category_id) in enumerate(category_ids.items()): # Check if association already exists existing_assoc = event_gift_category_crud.get( db, event_id=event.id, category_id=category_id ) if existing_assoc: logger.info(f"Category '{category_name}' already associated with event {event.title}") continue # Create association assoc_in = EventGiftCategoryCreate( event_id=event.id, category_id=category_id, display_order=idx, is_visible=True ) event_gift_category_crud.create(db, obj_in=assoc_in) logger.info(f"Associated category '{category_name}' with event {event.title}") return category_ids if __name__ == "__main__": # Configure logging to show info logs logging.basicConfig(level=logging.INFO) with Session(engine) as session: try: user = init_db(session) if user: print(f"Initialized database with superuser: {user.email}") # Print the number of events in the database events = event_crud.get_multi(session) print(f"Found {len(events)} events in the database") # Print the categories in the database categories = gift_category_crud.get_multi(session) print(f"Found {len(categories)} gift categories in the database:") for category in categories: print(f" - {category.name}: {category.description}") else: print("Failed to initialize database") except Exception as e: print(f"Error initializing database: {e}") finally: session.close()