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": "john@doe.com"},
            {"name": "Will Smith", "age": 42, "email": "will@smith.com"}
        ]

        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": "john@doe.com"
    },
    {
      "name": "Will Smith",
      "age": 42,
      "email": "will@smith.com"
    }
  ]
}

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:

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