93 lines
3.2 KiB
Python
93 lines
3.2 KiB
Python
import logging
|
|
import os
|
|
from pathlib import Path
|
|
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.config import settings
|
|
from app.db.session import Base, engine
|
|
from app.models import Category, Item, User # Will be implemented later
|
|
|
|
logging.basicConfig(level=logging.INFO)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def ensure_db_dir_exists() -> None:
|
|
"""
|
|
Ensure the database directory exists.
|
|
"""
|
|
try:
|
|
settings.DB_DIR.mkdir(parents=True, exist_ok=True)
|
|
logger.info(f"Database directory ensured at {settings.DB_DIR}")
|
|
except PermissionError:
|
|
logger.error(f"Permission denied when creating directory {settings.DB_DIR}")
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Error creating database directory: {str(e)}")
|
|
raise
|
|
|
|
|
|
def init_db() -> None:
|
|
"""
|
|
Initialize the database.
|
|
Creates all tables defined in the models.
|
|
"""
|
|
try:
|
|
# Ensure the database directory exists
|
|
ensure_db_dir_exists()
|
|
|
|
# Check if we can access the database file path
|
|
db_file = Path(settings.SQLALCHEMY_DATABASE_URL.replace("sqlite:///", ""))
|
|
db_dir = db_file.parent
|
|
|
|
if not db_dir.exists():
|
|
logger.warning(f"Database directory does not exist: {db_dir}")
|
|
db_dir.mkdir(parents=True, exist_ok=True)
|
|
logger.info(f"Created database directory: {db_dir}")
|
|
|
|
# Create tables
|
|
logger.info("Creating database tables")
|
|
Base.metadata.create_all(bind=engine)
|
|
logger.info("Database tables created")
|
|
|
|
except SQLAlchemyError as e:
|
|
logger.error(f"SQLAlchemy error during database initialization: {str(e)}")
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Unexpected error during database initialization: {str(e)}")
|
|
raise
|
|
|
|
|
|
def create_first_superuser(db: Session) -> None:
|
|
"""
|
|
Create the first superuser if it doesn't exist.
|
|
This can be called after init_db() to ensure there's at least one admin user.
|
|
"""
|
|
from app.core.security import get_password_hash
|
|
from app.crud.user import user
|
|
from app.schemas.user import UserCreate
|
|
|
|
# Check if there are any users in the database
|
|
first_user = db.query(User).first()
|
|
if not first_user:
|
|
logger.info("No users found. Creating initial superuser...")
|
|
try:
|
|
# Get superuser credentials from environment variables or use defaults
|
|
admin_email = os.getenv("FIRST_SUPERUSER_EMAIL", "admin@example.com")
|
|
admin_password = os.getenv("FIRST_SUPERUSER_PASSWORD", "adminpassword")
|
|
|
|
# Create user
|
|
user_in = UserCreate(
|
|
email=admin_email,
|
|
password=admin_password,
|
|
is_superuser=True,
|
|
full_name="Admin"
|
|
)
|
|
user.create(db, obj_in=user_in)
|
|
logger.info(f"Initial superuser created with email: {admin_email}")
|
|
except Exception as e:
|
|
logger.error(f"Error creating initial superuser: {str(e)}")
|
|
raise
|
|
else:
|
|
logger.info("Users exist in the database. Skipping superuser creation.") |