How to Integrate FastAPI for lending with PostgreSQL for RAG

FastAPI for lending gives you the API layer for loan workflows: intake, eligibility checks, document collection, and decisioning. PostgreSQL gives you durable storage for borrower profiles, loan records, and vector-backed retrieval for RAG, which means your agent can answer questions with context instead of guessing.

Prerequisites

• •Python 3.10+
• •A running PostgreSQL 14+ instance
• •fastapi
• •uvicorn
• •psycopg2-binary or psycopg
• •sqlalchemy
• •pgvector extension enabled in PostgreSQL
• •Access to your lending service credentials or internal API base URL
• •Basic knowledge of REST APIs and SQL

Install the Python packages:

pip install fastapi uvicorn sqlalchemy psycopg2-binary pgvector requests pydantic

Enable pgvector in your database:

CREATE EXTENSION IF NOT EXISTS vector;

Integration Steps

1. •

   Define your lending API client and database connection

   Start by separating the external lending API from your storage layer. In most systems, FastAPI is the service exposing lending endpoints, while PostgreSQL stores application state and retrieval data.

from typing import Optional
import os
import requests
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker

LENDING_API_BASE = os.getenv("LENDING_API_BASE", "http://localhost:8000")
DATABASE_URL = os.getenv("DATABASE_URL", "postgresql+psycopg2://postgres:postgres@localhost:5432/lending")

engine = create_engine(DATABASE_URL, pool_pre_ping=True)
SessionLocal = sessionmaker(bind=engine)

class LendingClient:
    def __init__(self, base_url: str):
        self.base_url = base_url.rstrip("/")

    def submit_application(self, payload: dict) -> dict:
        resp = requests.post(f"{self.base_url}/applications", json=payload, timeout=20)
        resp.raise_for_status()
        return resp.json()

    def get_application(self, application_id: str) -> dict:
        resp = requests.get(f"{self.base_url}/applications/{application_id}", timeout=20)
        resp.raise_for_status()
        return resp.json()

lending_client = LendingClient(LENDING_API_BASE)

2. •

   Create tables for lending records and RAG documents

   You need one table for structured loan data and another for retrieval chunks. For RAG, PostgreSQL stores both text and embeddings so the agent can fetch relevant context before answering.

from sqlalchemy import MetaData, Table, Column, String, Text, Integer, DateTime
from sqlalchemy.dialects.postgresql import JSONB
from pgvector.sqlalchemy import Vector

metadata = MetaData()

loan_applications = Table(
    "loan_applications",
    metadata,
    Column("id", String, primary_key=True),
    Column("borrower_name", String, nullable=False),
    Column("status", String, nullable=False),
    Column("amount", Integer, nullable=False),
    Column("payload", JSONB, nullable=False),
)

rag_chunks = Table(
    "rag_chunks",
    metadata,
    Column("id", String, primary_key=True),
    Column("source_type", String, nullable=False),
    Column("source_id", String, nullable=False),
    Column("content", Text, nullable=False),
    Column("embedding", Vector(1536), nullable=False),
)

metadata.create_all(engine)

3. •

   Sync lending events into PostgreSQL

   When a borrower submits an application through FastAPI for lending, persist the response in PostgreSQL. That gives your RAG layer something structured to query later.

import uuid

def sync_application(application_payload: dict) -> dict:
    created = lending_client.submit_application(application_payload)

    with SessionLocal() as db:
        db.execute(
            text("""
                INSERT INTO loan_applications (id, borrower_name, status, amount, payload)
                VALUES (:id, :borrower_name, :status, :amount, :payload::jsonb)
                ON CONFLICT (id) DO UPDATE SET
                    borrower_name = EXCLUDED.borrower_name,
                    status = EXCLUDED.status,
                    amount = EXCLUDED.amount,
                    payload = EXCLUDED.payload
            """),
            {
                "id": created["id"],
                "borrower_name": created["borrower_name"],
                "status": created["status"],
                "amount": created["amount"],
                "payload": created,
            },
        )
        db.commit()

    return created

4. •

   Store embeddings for retrieval

   For RAG you need embeddings. In production you’d generate them with your embedding model of choice; here the important part is how they land in PostgreSQL using pgvector.

from typing import List

def embed_text(text: str) -> List[float]:
    # Replace with your embedding provider.
    # Must return a fixed-length vector matching Vector(1536).
    return [0.0] * 1536

def index_rag_chunk(source_type: str, source_id: str, content: str) -> None:
    vector = embed_text(content)
    chunk_id = str(uuid.uuid4())

    with SessionLocal() as db:
        db.execute(
            text("""
                INSERT INTO rag_chunks (id, source_type, source_id, content, embedding)
                VALUES (:id, :source_type, :source_id, :content, :embedding)
            """),
            {
                "id": chunk_id,
                "source_type": source_type,
                "source_id": source_id,
                "content": content,
                "embedding": vector,
            },
        )
        db.commit()

5. •

   Retrieve context from PostgreSQL before calling the lending API

   This is the core RAG pattern: fetch relevant chunks from Postgres first, then pass that context into your agent or downstream lender workflow.

def retrieve_context(query_embedding: list[float], limit: int = 3):
    with SessionLocal() as db:
        rows = db.execute(
            text("""
                SELECT id, content
                FROM rag_chunks
                ORDER BY embedding <-> :query_embedding
                LIMIT :limit
            """),
            {"query_embedding": query_embedding, "limit": limit},
        ).fetchall()

    return [{"id": r.id if hasattr(r, "id") else r[0], "content": r.content if hasattr(r,"content") else r[1]} for r in rows]

def answer_lending_question(question: str) -> dict:
    q_emb = embed_text(question)
    context = retrieve_context(q_emb)

    app_data = lending_client.get_application(context[0]["id"]) if context else {}
    return {
        "question": question,
        "context": context,
        "application_snapshot": app_data,
    }

Testing the Integration

Run a quick end-to-end check by submitting an application and reading it back through PostgreSQL-backed retrieval.

if __name__ == "__main__":
    sample_application = {
        "borrower_name": "Amina Ndlovu",
        "amount": 25000,
        "income": 85000,
        "term_months": 24,
        "purpose": "Working capital",
    }

    created = sync_application(sample_application)
    print("Created:", created["id"], created["status"])

    index_rag_chunk(
        source_type="loan_application",
        source_id=created["id"],
        content=f"Borrower {created['borrower_name']} applied for {created['amount']} with status {created['status']}.",
    )

    result = answer_lending_question("What is the current status of Amina's loan?")
    print(result["context"][0]["content"])

Expected output:

Created: app_12345 submitted
Borrower Amina Ndlovu applied for 25000 with status submitted.

Real-World Use Cases

• •

  Loan servicing assistant

  • •Answer borrower questions using live application data from FastAPI plus historical policy docs stored in PostgreSQL.
• •

  Underwriting copilot

  • •Retrieve prior decisions and supporting notes from Postgres while your FastAPI service handles application state transitions.
• •

  Collections and support automation

  • •Pull repayment history and call notes from PostgreSQL to generate grounded responses for delinquency workflows.

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.