OpenAPI connector usage example

from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi
from fastapi.responses import Response
from pydantic import BaseModel
from typing import List
import random
import uvicorn
import yaml


class Transaction(BaseModel):
    txn_id: int
    acc_id: int
    txn_value: float
    txn_type: str


class TransactionBatch(BaseModel):
    total_count: int
    items: List[Transaction]


app = FastAPI(title="Transaction API", version="1.0.0")


def generate_transactions(n: int = 10) -> List[Transaction]:
    acc_ids = [1000, 1001, 1002]
    txn_types = ["deposit", "withdrawal", "spb"]
    transactions = []
    for txn_id in range(1, n + 1):
        transactions.append(
            Transaction(
                txn_id=txn_id,
                acc_id=random.choice(acc_ids),
                txn_value=round(random.uniform(0, 500), 2),
                txn_type=random.choice(txn_types),
            )
        )
    return transactions


transactions = generate_transactions(10)


@app.get("/transactions", response_model=List[Transaction])
def get_transactions():
    return transactions


# solely for x-unwrap demonstration
@app.get("/transactions_batch", response_model=TransactionBatch)
def get_transactions_batch():
    return {
        "total_count": len(transactions) / 2,
        "items": transactions[:5],
    }


def custom_openapi():
    if app.openapi_schema:
        return app.openapi_schema

    openapi_schema = get_openapi(
        title=app.title,
        version=app.version,
        description="Transaction API with x-unwrap demo",
        routes=app.routes,
    )

    # Add x-unwrap for /transactions_batch
    openapi_schema["paths"]["/transactions_batch"]["get"]["x-unwrap"] = {
        "resultParam": "$response.body#/items",
        "includeRoot": "false"
    }

    app.openapi_schema = openapi_schema
    return app.openapi_schema


app.openapi = custom_openapi


@app.get("/openapi.yaml", include_in_schema=False)
def get_openapi_yaml():
    openapi_schema = custom_openapi()
    yaml_schema = yaml.dump(openapi_schema, allow_unicode=True,
                            sort_keys=False)
    return Response(content=yaml_schema, media_type="application/yaml")


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8003)

annotated-types==0.7.0
anyio==4.10.0
click==8.2.1
exceptiongroup==1.3.0
fastapi==0.116.1
h11==0.16.0
idna==3.10
pydantic==2.11.7
pydantic_core==2.33.2
PyYAML==6.0.2
sniffio==1.3.1
starlette==0.47.2
typing-inspection==0.4.1
typing_extensions==4.14.1
uvicorn==0.35.0