Add a Read-Only Endpoint

Create a simple GET endpoint that queries the database and returns aggregated data.

Codebase Orientation

platform/
├── api/
│   ├── routes/
│   │   └── v1/
│   │       ├── scans.py      ← Add your endpoint here
│   │       ├── findings.py
│   │       └── leads.py
│   ├── schemas/
│   │   └── v1/
│   │       └── scan.py       ← Add response schema here
│   └── dependencies/
│       └── db.py             ← TenantDB dependency
├── db/
│   └── models/
│       ├── scan.py
│       ├── finding.py
│       └── lead.py
└── server/
    └── app.py                ← FastAPI app
            

What You'll Learn

• FastAPI routing: Path parameters, response models, status codes
• Dependency injection: Using Depends(get_tenant_db)
• Pydantic schemas: Defining response models
• SQLAlchemy queries: Simple count queries
• Tenant isolation: How TenantDB scopes queries

Implementation Guide

Step 1: Define the Response Schema

In platform/api/schemas/v1/scan.py, add:

from pydantic import BaseModel

class ScanSummaryResponse(BaseModel):
    scan_id: UUID
    findings_count: int
    leads_count: int
    endpoints_count: int

Step 2: Add the Route

In platform/api/routes/v1/scans.py, add:

from api.schemas.v1.scan import ScanSummaryResponse

@router.get("/v1/scans/{scan_id}/summary", response_model=ScanSummaryResponse)
async def get_scan_summary(
    scan_id: UUID,
    user: ActiveUser = Depends(get_current_user),
    tenant_db: TenantDB = Depends(get_tenant_db),
) -> ScanSummaryResponse:
    """Get summary counts for a scan."""
    
    # Verify scan exists and belongs to tenant
    service = ScanService(tenant_db, encryption_service)
    try:
        scan = await service.get_scan(scan_id)
    except NotFoundError:
        raise HTTPException(status_code=404, detail="Scan not found")
    
    # Count related entities
    # Option 1: Use existing services
    finding_service = FindingService(tenant_db, service)
    lead_service = LeadService(tenant_db)
    endpoint_service = EndpointService(tenant_db)
    
    # Option 2: Direct count queries (more efficient)
    # findings_count = await tenant_db.session.scalar(
    #     select(func.count(Finding.id)).where(Finding.scan_id == scan_id)
    # )
    
    return ScanSummaryResponse(
        scan_id=scan_id,
        findings_count=...,  # Your implementation
        leads_count=...,
        endpoints_count=...,
    )

Step 3: Test Locally

# Start the platform
cd ~/projects/tenzai
mise run platform

# Test your endpoint (replace with actual IDs)
curl http://localhost:8000/v1/scans/{scan_id}/summary \
  -H "Authorization: Bearer $TOKEN"

# Or use the Swagger UI
open http://localhost:8000/docs

Key Patterns to Notice

• TenantDB: All queries are automatically scoped to the current tenant
• ActiveUser: Authentication is handled by the dependency
• Service layer: Business logic lives in services, not routes
• HTTPException: Standard way to return error responses

Stretch Goals

• Add caching for the summary (it's expensive to recompute)
• Add severity breakdown for findings (critical/high/medium/low counts)
• Write a unit test for the endpoint