Building a Simple UI Application with Neuroglia¶

"""Task domain entity."""
from dataclasses import dataclass
from datetime import datetime
from typing import Optional

from neuroglia.data.abstractions import Entity


@dataclass
class Task(Entity):
    """Represents a task in the system."""

    title: str
    description: str
    assigned_to: str
    priority: str  # low, medium, high
    status: str  # pending, in_progress, completed
    created_at: datetime
    created_by: str

    def __init__(
        self,
        id: str,
        title: str,
        description: str,
        assigned_to: str,
        priority: str = "medium",
        status: str = "pending",
        created_by: str = "system"
    ):
        super().__init__()
        self.id = id
        self.title = title
        self.description = description
        self.assigned_to = assigned_to
        self.priority = priority
        self.status = status
        self.created_at = datetime.now()
        self.created_by = created_by

    def complete(self):
        """Mark task as completed."""
        self.status = "completed"

    def assign_to(self, user: str):
        """Assign task to a user."""
        self.assigned_to = user

"""Task repository interface."""
from abc import ABC, abstractmethod
from typing import List, Optional

from domain.entities.task import Task


class TaskRepository(ABC):
    """Abstract repository for task persistence."""

    @abstractmethod
    async def get_all_async(self) -> List[Task]:
        """Get all tasks."""
        pass

    @abstractmethod
    async def get_by_id_async(self, task_id: str) -> Optional[Task]:
        """Get task by ID."""
        pass

    @abstractmethod
    async def get_by_user_async(self, username: str) -> List[Task]:
        """Get tasks assigned to a specific user."""
        pass

    @abstractmethod
    async def save_async(self, task: Task) -> None:
        """Save a task."""
        pass

    @abstractmethod
    async def delete_async(self, task_id: str) -> None:
        """Delete a task."""
        pass

"""In-memory implementation of task repository."""
from typing import Dict, List, Optional

from domain.entities.task import Task
from domain.repositories.task_repository import TaskRepository


class InMemoryTaskRepository(TaskRepository):
    """In-memory task repository for demo purposes."""

    def __init__(self):
        self._tasks: Dict[str, Task] = {}
        self._initialize_sample_data()

    def _initialize_sample_data(self):
        """Initialize with sample tasks."""
        sample_tasks = [
            Task("1", "Review code PR #123", "Critical bug fix needed", "john.doe", "high", "in_progress", "admin"),
            Task("2", "Update documentation", "Add API docs for new endpoints", "jane.smith", "medium", "pending", "admin"),
            Task("3", "Deploy to staging", "Deploy v2.1.0 to staging environment", "admin", "high", "pending", "admin"),
            Task("4", "Client meeting prep", "Prepare slides for Q4 review", "jane.smith", "medium", "in_progress", "manager"),
            Task("5", "Database optimization", "Optimize slow queries in reports", "john.doe", "medium", "pending", "manager"),
            Task("6", "Bug fix: Login timeout", "Users reporting session timeouts", "john.doe", "high", "pending", "user"),
        ]

        for task in sample_tasks:
            self._tasks[task.id] = task

    async def get_all_async(self) -> List[Task]:
        """Get all tasks."""
        return list(self._tasks.values())

    async def get_by_id_async(self, task_id: str) -> Optional[Task]:
        """Get task by ID."""
        return self._tasks.get(task_id)

    async def get_by_user_async(self, username: str) -> List[Task]:
        """Get tasks assigned to a specific user."""
        return [task for task in self._tasks.values() if task.assigned_to == username]

    async def save_async(self, task: Task) -> None:
        """Save a task."""
        self._tasks[task.id] = task

    async def delete_async(self, task_id: str) -> None:
        """Delete a task."""
        if task_id in self._tasks:
            del self._tasks[task_id]

"""Command for creating a new task."""
from dataclasses import dataclass

from neuroglia.core import OperationResult
from neuroglia.mediation import Command, CommandHandler

from domain.entities.task import Task
from domain.repositories.task_repository import TaskRepository


@dataclass
class CreateTaskCommand(Command[OperationResult['TaskDto']]):
    """Command to create a new task."""
    title: str
    description: str
    assigned_to: str
    priority: str
    created_by: str


