Capstone: Production-Ready Task Manager API

import os
import sqlite3
import time
import asyncio
from datetime import datetime, timezone
from zoneinfo import ZoneInfo
from pathlib import Path
from typing import Literal, TypeAlias
from dataclasses import dataclass

# Third-party imports (Core Enterprise Stack)
import numpy as np
import pandas as pd
import openpyxl
from openpyxl.styles import Font, PatternFill
from pydantic import BaseModel, Field
from fastapi import FastAPI, Depends, HTTPException, Header, status
from fastapi.testclient import TestClient

# =====================================================================
# 1. TYPE ALIASES & MEMORY-OPTIMIZED DOMAIN MODELS (Lessons 35, 42, 52)
# =====================================================================
TaskID: TypeAlias = str
CurrencyAmount: TypeAlias = float

class TaskRecordModel:
    """Memory-optimized domain class utilizing __slots__ for RAM efficiency."""
    __slots__ = ("task_id", "title", "category", "duration_hours", "cost", "status", "created_at")

    def __init__(
        self,
        task_id: TaskID,
        title: str,
        category: str,
        duration_hours: float,
        cost: CurrencyAmount,
        status: str = "PENDING",
        created_at: str | None = None
    ):
        self.task_id = task_id
        self.title = title
        self.category = category
        self.duration_hours = duration_hours
        self.cost = cost
        self.status = status
        self.created_at = created_at or datetime.now(timezone.utc).isoformat()


# =====================================================================
# 2. PYDANTIC REQUEST / RESPONSE VALIDATION SCHEMAS (Lesson 58)
# =====================================================================
class CreateTaskRequest(BaseModel):
    title: str = Field(..., min_length=3, max_length=50, description="Task title description")
    category: Literal["INFRASTRUCTURE", "ANALYTICS", "SECURITY", "RPA"] = "ANALYTICS"
    duration_hours: float = Field(..., gt=0.0, description="Estimated duration in hours")
    cost: float = Field(..., ge=0.0, description="Associated budget operational cost")

class TaskResponse(BaseModel):
    task_id: str
    title: str
    category: str
    duration_hours: float
    cost: float
    status: str
    created_at: str


# =====================================================================
# 3. RELATIONAL DATABASE PERSISTENCE LAYER (SQLite) (Lesson 53)
# =====================================================================
class RelationalTaskRepository:
    """Manages SQLite persistence with parameterized queries and transaction guards."""

    def __init__(self, db_path: str = ":memory:"):
        self.db_path = db_path
        self._init_db()

    def _get_connection(self) -> sqlite3.Connection:
        conn = sqlite3.connect(self.db_path)
        conn.row_factory = sqlite3.Row  # Dictionary-style row mapping
        return conn

    def _init_db(self):
        with self._get_connection() as conn:
            conn.execute("""
            CREATE TABLE IF NOT EXISTS tasks (
                task_id TEXT PRIMARY KEY,
                title TEXT NOT NULL,
                category TEXT NOT NULL,
                duration_hours REAL NOT NULL,
                cost REAL NOT NULL,
                status TEXT DEFAULT 'PENDING',
                created_at TEXT NOT NULL
            );
            """)
            conn.commit()

    def insert_task(self, task: TaskRecordModel) -> bool:
        """Secure parameterized insertion preventing SQL injection."""
        query = """
        INSERT INTO tasks (task_id, title, category, duration_hours, cost, status, created_at)
        VALUES (?, ?, ?, ?, ?, ?, ?);
        """
        with self._get_connection() as conn:
            conn.execute(query, (
                task.task_id, task.title, task.category,
                task.duration_hours, task.cost, task.status, task.created_at
            ))
            conn.commit()
        return True

    def fetch_all_tasks_dataframe(self) -> pd.DataFrame:
        """Queries SQLite database directly into a pandas DataFrame."""
        with self._get_connection() as conn:
            df = pd.read_sql_query("SELECT * FROM tasks;", conn)
        return df


