How to Integrate FastAPI for payments with PostgreSQL for startups

FastAPI for payments plus PostgreSQL is a solid default for startup systems that need to accept money, persist transaction state, and expose clean APIs to an AI agent. The pattern is simple: FastAPI handles payment orchestration at the edge, PostgreSQL stores durable payment records, idempotency keys, and audit trails the agent can reason over later.

Prerequisites

• •Python 3.10+
• •A running PostgreSQL instance
• •pip or uv
• •A FastAPI project scaffold
• •Payment provider credentials for your gateway of choice
• •Basic familiarity with:
  • •fastapi.FastAPI
  • •PostgreSQL connection strings
  • •SQLAlchemy or asyncpg
• •Environment variables configured:
  • •DATABASE_URL
  • •PAYMENT_API_KEY

Integration Steps

1. •

   Install the core packages

   Use FastAPI for the API layer and SQLAlchemy for PostgreSQL access. If you want async I/O, use asyncpg under SQLAlchemy’s async engine.

   pip install fastapi uvicorn sqlalchemy asyncpg pydantic python-dotenv
   

2. •

   Create the PostgreSQL connection layer

   Define a database engine and session factory. This gives your payment service a durable place to store payment intents, status updates, and webhook events.

   from sqlalchemy import create_engine, Column, Integer, String, DateTime, Numeric
   from sqlalchemy.orm import declarative_base, sessionmaker
   from sqlalchemy.sql import func
   import os
   
   DATABASE_URL = os.getenv("DATABASE_URL", "postgresql+asyncpg://user:pass@localhost:5432/payments_db")
   
   Base = declarative_base()
   
   class Payment(Base):
       __tablename__ = "payments"
   
       id = Column(Integer, primary_key=True)
       customer_id = Column(String(64), nullable=False)
       payment_provider_id = Column(String(128), unique=True, nullable=False)
       amount = Column(Numeric(12, 2), nullable=False)
       currency = Column(String(8), nullable=False)
       status = Column(String(32), nullable=False)
       created_at = Column(DateTime(timezone=True), server_default=func.now())
   
   # For sync setups; switch to AsyncEngine/AsyncSession if needed.
   engine = create_engine(DATABASE_URL.replace("+asyncpg", ""), echo=True)
   SessionLocal = sessionmaker(bind=engine)
   
   Base.metadata.create_all(bind=engine)
   

3. •

   Build the FastAPI payment endpoint

   This endpoint accepts a payment request, calls your payment provider SDK/API, then persists the result in PostgreSQL. The important part is that the API response and database write happen in a controlled flow.

   from fastapi import FastAPI, HTTPException, Depends
   from pydantic import BaseModel
   from sqlalchemy.orm import Session
   import os
   
   app = FastAPI()
   
   class PaymentRequest(BaseModel):
       customer_id: str
       amount: float
       currency: str = "USD"
       payment_method_id: str
   
   class PaymentResponse(BaseModel):
       payment_id: str
       status: str
   
   def get_db():
       db = SessionLocal()
       try:
           yield db
       finally:
           db.close()
   
   @app.post("/payments", response_model=PaymentResponse)
   async def create_payment(payload: PaymentRequest, db: Session = Depends(get_db)):
       # Example call shape; replace with your actual provider SDK method.
       # For Stripe-style integrations this is typically:
       # stripe.PaymentIntent.create(amount=..., currency=..., payment_method=..., confirm=True)
       provider_payment = {
           "id": f"pi_{payload.customer_id}_001",
           "status": "succeeded"
       }
   
       record = Payment(
           customer_id=payload.customer_id,
           payment_provider_id=provider_payment["id"],
           amount=payload.amount,
           currency=payload.currency,
           status=provider_payment["status"],
       )
       db.add(record)
       db.commit()
       db.refresh(record)
   
       return PaymentResponse(payment_id=record.payment_provider_id, status=record.status)
   

4. •

   Add webhook handling for provider updates

   Real payment systems do not rely only on synchronous API responses. You need webhooks so your database stays aligned with provider-side events like succeeded, failed, or refunded.

   from fastapi import Request
   
   @app.post("/webhooks/payments")
   async def payment_webhook(request: Request, db: Session = Depends(get_db)):
       event = await request.json()
   
       # Example event shape; validate signature in production.
       event_type = event.get("type")
       payment_id = event.get("data", {}).get("object", {}).get("id")
   
       if event_type == "payment_intent.succeeded":
           payment = db.query(Payment).filter_by(payment_provider_id=payment_id).first()
           if not payment:
               raise HTTPException(status_code=404, detail="Payment not found")
   
           payment.status = "succeeded"
           db.commit()
   
       return {"received": True}
   

5. •

   Expose a read endpoint for your AI agent

   Your agent needs state it can trust. A simple query endpoint lets it check whether a user has paid before unlocking an action or generating a receipt.

    @app.get("/payments/{payment_provider_id}")
    def get_payment(payment_provider_id: str, db: Session = Depends(get_db)):
        payment = db.query(Payment).filter_by(payment_provider_id=payment_provider_id).first()
        if not payment:
            raise HTTPException(status_code=404, detail="Payment not found")
   
        return {
            "payment_provider_id": payment.payment_provider_id,
            "customer_id": payment.customer_id,
            "amount": float(payment.amount),
            "currency": payment.currency,
            "status": payment.status,
        }
   

Testing the Integration

Run the app:

uvicorn main:app --reload

Then test it with a POST request:

import requests

response = requests.post(
    "http://127.0.0.1:8000/payments",
    json={
        "customer_id": "cust_123",
        "amount": 49.99,
        "currency": "USD",
        "payment_method_id": "pm_test_001"
    }
)

print(response.status_code)
print(response.json())

Expected output:

200
{'payment_id': 'pi_cust_123_001', 'status': 'succeeded'}

You can also verify persistence directly in PostgreSQL:

SELECT customer_id, payment_provider_id, amount, currency, status
FROM payments;

Real-World Use Cases

• •

  Subscription billing for AI agents

  • •Let an agent check whether a startup user has an active paid plan before allowing premium workflows.
• •

  Usage-based invoicing

  • •Store token usage or task completion counts in PostgreSQL and charge through FastAPI when thresholds are reached.
• •

  Receipt and audit pipelines

  • •Keep immutable transaction records in PostgreSQL so finance teams can reconcile provider events with internal agent actions.

Keep learning

• •The complete AI Agents Roadmap — my full 8-step breakdown
• •Free: The AI Agent Starter Kit — PDF checklist + starter code
• •Work with me — I build AI for banks and insurance companies

By Cyprian Aarons, AI Consultant at Topiax.