@dataclass
class TaskDto:
    """Data transfer object for tasks."""
    id: str
    title: str
    description: str
    assigned_to: str
    priority: str
    status: str
    created_by: str


class CreateTaskHandler(CommandHandler[CreateTaskCommand, OperationResult[TaskDto]]):
    """Handler for creating tasks."""

    def __init__(self, task_repository: TaskRepository):
        super().__init__()
        self.task_repository = task_repository

    async def handle_async(self, command: CreateTaskCommand) -> OperationResult[TaskDto]:
        """Handle task creation."""
        # Validation
        if not command.title or not command.title.strip():
            return self.bad_request("Task title is required")

        if not command.assigned_to or not command.assigned_to.strip():
            return self.bad_request("Task must be assigned to a user")

        # Generate ID (in production, use proper ID generation)
        import uuid
        task_id = str(uuid.uuid4())[:8]

        # Create task entity
        task = Task(
            id=task_id,
            title=command.title,
            description=command.description,
            assigned_to=command.assigned_to,
            priority=command.priority,
            created_by=command.created_by
        )

        # Save to repository
        await self.task_repository.save_async(task)

        # Return DTO
        dto = TaskDto(
            id=task.id,
            title=task.title,
            description=task.description,
            assigned_to=task.assigned_to,
            priority=task.priority,
            status=task.status,
            created_by=task.created_by
        )

        return self.created(dto)

"""Query for retrieving tasks."""
from dataclasses import dataclass
from typing import List, Optional

from neuroglia.core import OperationResult
from neuroglia.mediation import Query, QueryHandler

from application.commands.create_task_command import TaskDto
from domain.repositories.task_repository import TaskRepository


@dataclass
class GetTasksQuery(Query[OperationResult[List[TaskDto]]]):
    """Query to get tasks, optionally filtered by user."""
    username: Optional[str] = None
    role: Optional[str] = None


class GetTasksHandler(QueryHandler[GetTasksQuery, OperationResult[List[TaskDto]]]):
    """Handler for retrieving tasks."""

    def __init__(self, task_repository: TaskRepository):
        super().__init__()
        self.task_repository = task_repository

    async def handle_async(self, query: GetTasksQuery) -> OperationResult[List[TaskDto]]:
        """Handle task retrieval with role-based filtering."""

        # Get tasks based on role
        if query.role == "admin":
            # Admins see all tasks
            tasks = await self.task_repository.get_all_async()
        elif query.role == "manager":
            # Managers see all tasks except admin-created ones
            all_tasks = await self.task_repository.get_all_async()
            tasks = [t for t in all_tasks if t.created_by != "admin"]
        else:
            # Regular users see only their assigned tasks
            if not query.username:
                return self.bad_request("Username required for user role")
            tasks = await self.task_repository.get_by_user_async(query.username)

        # Convert to DTOs
        dtos = [
            TaskDto(
                id=task.id,
                title=task.title,
                description=task.description,
                assigned_to=task.assigned_to,
                priority=task.priority,
                status=task.status,
                created_by=task.created_by
            )
            for task in tasks
        ]

        return self.ok(dtos)

"""Application settings and configuration."""
import logging
import os
import sys
from dataclasses import dataclass


@dataclass
class AppSettings:
    """Application settings for JWT authentication."""

    # Application
    app_name: str = "Simple UI"
    app_version: str = "1.0.0"
    debug: bool = True

    # JWT Settings
    jwt_secret_key: str = os.getenv("JWT_SECRET_KEY", "your-secret-key-change-in-production")
    jwt_algorithm: str = "HS256"
    jwt_expiration_minutes: int = 60


# Global settings instance
app_settings = AppSettings()


def configure_logging(log_level: str = "INFO") -> None:
    """Configure application-wide logging."""
    logging.basicConfig(
        level=getattr(logging, log_level.upper(), logging.INFO),
        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
        handlers=[logging.StreamHandler(sys.stdout)]
    )

    # Set third-party loggers to WARNING to reduce noise
    logging.getLogger("uvicorn").setLevel(logging.WARNING)
    logging.getLogger("fastapi").setLevel(logging.WARNING)

"""Authentication controller - JWT-only, no sessions."""
from datetime import datetime, timedelta
from typing import Optional