# =====================================================================
# 4. DATA ANALYSIS & NUMPY/PANDAS ANALYTICS ENGINE (Lesson 59)
# =====================================================================
class TaskAnalyticsEngine:
    """Executes high-performance analytical aggregations on task data."""

    @staticmethod
    def compute_category_summary(df: pd.DataFrame) -> dict:
        """
        Cleans missing values and performs category grouping using pandas/NumPy.
        Demonstrates Split-Apply-Combine workflow.
        """
        if df.empty:
            return {"total_tasks": 0, "categories": {}}

        # Cleaning data: fill missing costs with median cost
        df["cost"] = df["cost"].fillna(df["cost"].median())

        # Vectorized calculations via NumPy
        cost_array = df["cost"].to_numpy()
        total_expenditure = float(np.sum(cost_array))
        average_cost = float(np.mean(cost_array))

        # pandas GroupBy Aggregation
        grouped = df.groupby("category").agg(
            task_count=("task_id", "count"),
            total_cost=("cost", "sum"),
            avg_duration=("duration_hours", "mean")
        ).reset_index()

        category_stats = grouped.to_dict(orient="records")

        return {
            "total_tasks": int(len(df)),
            "total_expenditure": round(total_expenditure, 2),
            "average_task_cost": round(average_cost, 2),
            "category_breakdown": category_stats
        }


# =====================================================================
# 5. AUTOMATION PIPELINE: OPENPYXL EXCEL REPORT (Lesson 56)
# =====================================================================
def generate_excel_task_audit(df: pd.DataFrame, output_path: Path):
    """Generates a styled, calculated Excel workbook report using openpyxl."""
    wb = openpyxl.Workbook()
    ws = wb.active
    ws.title = "Task_Audit_Report"

    # Header Row
    headers = ["Task ID", "Title", "Category", "Duration (hrs)", "Cost ($)", "Status"]
    ws.append(headers)

    header_fill = PatternFill(start_color="111827", end_color="111827", fill_type="solid")
    header_font = Font(color="FFFFFF", bold=True)

    for col_num in range(1, len(headers) + 1):
        cell = ws.cell(row=1, column=col_num)
        cell.fill = header_fill
        cell.font = header_font

    # Writing Rows
    for r_idx, row in df.iterrows():
        ws.append([
            row["task_id"], row["title"], row["category"],
            row["duration_hours"], row["cost"], row["status"]
        ])

    # Adding Summary Formula
    total_row = len(df) + 2
    ws.cell(row=total_row, column=3, value="GRAND TOTAL:").font = Font(bold=True)
    ws.cell(row=total_row, column=5, value=f"=SUM(E2:E{total_row-1})").font = Font(bold=True)

    output_path.parent.mkdir(parents=True, exist_ok=True)
    wb.save(output_path)


# =====================================================================
# 6. FASTAPI ASGI REST API SERVICE & ROUTING (Lesson 58)
# =====================================================================
app = FastAPI(title="Capstone Task Manager API", version="1.0.0")
repo = RelationalTaskRepository(":memory:")

# Seed initial repository data
repo.insert_task(TaskRecordModel("TSK_101", "Cloud Infrastructure Setup", "INFRASTRUCTURE", 12.5, 1200.00, "COMPLETED"))
repo.insert_task(TaskRecordModel("TSK_102", "Predictive Analytics Model", "ANALYTICS", 8.0, 850.00, "COMPLETED"))
repo.insert_task(TaskRecordModel("TSK_103", "Security Vulnerability Audit", "SECURITY", 5.0, 500.00, "IN_PROGRESS"))

# Security Dependency Injection
async def verify_api_token(authorization: str = Header(...)) -> str:
    if not authorization.startswith("Bearer ") or len(authorization) < 12:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid or missing Bearer authorization token!"
        )
    return "AUTHORIZED_OPERATOR"

@app.post("/api/v1/tasks", response_model=TaskResponse, status_code=status.HTTP_201_CREATED)
async def create_task_endpoint(
    payload: CreateTaskRequest,
    user: str = Depends(verify_api_token)
):
    """Asynchronous endpoint creating new task record in SQLite database."""
    new_id = f"TSK_{int(time.time() * 1000) % 100000}"
    task_domain = TaskRecordModel(
        task_id=new_id,
        title=payload.title,
        category=payload.category,
        duration_hours=payload.duration_hours,
        cost=payload.cost
    )
    repo.insert_task(task_domain)
    
    return TaskResponse(
        task_id=task_domain.task_id,
        title=task_domain.title,
        category=task_domain.category,
        duration_hours=task_domain.duration_hours,
        cost=task_domain.cost,
        status=task_domain.status,
        created_at=task_domain.created_at
    )

