diegosouzapw API Scaffolding for SMB Automation
Spin up professional REST APIs in minutes for client deliverables.
When to Use This Skill
- Creating new API endpoints for client workflows
- Building webhook receivers (Stripe, ShipStation, etc.)
- Setting up FastAPI projects from scratch
- Adding endpoints to existing
execution/scripts
Quick Start: New Endpoint
1. Define the Pydantic Models
# execution/schemas/invoice.py
from pydantic import BaseModel, EmailStr, Field
from typing import Optional
from datetime import datetime
class LineItem(BaseModel):
description: str
quantity: int = Field(gt=0)
unit_price: float = Field(gt=0)
class CreateInvoiceRequest(BaseModel):
customer_email: EmailStr
customer_name: str = Field(min_length=1, max_length=100)
line_items: list[LineItem] = Field(min_length=1)
due_date: datetime
memo: Optional[str] = None
class InvoiceResponse(BaseModel):
invoice_id: str
qbo_id: Optional[str] = None
status: str
total: float
created_at: datetime
2. Create the Endpoint
# execution/api/invoices.py
from fastapi import APIRouter, HTTPException, Depends
from ..schemas.invoice import CreateInvoiceRequest, InvoiceResponse
from ..services.qbo_service import qbo_service
import logging
router = APIRouter(prefix="/invoices", tags=["invoices"])
log = logging.getLogger(__name__)
@router.post("/", response_model=InvoiceResponse, status_code=201)
async def create_invoice(request: CreateInvoiceRequest):
"""Create a new invoice in QuickBooks."""
try:
# Calculate total
total = sum(item.quantity * item.unit_price for item in request.line_items)
# Create in QBO
qbo_invoice = await qbo_service.create_invoice(
customer_email=request.customer_email,
customer_name=request.customer_name,
line_items=[item.model_dump() for item in request.line_items],
due_date=request.due_date,
memo=request.memo
)
log.info("Invoice created", extra={
"invoice_id": qbo_invoice.id,
"customer": request.customer_email,
"total": total
})
return InvoiceResponse(
invoice_id=qbo_invoice.doc_number,
qbo_id=qbo_invoice.id,
status="created",
total=total,
created_at=datetime.utcnow()
)
except QBOAuthError:
raise HTTPException(status_code=401, detail="QuickBooks auth expired")
except QBOValidationError as e:
raise HTTPException(status_code=422, detail=str(e))
except Exception as e:
log.exception("Invoice creation failed")
raise HTTPException(status_code=500, detail="Internal error")
3. Wire Up the Router
# execution/main.py
from fastapi import FastAPI
from .api import invoices, webhooks, inventory
app = FastAPI(
title="SMB Automation API",
version="1.0.0"
)
app.include_router(invoices.router, prefix="/api/v1")
app.include_router(webhooks.router, prefix="/api/v1")
app.include_router(inventory.router, prefix="/api/v1")
@app.get("/health")
async def health():
return {"status": "ok"}
Common Endpoint Patterns
Webhook Receiver
# execution/api/webhooks.py
from fastapi import APIRouter, Request, HTTPException, Header
import hmac
import hashlib
router = APIRouter(prefix="/webhooks", tags=["webhooks"])
@router.post("/stripe")
async def stripe_webhook(
request: Request,
stripe_signature: str = Header(alias="Stripe-Signature")
):
"""Handle Stripe webhook events."""
payload = await request.body()
# Verify signature
try:
event = stripe.Webhook.construct_event(
payload, stripe_signature, WEBHOOK_SECRET
)
except stripe.error.SignatureVerificationError:
raise HTTPException(status_code=400, detail="Invalid signature")
# Process idempotently
event_id = event["id"]
if await is_processed(event_id):
return {"status": "duplicate"}
# Handle event types
match event["type"]:
case "payment_intent.succeeded":
await handle_payment_success(event["data"]["object"])
case "invoice.paid":
await handle_invoice_paid(event["data"]["object"])
await mark_processed(event_id)
return {"status": "processed"}
CRUD Resource
# execution/api/customers.py
from fastapi import APIRouter, HTTPException, Query
from typing import Optional
router = APIRouter(prefix="/customers", tags=["customers"])
@router.get("/")
async def list_customers(
skip: int = Query(0, ge=0),
limit: int = Query(50, ge=1, le=100),
status: Optional[str] = None
):
"""List customers with pagination."""
customers = await customer_service.list(skip=skip, limit=limit, status=status)
return {"customers": customers, "total": len(customers)}
@router.get("/{customer_id}")
async def get_customer(customer_id: str):
"""Get customer by ID."""
customer = await customer_service.get(customer_id)
if not customer:
raise HTTPException(status_code=404, detail="Customer not found")
return customer
@router.post("/", status_code=201)
async def create_customer(request: CreateCustomerRequest):
"""Create new customer."""
customer = await customer_service.create(request)
return customer
@router.patch("/{customer_id}")
async def update_customer(customer_id: str, request: UpdateCustomerRequest):
"""Update customer."""
customer = await customer_service.update(customer_id, request)
if not customer:
raise HTTPException(status_code=404, detail="Customer not found")
return customer
@router.delete("/{customer_id}", status_code=204)
async def delete_customer(customer_id: str):
"""Delete customer."""
deleted = await customer_service.delete(customer_id)
if not deleted:
raise HTTPException(status_code=404, detail="Customer not found")
Background Task Trigger
# execution/api/jobs.py
from fastapi import APIRouter, BackgroundTasks
router = APIRouter(prefix="/jobs", tags=["jobs"])
@router.post("/sync-inventory")
async def trigger_inventory_sync(background_tasks: BackgroundTasks):
"""Trigger background inventory sync."""
job_id = generate_job_id()
background_tasks.add_task(sync_inventory_task, job_id)
return {
"job_id": job_id,
"status": "queued",
"check_status": f"/api/v1/jobs/{job_id}"
}
@router.get("/{job_id}")
async def get_job_status(job_id: str):
"""Check job status."""
job = await job_store.get(job_id)
if not job:
raise HTTPException(status_code=404, detail="Job not found")
return job
Project Structure
execution/
├── main.py # FastAPI app entry
├── api/
│ ├── __init__.py
│ ├── invoices.py # Invoice endpoints
│ ├── webhooks.py # Webhook receivers
│ ├── customers.py # Customer CRUD
│ └── jobs.py # Background job triggers
├── schemas/
│ ├── __init__.py
│ ├── invoice.py # Invoice Pydantic models
│ └── customer.py # Customer Pydantic models
├── services/
│ ├── __init__.py
│ ├── qbo_service.py # QuickBooks API client
│ └── stripe_service.py # Stripe API client
└── core/
├── config.py # Settings from .env
├── database.py # DB connection (if needed)
└── security.py # Auth helpers
Essential Dependencies
# requirements.txt
fastapi>=0.109.0
uvicorn>=0.27.0
pydantic>=2.5.0
pydantic-settings>=2.1.0
httpx>=0.26.0
python-dotenv>=1.0.0
Running Locally
# Install
pip install -r requirements.txt
# Run
uvicorn execution.main:app --reload --port 8000
# Docs
open http://localhost:8000/docs
Deployment Checklist
- All endpoints have Pydantic request/response models
- Error handling returns proper status codes
- Logging includes context (IDs, user info)
- Secrets in
.env, not hardcoded - Health endpoint at
/health - CORS configured if needed
- Rate limiting for public endpoints