Add default categories to init_db.py

This commit is contained in:
2025-03-16 17:36:16 +01:00
parent 1dcad21bed
commit 16b6269d05

View File

@@ -1,10 +1,14 @@
# app/core/init_db.py
import logging
from typing import Optional
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__)
@@ -13,40 +17,138 @@ 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")
return None
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=settings.FIRST_SUPERUSER_EMAIL
db, email=superuser_email
)
if existing_user:
logger.info("Superuser already exists, skipping creation")
return None
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
)
# Create superuser if doesn't exist
user_in = UserCreate(
email=settings.FIRST_SUPERUSER_EMAIL,
password=settings.FIRST_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)
user = user_crud.create(db, obj_in=user_in)
logger.info(f"Created first superuser: {user.email}")
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:
init_db(session)
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()