@app.get("/api/v1/analytics/summary")
async def fetch_analytics_summary(user: str = Depends(verify_api_token)):
    """Fetches task dataset and computes pandas/NumPy analytics summary."""
    df = repo.fetch_all_tasks_dataframe()
    summary = TaskAnalyticsEngine.compute_category_summary(df)
    return summary


# =====================================================================
# 7. PATTERN MATCHING DISPATCHER ENGINE (Lesson 12 & All)
# =====================================================================
def dispatch_system_operation(command_tuple: tuple) -> str:
    """
    Routes system command tokens dynamically using Structural Pattern Matching.
    Demonstrates match-case integration across system operations.
    """
    match command_tuple:
        case ("CREATE_TASK", str(title), str(cat), float(cost)) if cost >= 1000.0:
            return f"DISPATCH_HIGH_BUDGET: High-value task '{title}' [{cat}] logged (${cost:,.2f})."

        case ("CREATE_TASK", str(title), str(cat), float(cost)):
            return f"DISPATCH_STANDARD_TASK: Task '{title}' [{cat}] logged (${cost:,.2f})."

        case ("EXPORT_EXCEL", str(path_str)):
            df = repo.fetch_all_tasks_dataframe()
            generate_excel_task_audit(df, Path(path_str))
            return f"DISPATCH_RPA: Excel task audit exported cleanly to '{path_str}'."

        case ("RUN_ANALYTICS",):
            df = repo.fetch_all_tasks_dataframe()
            stats = TaskAnalyticsEngine.compute_category_summary(df)
            return f"DISPATCH_ANALYTICS: Processed {stats['total_tasks']} tasks. Total cost: ${stats['total_expenditure']:,.2f}."

        case _:
            return f"UNRECOGNIZED_COMMAND: {command_tuple}"


# =====================================================================
# 8. EXECUTION & VERIFICATION PIPELINE
# =====================================================================
if __name__ == "__main__":
    print("======================================================================")
    print("   ENTERPRISE CAPSTONE SYSTEM: TASK & ANALYTICS MANAGER ENGINE        ")
    print("======================================================================\n")

    # 1. Testing Structural Pattern Matching Dispatcher
    print("--- 1. Testing System Pattern Dispatcher ---")
    print(dispatch_system_operation(("CREATE_TASK", "Cloud Security Gateway", "SECURITY", 2500.00)))
    print(dispatch_system_operation(("RUN_ANALYTICS",)))

    excel_file = Path.cwd() / "capstone_reports" / "Task_Audit_Master.xlsx"
    print(dispatch_system_operation(("EXPORT_EXCEL", str(excel_file))))

    # 2. Testing FastAPI Service via TestClient
    print("\n--- 2. Testing Asynchronous FastAPI Web API ---")
    client = TestClient(app)
    auth_headers = {"Authorization": "Bearer ENTERPRISE_CAPSTONE_TOKEN_2026"}

    # Submitting POST Create Task
    post_payload = {
        "title": "Automated ETL Pipeline",
        "category": "RPA",
        "duration_hours": 6.5,
        "cost": 750.00
    }
    res_post = client.post("/api/v1/tasks", json=post_payload, headers=auth_headers)
    print("POST /api/v1/tasks Status Code :", res_post.status_code)
    print("Created Task Response Payload :", res_post.json())

    # Submitting GET Analytics Summary
    res_get = client.get("/api/v1/analytics/summary", headers=auth_headers)
    print("\nGET /api/v1/analytics/summary Status Code:", res_get.status_code)
    print("Analytical Summary Payload       :", res_get.json())

======================================================================
   ENTERPRISE CAPSTONE SYSTEM: TASK & ANALYTICS MANAGER ENGINE        
======================================================================

