diff --git a/media_manager/auth/users.py b/media_manager/auth/users.py index 45835bb..d62fffc 100644 --- a/media_manager/auth/users.py +++ b/media_manager/auth/users.py @@ -1,3 +1,4 @@ +import contextlib import logging import uuid from typing import Optional @@ -14,10 +15,11 @@ from fastapi_users.db import SQLAlchemyUserDatabase from httpx_oauth.clients.openid import OpenID from fastapi.responses import RedirectResponse, Response from starlette import status +from sqlalchemy import select, func import media_manager.notification.utils -from media_manager.auth.db import User, get_user_db -from media_manager.auth.schemas import UserUpdate +from media_manager.auth.db import User, get_user_db, get_async_session +from media_manager.auth.schemas import UserUpdate, UserCreate from media_manager.config import AllEncompassingConfig log = logging.getLogger(__name__) @@ -100,6 +102,68 @@ async def get_user_manager(user_db: SQLAlchemyUserDatabase = Depends(get_user_db yield UserManager(user_db) +get_async_session_context = contextlib.asynccontextmanager(get_async_session) +get_user_db_context = contextlib.asynccontextmanager(get_user_db) +get_user_manager_context = contextlib.asynccontextmanager(get_user_manager) + + +async def create_default_admin_user(): + """Create a default admin user if no users exist in the database""" + try: + async with get_async_session_context() as session: + async with get_user_db_context(session) as user_db: + async with get_user_manager_context(user_db) as user_manager: + # Check if any users exist + stmt = select(func.count(User.id)) + result = await session.execute(stmt) + user_count = result.scalar() + config = AllEncompassingConfig() + if user_count == 0: + log.info( + "No users found in database. Creating default admin user..." + ) + + # Use the first admin email from config, or default + admin_email = ( + config.auth.admin_emails[0] + if config.auth.admin_emails + else "admin@mediamanager.local" + ) + default_password = "admin" # Simple default password + + user_create = UserCreate( + email=admin_email, + password=default_password, + is_superuser=True, + is_verified=True, + ) + + user = await user_manager.create(user_create) + log.info("=" * 60) + log.info("DEFAULT ADMIN USER CREATED!") + log.info(f" Email: {admin_email}") + log.info(f" Password: {default_password}") + log.info(f" User ID: {user.id}") + log.info( + "IMPORTANT: Please change this password after first login!" + ) + log.info("=" * 60) + + else: + log.info( + f"Found {user_count} existing users. Skipping default user creation." + ) + except Exception as e: + log.error(f"Failed to create default admin user: {e}") + log.info( + "You can create an admin user manually by registering with an email from the admin_emails list in your config." + ) + + +async def get_user_manager(user_db: SQLAlchemyUserDatabase = Depends(get_user_db)): + yield UserManager(user_db) + + def get_jwt_strategy() -> JWTStrategy[models.UP, models.ID]: return JWTStrategy(secret=SECRET, lifetime_seconds=LIFETIME) diff --git a/media_manager/main.py b/media_manager/main.py index 6c8beed..adc4dd5 100644 --- a/media_manager/main.py +++ b/media_manager/main.py @@ -74,6 +74,7 @@ from media_manager.auth.users import ( # noqa: E402 fastapi_users, cookie_auth_backend, openid_cookie_auth_backend, + create_default_admin_user, ) from media_manager.exceptions import ( # noqa: E402 NotFoundError, @@ -154,56 +155,6 @@ async def lifespan(app: FastAPI): scheduler.shutdown() -async def create_default_admin_user(): - """Create a default admin user if no users exist in the database""" - try: - from media_manager.auth.db import get_user_db - from media_manager.auth.users import get_user_manager - from media_manager.auth.schemas import UserCreate - from media_manager.database import get_session - from sqlalchemy import select, func - from media_manager.auth.db import User - - async for session in get_session(): - async for user_db in get_user_db(session): - async for user_manager in get_user_manager(user_db): - # Check if any users exist - result = await session.execute(select(func.count(User.id))) - user_count = result.scalar() - - if user_count == 0: - log.info("No users found in database. Creating default admin user...") - - # Use the first admin email from config, or default - admin_email = config.auth.admin_emails[0] if config.auth.admin_emails else "admin@mediamanager.local" - default_password = "admin" # Simple default password - - user_create = UserCreate( - email=admin_email, - password=default_password, - is_superuser=True, - is_verified=True - ) - - user = await user_manager.create(user_create) - log.info("=" * 60) - log.info("✅ DEFAULT ADMIN USER CREATED!") - log.info(f" 📧 Email: {admin_email}") - log.info(f" 🔑 Password: {default_password}") - log.info(f" 👤 User ID: {user.id}") - log.info(" ⚠️ IMPORTANT: Please change this password after first login!") - log.info("=" * 60) - - else: - log.info(f"Found {user_count} existing users. Skipping default user creation.") - break - break - break - except Exception as e: - log.error(f"Failed to create default admin user: {e}") - log.info("You can create an admin user manually by registering with an email from the admin_emails list in your config.") - - BASE_PATH = os.getenv("BASE_PATH", "") FRONTEND_FILES_DIR = os.getenv("FRONTEND_FILES_DIR") @@ -319,18 +270,22 @@ app.mount("/web", StaticFiles(directory=FRONTEND_FILES_DIR, html=True), name="fr # Redirects to frontend # ---------------------------- + @app.get("/") async def root(): return RedirectResponse(url="/web/") + @app.get("/dashboard") async def dashboard(): return RedirectResponse(url="/web/") + @app.get("/login") async def login(): return RedirectResponse(url="/web/") + # ---------------------------- # Custom Exception Handlers # ----------------------------