from application.settings import app_settings
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import JWTError, jwt
from pydantic import BaseModel

from neuroglia.dependency_injection.service_provider import ServiceProviderBase
from neuroglia.mapping import Mapper
from neuroglia.mediation import Mediator
from neuroglia.mvc import ControllerBase
from classy_fastapi import post

# JWT Configuration - use shared settings
SECRET_KEY = app_settings.jwt_secret_key
ALGORITHM = app_settings.jwt_algorithm
ACCESS_TOKEN_EXPIRE_MINUTES = app_settings.jwt_expiration_minutes

# Mock user database (in production, use real database)
USERS_DB = {
    "admin": {"username": "admin", "password": "admin123", "role": "admin"},
    "manager": {"username": "manager", "password": "manager123", "role": "manager"},
    "john.doe": {"username": "john.doe", "password": "user123", "role": "user"},
    "jane.smith": {"username": "jane.smith", "password": "user123", "role": "user"},
}


class LoginRequest(BaseModel):
    """Login request model."""
    username: str
    password: str


class TokenResponse(BaseModel):
    """Token response model."""
    access_token: str
    token_type: str
    username: str
    role: str


class UserInfo(BaseModel):
    """User information."""
    username: str
    role: str


security = HTTPBearer()


def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    """Create JWT access token."""
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt


async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)) -> UserInfo:
    """Get current user from JWT token."""
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        token = credentials.credentials
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        # Use 'username' field from token, not 'sub' (which contains user ID)
        username: str = payload.get("username")
        role: str = payload.get("role")
        if username is None or role is None:
            raise credentials_exception
        return UserInfo(username=username, role=role)
    except JWTError:
        raise credentials_exception


class AuthController(ControllerBase):
    """Authentication controller."""

    def __init__(self, service_provider: ServiceProviderBase, mapper: Mapper, mediator: Mediator):
        super().__init__(service_provider, mapper, mediator)

    @post("/login", response_model=TokenResponse)
    async def login(self, request: LoginRequest) -> TokenResponse:
        """Authenticate user and return JWT token."""
        # Validate credentials
        user = USERS_DB.get(request.username)
        if not user or user["password"] != request.password:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Incorrect username or password",
                headers={"WWW-Authenticate": "Bearer"},
            )

        # Create token
        access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
        access_token = create_access_token(
            data={"sub": user["username"], "role": user["role"]},
            expires_delta=access_token_expires
        )

        return TokenResponse(
            access_token=access_token,
            token_type="bearer",
            username=user["username"],
            role=user["role"]
        )

"""Tasks API controller."""
from typing import List

from fastapi import Depends

from neuroglia.dependency_injection.service_provider import ServiceProviderBase
from neuroglia.mapping import Mapper
from neuroglia.mediation import Mediator
from neuroglia.mvc import ControllerBase
from classy_fastapi import get, post

from api.controllers.auth_controller import get_current_user, UserInfo
from application.commands.create_task_command import CreateTaskCommand, TaskDto
from application.queries.get_tasks_query import GetTasksQuery


class TasksController(ControllerBase):
    """Tasks management controller."""

    def __init__(self, service_provider: ServiceProviderBase, mapper: Mapper, mediator: Mediator):
        super().__init__(service_provider, mapper, mediator)

    @get("/", response_model=List[TaskDto])
    async def get_tasks(self, current_user: UserInfo = Depends(get_current_user)) -> List[TaskDto]:
        """Get tasks based on user role."""
        query = GetTasksQuery(username=current_user.username, role=current_user.role)
        result = await self.mediator.execute_async(query)
        return self.process(result)

    @post("/", response_model=TaskDto, status_code=201)
    async def create_task(
        self,
        command: CreateTaskCommand,
        current_user: UserInfo = Depends(get_current_user)
    ) -> TaskDto:
        """Create a new task (admin only)."""
        if current_user.role != "admin":
            raise HTTPException(status_code=403, detail="Only admins can create tasks")

        command.created_by = current_user.username
        result = await self.mediator.execute_async(command)
        return self.process(result)

"""UI controller for serving HTML pages."""
from fastapi import Request
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates

from neuroglia.dependency_injection.service_provider import ServiceProviderBase
from neuroglia.mapping import Mapper
from neuroglia.mediation import Mediator
from neuroglia.mvc import ControllerBase
from classy_fastapi import get


templates = Jinja2Templates(directory="ui/templates")


class UIController(ControllerBase):
    """Controller for UI pages."""

    def __init__(self, service_provider: ServiceProviderBase, mapper: Mapper, mediator: Mediator):
        super().__init__(service_provider, mapper, mediator)

    @get("/", response_class=HTMLResponse)
    async def index(self, request: Request):
        """Render main application page."""
        return templates.TemplateResponse("index.html", {"request": request})

"""Simple UI application entry point."""
import logging
import sys
from pathlib import Path

from fastapi import FastAPI

# Add parent directories to Python path for framework imports
project_root = Path(__file__).parent.parent.parent
sys.path.insert(0, str(project_root / "src"))
sys.path.insert(0, str(Path(__file__).parent))

from neuroglia.hosting.web import WebApplicationBuilder, SubAppConfig
from neuroglia.mediation import Mediator
from neuroglia.mapping import Mapper
from neuroglia.serialization.json import JsonSerializer

from domain.repositories.task_repository import TaskRepository
from integration.repositories.in_memory_task_repository import InMemoryTaskRepository

# Configure logging
from application.settings import configure_logging
configure_logging(log_level="INFO")
log = logging.getLogger(__name__)


def create_app() -> FastAPI:
    """
    Create Simple UI application with JWT-only authentication.

    Architecture:
    - API sub-app (/api prefix) - REST API with JWT authentication
    - UI sub-app (/ prefix) - Web interface (no session middleware)

    Authentication: Pure stateless JWT - tokens stored client-side in localStorage
    """

    log.info("🚀 Creating Simple UI application...")

    # Create application builder
    builder = WebApplicationBuilder()

    # Configure services
    services = builder.services

    # Register repositories
    services.add_singleton(TaskRepository, InMemoryTaskRepository)

    # Configure Core services using native .configure() methods
    Mediator.configure(builder, ["application.commands", "application.queries"])
    Mapper.configure(builder, ["application.commands", "domain.entities"])
    JsonSerializer.configure(builder, ["domain.entities"])

    # Configure sub-applications declaratively
    # API sub-app: REST API with JWT authentication
    builder.add_sub_app(
        SubAppConfig(
            path="/api",
            name="api",
            title="Simple UI API",
            description="Task management REST API with JWT authentication",
            version="1.0.0",
            controllers=["api.controllers"],
            docs_url="/docs",
        )
    )

    # UI sub-app: Web interface (JWT-only auth, no session middleware needed)
    static_dir = Path(__file__).parent / "static"
    templates_dir = Path(__file__).parent / "ui" / "templates"

    builder.add_sub_app(
        SubAppConfig(
            path="/",
            name="ui",
            title="Simple UI",
            description="Task management web interface",
            version="1.0.0",
            controllers=["ui.controllers"],
            static_files={"/static": str(static_dir)},
            templates_dir=str(templates_dir),
            docs_url=None,  # Disable docs for UI
            # No SessionMiddleware - using JWT-only authentication
        )
    )

    # Build the complete application
    app = builder.build_app_with_lifespan(
        title="Simple UI",
        description="Task management application with JWT-only authentication",
        version="1.0.0",
        debug=True,
    )

    log.info("✅ Application created successfully!")
    log.info("📊 Access points:")
    log.info("   - UI: http://localhost:8082/")
    log.info("   - API Docs: http://localhost:8082/api/docs")
    log.info("   - Auth: POST /api/auth/login")
    log.info("   - Tasks: GET /api/tasks/")

    return app


if __name__ == "__main__":
    import uvicorn

    app = create_app()

    log.info("🌐 Starting server on http://localhost:8000")
    log.info("👤 Demo users:")
    log.info("   - admin / admin123 (can see all tasks, can create)")
    log.info("   - manager / manager123 (can see non-admin tasks)")
    log.info("   - john.doe / user123 (can see own tasks)")
    log.info("   - jane.smith / user123 (can see own tasks)")

    uvicorn.run(app, host="0.0.0.0", port=8000)