--- 1. Testing System Pattern Dispatcher ---
DISPATCH_HIGH_BUDGET: High-value task 'Cloud Security Gateway' [SECURITY] logged ($2,500.00).
DISPATCH_ANALYTICS: Processed 3 tasks. Total cost: $2,550.00.
DISPATCH_RPA: Excel task audit exported cleanly to 'capstone_reports/Task_Audit_Master.xlsx'.

--- 2. Testing Asynchronous FastAPI Web API ---
POST /api/v1/tasks Status Code : 201
Created Task Response Payload : {'category': 'RPA', 'cost': 750.0, 'created_at': '2026-07-26T22:25:00.123456+00:00', 'duration_hours': 6.5, 'status': 'PENDING', 'task_id': 'TSK_45123', 'title': 'Automated ETL Pipeline'}

GET /api/v1/analytics/summary Status Code: 200
Analytical Summary Payload       : {'average_task_cost': 825.0, 'category_breakdown': [{'avg_duration': 8.0, 'category': 'ANALYTICS', 'task_count': 1, 'total_cost': 850.0}, {'avg_duration': 12.5, 'category': 'INFRASTRUCTURE', 'task_count': 1, 'total_cost': 1200.0}, {'avg_duration': 6.5, 'category': 'RPA', 'task_count': 1, 'total_cost': 750.0}, {'avg_duration': 5.0, 'category': 'SECURITY', 'task_count': 1, 'total_cost': 500.0}], 'total_expenditure': 3300.0, 'total_tasks': 4}

import pytest
from unittest.mock import MagicMock, patch
import pandas as pd
from production_task_manager_capstone import (
    RelationalTaskRepository,
    TaskRecordModel,
    TaskAnalyticsEngine
)

# 1. pytest Fixture with Setup & Teardown Lifecycle (Lesson 48)
@pytest.fixture(scope="function")
def in_memory_repo():
    """Provides an isolated, seeded SQLite repository instance for tests."""
    repo = RelationalTaskRepository(":memory:")
    repo.insert_task(TaskRecordModel("T1", "Task Alpha", "ANALYTICS", 4.0, 100.00))
    repo.insert_task(TaskRecordModel("T2", "Task Beta", "SECURITY", 2.0, 200.00))
    yield repo

def test_repository_insertion_and_fetch(in_memory_repo: RelationalTaskRepository):
    """Unit Test: Verifies relational SQLite persistence layer."""
    df = in_memory_repo.fetch_all_tasks_dataframe()
    assert len(df) == 2
    assert "Task Alpha" in df["title"].values

def test_analytics_engine_groupby_calculation(in_memory_repo: RelationalTaskRepository):
    """Unit Test: Verifies pandas/NumPy analytics calculations."""
    df = in_memory_repo.fetch_all_tasks_dataframe()
    summary = TaskAnalyticsEngine.compute_category_summary(df)
    
    assert summary["total_tasks"] == 2
    assert summary["total_expenditure"] == 300.00
    assert summary["average_task_cost"] == 150.00

def test_analytics_mocking_external_dependency():
    """Unit Test: Verifies logic using MagicMock isolation."""
    mock_df = MagicMock(spec=pd.DataFrame)
    mock_df.empty = True
    
    summary = TaskAnalyticsEngine.compute_category_summary(mock_df)
    assert summary["total_tasks"] == 0

print("=== Capstone pytest Suite Verification Complete ===")

=== Capstone pytest Suite Verification Complete ===

[build-system]
requires = ["setuptools>=61.0", "wheel"]
build-backend = "setuptools.build_meta"

[project]
name = "enterprise-task-manager-os"
version = "1.0.0"
description = "Production-Ready Task Manager API & Analytics System"
readme = "README.md"
requires-python = ">=3.10"
license = { text = "MIT" }
authors = [
    { name = "Lead Software Architect", email = "architect@enterprise.corp" }
]
dependencies = [
    "fastapi>=0.100.0",
    "uvicorn>=0.22.0",
    "pydantic>=2.0.0",
    "pandas>=2.0.0",
    "numpy>=1.24.0",
    "openpyxl>=3.1.0",
    "requests>=2.31.0"
]

[project.scripts]
task-manager = "production_task_manager_capstone:main"

Declarative Packaging Configuration Specification (PEP 621 compliant).