Files
claudetools/api/routers/admin_quotes.py
azcomputerguru a1a19f8c00 sync: Auto-sync from Mikes-MacBook-Air.local at 2026-03-09 08:14:13
Synced files:
- Session logs updated
- Latest context and credentials
- Command/directive updates

Machine: Mikes-MacBook-Air.local
Timestamp: 2026-03-09 08:14:13

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-03-09 08:14:13 -07:00

385 lines
11 KiB
Python

"""
Admin Quote API router for ClaudeTools.
This module defines all admin REST API endpoints for managing quotes,
requiring JWT authentication for access.
"""
from typing import Optional
from uuid import UUID
from fastapi import APIRouter, Depends, Query, status
from sqlalchemy.orm import Session
from api.database import get_db
from api.middleware.auth import get_current_user
from api.schemas.quote import (
QuoteAdminResponse,
QuoteAdminUpdate,
QuoteActivityResponse,
QuoteItemResponse,
QuoteListItem,
QuoteListResponse,
QuoteNotificationResponse,
QuoteStatsResponse,
QuoteStatus,
)
from api.services import quote_service
# Create router with authentication required
router = APIRouter()
@router.get(
"",
response_model=QuoteListResponse,
summary="List all quotes",
description="Retrieve a paginated list of all quotes with optional filtering",
status_code=status.HTTP_200_OK,
)
def list_quotes(
skip: int = Query(
default=0,
ge=0,
description="Number of records to skip for pagination"
),
limit: int = Query(
default=100,
ge=1,
le=1000,
description="Maximum number of records to return (max 1000)"
),
status_filter: Optional[str] = Query(
default=None,
alias="status",
description="Filter by status (draft, submitted, reviewing, approved, rejected, expired)"
),
search: Optional[str] = Query(
default=None,
description="Search in company_name, contact_name, contact_email"
),
db: Session = Depends(get_db),
current_user: dict = Depends(get_current_user),
):
"""
List all quotes with pagination and filtering.
- **skip**: Number of quotes to skip (default: 0)
- **limit**: Maximum number of quotes to return (default: 100, max: 1000)
- **status**: Filter by quote status
- **search**: Search in company name, contact name, or email
Returns a list of quotes with pagination metadata.
**Example Request:**
```
GET /api/admin/quotes?skip=0&limit=50&status=submitted
Authorization: Bearer <token>
```
**Example Response:**
```json
{
"total": 25,
"skip": 0,
"limit": 50,
"quotes": [
{
"id": "123e4567-e89b-12d3-a456-426614174000",
"access_token": "xYz123...",
"status": "submitted",
"company_name": "Acme Corporation",
"contact_name": "John Doe",
"contact_email": "john@acme.com",
"employee_count": 25,
"monthly_total": "450.00",
"setup_total": "500.00",
"item_count": 3,
"submitted_at": "2024-01-15T14:30:00Z",
"created_at": "2024-01-15T10:30:00Z"
}
]
}
```
"""
quotes, total = quote_service.list_quotes(
db=db,
skip=skip,
limit=limit,
status_filter=status_filter,
search=search
)
# Build list items with item counts
quote_items = []
for quote in quotes:
quote_items.append(QuoteListItem(
id=quote.id,
access_token=quote.access_token,
status=quote.status,
company_name=quote.company_name,
contact_name=quote.contact_name,
contact_email=quote.contact_email,
employee_count=quote.employee_count,
monthly_total=quote.monthly_total,
setup_total=quote.setup_total,
item_count=len(quote.items),
submitted_at=quote.submitted_at,
created_at=quote.created_at
))
return QuoteListResponse(
total=total,
skip=skip,
limit=limit,
quotes=quote_items
)
@router.get(
"/stats",
response_model=QuoteStatsResponse,
summary="Get quote statistics",
description="Get dashboard statistics for quotes",
status_code=status.HTTP_200_OK,
)
def get_stats(
db: Session = Depends(get_db),
current_user: dict = Depends(get_current_user),
):
"""
Get quote statistics for the admin dashboard.
Returns aggregate statistics including totals, counts by status,
and conversion rates.
**Example Request:**
```
GET /api/admin/quotes/stats
Authorization: Bearer <token>
```
**Example Response:**
```json
{
"total_quotes": 150,
"quotes_by_status": {
"draft": 45,
"submitted": 60,
"reviewing": 15,
"approved": 25,
"rejected": 3,
"expired": 2
},
"total_monthly_value": "12500.00",
"total_setup_value": "8500.00",
"quotes_this_month": 28,
"quotes_submitted_this_month": 18,
"average_monthly_value": "125.00",
"conversion_rate": "66.67"
}
```
"""
return quote_service.get_quote_stats(db)
@router.get(
"/{quote_id}",
response_model=QuoteAdminResponse,
summary="Get quote by ID",
description="Retrieve a single quote by its ID with full details",
status_code=status.HTTP_200_OK,
responses={
200: {
"description": "Quote found and returned",
"model": QuoteAdminResponse,
},
404: {
"description": "Quote not found",
"content": {
"application/json": {
"example": {"detail": "Quote with ID 123e4567-e89b-12d3-a456-426614174000 not found"}
}
},
},
},
)
def get_quote(
quote_id: UUID,
db: Session = Depends(get_db),
current_user: dict = Depends(get_current_user),
):
"""
Get a specific quote by ID with full admin details.
Returns the quote with all items, activities, and notifications.
**Example Request:**
```
GET /api/admin/quotes/123e4567-e89b-12d3-a456-426614174000
Authorization: Bearer <token>
```
**Example Response:**
```json
{
"id": "123e4567-e89b-12d3-a456-426614174000",
"access_token": "xYz123...",
"status": "submitted",
"company_name": "Acme Corporation",
"contact_name": "John Doe",
"contact_email": "john@acme.com",
"admin_notes": "Follow up scheduled for next week",
"ip_address": "192.168.1.100",
"user_agent": "Mozilla/5.0...",
"items": [...],
"activities": [
{
"id": "789...",
"action": "submitted",
"description": "Quote submitted by John Doe (john@acme.com)",
"actor": "john@acme.com",
"created_at": "2024-01-15T14:30:00Z"
}
],
"notifications": [...]
}
```
"""
quote = quote_service.get_quote_by_id(db, quote_id)
# Build response with all related data
items_response = []
for item in quote.items:
items_response.append(QuoteItemResponse(
id=item.id,
quote_id=item.quote_id,
service_name=item.service_name,
service_description=item.service_description,
category=item.category,
billing_frequency=item.billing_frequency,
unit_price=item.unit_price,
quantity=item.quantity,
setup_fee=item.setup_fee,
is_required=item.is_required,
sort_order=item.sort_order,
line_total=item.line_total,
monthly_amount=item.monthly_amount,
created_at=item.created_at,
updated_at=item.updated_at
))
activities_response = []
for activity in quote.activities:
activities_response.append(QuoteActivityResponse(
id=activity.id,
quote_id=activity.quote_id,
action=activity.action,
description=activity.description,
actor=activity.actor,
ip_address=activity.ip_address,
created_at=activity.created_at
))
notifications_response = []
for notification in quote.notifications:
notifications_response.append(QuoteNotificationResponse(
id=notification.id,
quote_id=notification.quote_id,
notification_type=notification.notification_type,
recipient=notification.recipient,
subject=notification.subject,
status=notification.status,
sent_at=notification.sent_at,
error_message=notification.error_message,
created_at=notification.created_at
))
return QuoteAdminResponse(
id=quote.id,
access_token=quote.access_token,
status=quote.status,
company_name=quote.company_name,
contact_name=quote.contact_name,
contact_email=quote.contact_email,
contact_phone=quote.contact_phone,
employee_count=quote.employee_count,
notes=quote.notes,
admin_notes=quote.admin_notes,
monthly_total=quote.monthly_total,
setup_total=quote.setup_total,
annual_total=quote.annual_total,
expires_at=quote.expires_at,
submitted_at=quote.submitted_at,
ip_address=quote.ip_address,
user_agent=quote.user_agent,
created_at=quote.created_at,
updated_at=quote.updated_at,
items=items_response,
activities=activities_response,
notifications=notifications_response
)
@router.put(
"/{quote_id}",
response_model=QuoteAdminResponse,
summary="Update quote status/notes",
description="Update a quote's status or admin notes",
status_code=status.HTTP_200_OK,
responses={
200: {
"description": "Quote updated successfully",
"model": QuoteAdminResponse,
},
404: {
"description": "Quote not found",
},
},
)
def update_quote(
quote_id: UUID,
update_data: QuoteAdminUpdate,
db: Session = Depends(get_db),
current_user: dict = Depends(get_current_user),
):
"""
Update a quote's status or admin notes.
Admins can change the quote status (e.g., from submitted to reviewing
or approved) and add internal notes.
**Example Request:**
```json
PUT /api/admin/quotes/123e4567-e89b-12d3-a456-426614174000
Authorization: Bearer <token>
Content-Type: application/json
{
"status": "reviewing",
"admin_notes": "Assigned to sales team. Follow up scheduled for Monday."
}
```
**Example Response:**
```json
{
"id": "123e4567-e89b-12d3-a456-426614174000",
"status": "reviewing",
"admin_notes": "Assigned to sales team. Follow up scheduled for Monday.",
...
}
```
"""
# Get admin username from token
admin_user = current_user.get("sub", "admin")
quote_service.update_quote_status(
db=db,
quote_id=quote_id,
update_data=update_data,
admin_user=admin_user
)
return get_quote(quote_id, db, current_user)