surquest-fastapi-utils 0.2.9

This project provides collection of utilities for FastAPI framework as: Catcher, Middleware, etc.

Introduction

This project provides collection of utilities for smooth integration of FastAPI framework with Google Cloud Platform services as logging and tracing.

The key features of this project are:

• Logging to Cloud Logging
• Tracing to Cloud Logging
• Error Reporting via Cloud Logging
• Custom middleware for configuration of logging
• Custom exception handlers treating HTTP and validation exceptions
• Custom routes for documentation and favicon
• Custom responses with statuses success, warning and error and standardized error messages

Quick Start

This section shows how to use the utilities provided by this project:

"""File main.py with FastAPI app"""
import os
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException
from fastapi import FastAPI, Request, Query

# import surquest modules and objects
from surquest.fastapi.utils.route import Route  # custom routes for documentation and FavIcon
from surquest.fastapi.utils.GCP.tracer import Tracer
from surquest.fastapi.utils.GCP.logging import Logger
from surquest.fastapi.schemas.responses import Response
from surquest.fastapi.utils.GCP.middleware import LoggingMiddleware
from surquest.fastapi.utils.GCP.catcher import (
    catch_validation_exceptions,
    catch_http_exceptions,
)

PATH_PREFIX = os.getenv('PATH_PREFIX','')

app = FastAPI(
    title="Exchange Rates ETL",
    openapi_url=F"{PATH_PREFIX}/openapi.json"
)

# add middleware
app.add_middleware(LoggingMiddleware)

# exception handlers
app.add_exception_handler(HTTPException, catch_http_exceptions)
app.add_exception_handler(RequestValidationError, catch_validation_exceptions)

# custom routes to documentation and favicon
app.add_api_route(path=F"{PATH_PREFIX}/", endpoint=Route.get_documentation, include_in_schema=False)
app.add_api_route(path=PATH_PREFIX, endpoint=Route.get_favicon, include_in_schema=False)

# custom route to illustrate logging and tracing
@app.get(F"{PATH_PREFIX}/users")
async def get_users(
    age: int = Query(
        default=18,
        description="Minimal age of the user",
        example=30,

    ),
):

    with Tracer.start_span("Generate users"):

        users = [
            {"name": "John Doe", "age": 30, "email": "[e-mail]"},
            {"name": "Will Smith", "age": 42, "email": "[e-mail]"}
        ]

        Logger.info('Found %s users', len(users), extra={"users": users})

    with Tracer.start_span("Filtering users"):

        output = []
        excluded = []
        Logger.debug(F"Filtering users by age > {age}")

        for user in users:

            if user["age"] > age:
                output.append(user)
            else:
                excluded.append(user)

        Logger.debug(
            'Number of excluded users: %s', len(excluded),
            extra={"excluded": excluded}
        )

    return Response.set(data=output)

The endpoint /users will return the following standard response:

{
  "info": {
    "status": "success"
  },
  "data": [
    {
      "name": "John Doe",
      "age": 30,
      "email": "[e-mail]"
    },
    {
      "name": "Will Smith",
      "age": 42,
      "email": "[e-mail]"
    }
  ]
}

and the logs will are available in Google Cloud Platform console within Stackdriver Logging:

as well as the traces are available in Google Cloud Platform console within Stackdriver Trace:

API Gateway Integration

This package also provides a FastAPI extension that generates OpenAPI specifications compatible with Google Cloud API Gateway. By default, it generates an OpenAPI 3.0.0 specification, which is the current standard. If you are using legacy systems, you can also output a Swagger 2.0 (OpenAPI 2.0) specification by passing openapi_version="2.0" to the gateway options.

from fastapi import FastAPI
from pydantic import BaseModel
from surquest.fastapi.utils.GCP.api_gateway import (
    GCPGateway, cloud_run_backend, SecurityDefinition,
)

app = FastAPI(title="My Service", version="1.0.0")


class Item(BaseModel):
    name: str
    price: float


@app.get("/items/{item_id}", tags=["public", "items"])
def get_item(item_id: int):
    return {"id": item_id}


@app.post("/items", tags=["admin", "items"])
def create_item(item: Item):
    return item


@app.get("/internal/health", tags=["internal"])
def health():
    return {"ok": True}


# OpenAPI 3.0.0 output by default:
GCPGateway(
    app,
    backend=cloud_run_backend(
        address="https://my-service-abc-uc.a.run.app",
        deadline=30.0,
    ),
    host="my-gateway-xxxx.uc.gateway.dev",
    security_definitions=[
        SecurityDefinition(name="api_key", in_="query", key_name="key"),
    ],
    default_security=[{"api_key": []}],
    # openapi_version="2.0" # Option for Swagger 2.0 output
)

Generated Endpoints

The extension automatically creates the following endpoints:

• GET /spec/apiGateway/yaml — full spec
• GET /spec/apiGateway/json — full spec
• GET /spec/apiGateway/yaml?include_tags=public&include_tags=items — only public+items endpoints
• GET /spec/apiGateway/yaml?exclude_tags=internal — exclude internal endpoints
• GET /spec/apiGateway/yaml?exclude_responses=true — exclude responses and nested component schema definitions (creates a lightweight spec)
• GET /spec/apiGateway/yaml?exclude_parameters_schemas=true — exclude all schemas for parameters and request body (simplifies them to basic string/object)

You can then deploy it with gcloud by extracting the output of the spec:

curl "http://localhost:8000/spec/apiGateway/yaml?exclude_tags=internal&exclude_responses=true&exclude_parameters_schemas=true" > openapi.yaml

gcloud api-gateway api-configs create my-config \
  --api=my-api --openapi-spec=openapi.yaml \
  --backend-auth-service-account=[e-mail]

Local development

You are more than welcome to contribute to this project. To make your start easier we have prepared a docker image with all the necessary tools to run it as interpreter for Pycharm or to run tests.

Build docker image

docker build `
     --tag surquest/fastapi/utils `
     --file package.base.dockerfile `
     --target test .

Run tests

docker run --rm -it `
 -v "${pwd}:/opt/project" `
 -e "GOOGLE_APPLICATION_CREDENTIALS=/opt/project/credentials/keyfile.json" `
 -w "/opt/project/test" `
 surquest/fastapi/utils pytest