Merge pull request #181 from lbedner/auth-docs

Update Auth Docs

Author: lbedner

aegis/templates/cookiecutter-aegis-project/hooks/post_gen_project.py

@@ -292,6 +292,8 @@ def main():
         remove_file("app/models/user.py")
         remove_dir("app/services/auth")
         remove_file("app/core/security.py")
+        # Remove auth CLI
+        remove_file("app/cli/auth.py")
         # Remove auth-related tests if they exist
         remove_file("tests/api/test_auth_endpoints.py")
         remove_file("tests/services/test_auth_service.py")

aegis/templates/cookiecutter-aegis-project/{{cookiecutter.project_slug}}/app/cli/auth.py.j2

@@ -4,6 +4,7 @@ Authentication CLI commands.
 Command-line interface for auth service management tasks.
 """
 
+import asyncio
 import secrets
 import string
 from typing import TYPE_CHECKING
@@ -13,7 +14,7 @@ if TYPE_CHECKING:
 
 import typer
 
-from app.core.db import db_session
+from app.core.db import get_async_session
 from app.models.user import UserCreate
 from app.services.auth.user_service import UserService
 
@@ -26,12 +27,14 @@ def generate_password(length: int = 12) -> str:
     return "".join(secrets.choice(alphabet) for _ in range(length))
 
 
-def find_next_available_email(
+async def find_next_available_email(
     user_service: UserService, prefix: str = "test", domain: str = "example.com"
 ) -> str:
     """Find the next available email with auto-increment."""
     # Get existing emails with this prefix
-    existing_emails = user_service.find_existing_emails_with_prefix(prefix, domain)
+    existing_emails = await user_service.find_existing_emails_with_prefix(
+        prefix, domain
+    )
 
     if not existing_emails:
         # No existing emails, start with prefix@domain
@@ -82,7 +85,17 @@ def create_test_user(
     ),
 ) -> None:
     """Create a test user for development and testing."""
+    asyncio.run(_create_test_user(email, password, full_name, prefix, domain))
 
+
+async def _create_test_user(
+    email: str | None,
+    password: str | None,
+    full_name: str | None,
+    prefix: str,
+    domain: str,
+) -> None:
+    """Async implementation of create_test_user."""
     # Generate password if not provided
     if password is None:
         password = generate_password()
@@ -91,16 +104,16 @@ def create_test_user(
         generated_password = False
 
     try:
-        with db_session() as session:
+        async with get_async_session() as session:
             user_service = UserService(session)
 
             # Auto-generate email if not provided
             if email is None:
-                email = find_next_available_email(user_service, prefix, domain)
+                email = await find_next_available_email(user_service, prefix, domain)
                 typer.echo(f"📧 Auto-generated email: {email} (next in sequence)")
 
             # Check if user already exists
-            existing_user = user_service.get_user_by_email(email)
+            existing_user = await user_service.get_user_by_email(email)
             if existing_user:
                 typer.echo(f"❌ User with email '{email}' already exists", err=True)
                 raise typer.Exit(1)
@@ -113,7 +126,7 @@ def create_test_user(
             )
 
             # Create the user
-            user = user_service.create_user(user_data)
+            user = await user_service.create_user(user_data)
 
             # Display success message
             typer.echo("✅ Test user created successfully!")
@@ -145,7 +158,13 @@ def create_test_users(
     ),
 ) -> None:
     """Create multiple test users for development and testing."""
+    asyncio.run(_create_test_users(count, prefix, domain, password))
+
 
+async def _create_test_users(
+    count: int, prefix: str, domain: str, password: str | None
+) -> None:
+    """Async implementation of create_test_users."""
     # Generate password if not provided
     if password is None:
         password = generate_password()
@@ -158,7 +177,7 @@ def create_test_users(
         raise typer.Exit(1)
 
     try:
-        with db_session() as session:
+        async with get_async_session() as session:
             user_service = UserService(session)
             created_users = []
 
@@ -167,7 +186,7 @@ def create_test_users(
 
             for i in range(count):
                 # Find next available email
-                email = find_next_available_email(user_service, prefix, domain)
+                email = await find_next_available_email(user_service, prefix, domain)
 
                 # Create user data
                 user_data = UserCreate(
@@ -177,7 +196,7 @@ def create_test_users(
                 )
 
                 # Create the user
-                user = user_service.create_user(user_data)
+                user = await user_service.create_user(user_data)
                 created_users.append(user)
 
                 typer.echo(f"✅ Created: {user.email} (ID: {user.id})")
@@ -200,10 +219,15 @@ def create_test_users(
 @app.command()
 def list_users() -> None:
     """List all users in the system."""
+    asyncio.run(_list_users())
+
+
+async def _list_users() -> None:
+    """Async implementation of list_users."""
     try:
-        with db_session() as session:
+        async with get_async_session() as session:
             user_service = UserService(session)
-            users = user_service.list_users()
+            users = await user_service.list_users()
 
             if not users:
                 typer.echo("No users found.")

aegis/templates/cookiecutter-aegis-project/{{cookiecutter.project_slug}}/app/cli/{% if cookiecutter.include_auth == "yes" %}auth.py{% endif %}

@@ -4,7 +4,7 @@
 from app.core.security import create_access_token, verify_password
 from app.models.user import UserCreate, UserResponse
 from app.services.auth.auth_service import get_current_user_from_token
-from app.services.auth.user_service import AsyncUserService
+from app.services.auth.user_service import UserService
 from fastapi import APIRouter, Depends, HTTPException, status
 from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
 from sqlmodel.ext.asyncio.session import AsyncSession
@@ -17,7 +17,7 @@
 @router.post("/register", response_model=UserResponse)
 async def register(user_data: UserCreate, db: AsyncSession = Depends(get_async_db)):
     """Register a new user."""
-    user_service = AsyncUserService(db)
+    user_service = UserService(db)
 
     # Check if user already exists
     existing_user = await user_service.get_user_by_email(user_data.email)
@@ -37,7 +37,7 @@ async def login(
     db: AsyncSession = Depends(get_async_db),
 ):
     """Login and get access token."""
-    user_service = AsyncUserService(db)
+    user_service = UserService(db)
 
     # Get user by email (username field in OAuth2 form)
     user = await user_service.get_user_by_email(form_data.username)

aegis/templates/cookiecutter-aegis-project/{{cookiecutter.project_slug}}/app/components/backend/api/auth/router.py

@@ -5,7 +5,7 @@
 
 from app.core.security import verify_token
 from app.models.user import User
-from app.services.auth.user_service import AsyncUserService
+from app.services.auth.user_service import UserService
 
 
 async def get_current_user_from_token(token: str, db: AsyncSession) -> User:
@@ -27,7 +27,7 @@ async def get_current_user_from_token(token: str, db: AsyncSession) -> User:
         raise credentials_exception
 
     # Get user from database
-    user_service = AsyncUserService(db)
+    user_service = UserService(db)
     user = await user_service.get_user_by_email(email)
     if user is None:
         raise credentials_exception

aegis/templates/cookiecutter-aegis-project/{{cookiecutter.project_slug}}/app/services/auth/auth_service.py

@@ -2,87 +2,15 @@
 
 from datetime import UTC, datetime
 
-from sqlmodel import Session, select
+from sqlmodel import select
 from sqlmodel.ext.asyncio.session import AsyncSession
 
 from app.core.security import get_password_hash
 from app.models.user import User, UserCreate
 
 
 class UserService:
-    """Service for managing users."""
-
-    def __init__(self, db: Session):
-        self.db = db
-
-    def create_user(self, user_data: UserCreate) -> User:
-        """Create a new user."""
-        # Hash the password
-        hashed_password = get_password_hash(user_data.password)
-
-        # Create user object
-        user = User(
-            email=user_data.email,
-            full_name=user_data.full_name,
-            hashed_password=hashed_password,
-            is_active=user_data.is_active,
-            created_at=datetime.now(UTC),
-        )
-
-        # Save to database
-        self.db.add(user)
-        self.db.commit()
-        self.db.refresh(user)
-
-        return user
-
-    def get_user_by_email(self, email: str) -> User | None:
-        """Get user by email address."""
-        statement = select(User).where(User.email == email)
-        result = self.db.exec(statement)
-        return result.first()
-
-    def get_user_by_id(self, user_id: int) -> User | None:
-        """Get user by ID."""
-        return self.db.get(User, user_id)
-
-    def update_user(self, user_id: int, **updates) -> User | None:
-        """Update user data."""
-        user = self.get_user_by_id(user_id)
-        if not user:
-            return None
-
-        for field, value in updates.items():
-            if hasattr(user, field):
-                setattr(user, field, value)
-
-        user.updated_at = datetime.now(UTC)
-        self.db.add(user)
-        self.db.commit()
-        self.db.refresh(user)
-
-        return user
-
-    def deactivate_user(self, user_id: int) -> User | None:
-        """Deactivate a user account."""
-        return self.update_user(user_id, is_active=False)
-
-    def list_users(self) -> list[User]:
-        """List all users in the system."""
-        statement = select(User).order_by(User.created_at.desc())
-        result = self.db.exec(statement)
-        return list(result.all())
-
-    def find_existing_emails_with_prefix(self, prefix: str, domain: str) -> list[str]:
-        """Find existing emails that match the pattern prefix{number}@domain."""
-        pattern = f"{prefix}%@{domain}"
-        statement = select(User.email).where(User.email.like(pattern))
-        result = self.db.exec(statement)
-        return list(result.all())
-
-
-class AsyncUserService:
-    """Async service for managing users with non-blocking database operations."""
+    """Service for managing users with async database operations."""
 
     def __init__(self, db: AsyncSession):
         self.db = db

aegis/templates/cookiecutter-aegis-project/{{cookiecutter.project_slug}}/app/services/auth/user_service.py

@@ -8,7 +8,7 @@ login, token validation, and protected endpoint access.
 import pytest
 from app.core.security import create_access_token
 from app.models.user import UserCreate
-from app.services.auth.user_service import AsyncUserService, UserService
+from app.services.auth.user_service import UserService
 from fastapi import status
 from fastapi.testclient import TestClient
 from sqlmodel import Session
@@ -48,7 +48,7 @@ class TestAuthEndpoints:
     ):
         """Test registration with already existing email."""
         # Create a user first
-        user_service = AsyncUserService(async_db_session)
+        user_service = UserService(async_db_session)
         existing_user = UserCreate(
             email="existing@example.com",
             full_name="Existing User",
@@ -90,7 +90,7 @@ class TestAuthEndpoints:
     ):
         """Test login with valid credentials."""
         # Create a user first
-        user_service = AsyncUserService(async_db_session)
+        user_service = UserService(async_db_session)
         user_data = UserCreate(
             email="login@example.com",
             full_name="Login User",
@@ -131,7 +131,7 @@ class TestAuthEndpoints:
     ):
         """Test login with wrong password."""
         # Create a user first
-        user_service = AsyncUserService(async_db_session)
+        user_service = UserService(async_db_session)
         user_data = UserCreate(
             email="wrongpass@example.com",
             full_name="Wrong Pass User",
@@ -156,7 +156,7 @@ class TestAuthEndpoints:
     ):
         """Test getting current user with valid token."""
         # Create a user first
-        user_service = AsyncUserService(async_db_session)
+        user_service = UserService(async_db_session)
         user_data = UserCreate(
             email="currentuser@example.com",
             full_name="Current User",
@@ -204,7 +204,7 @@ class TestAuthEndpoints:
     ):
         """Test getting current user with expired token."""
         # Create a user first
-        user_service = AsyncUserService(async_db_session)
+        user_service = UserService(async_db_session)
         user_data = UserCreate(
             email="expiredtoken@example.com",
             full_name="Expired Token User",
@@ -298,7 +298,7 @@ class TestAuthIntegration:
         assert login_response.status_code == status.HTTP_200_OK
 
         # Verify user exists in database
-        user_service = AsyncUserService(async_db_session)
+        user_service = UserService(async_db_session)
         db_user = await user_service.get_user_by_email(user_data["email"])
 
         assert db_user is not None