GCM Diagnostics

Library for collecting errors from validation logic and presenting them to user.

Mainly intended to be used with FastAPI endpoints for doing business validation which Pydantic is not intended for. Errors generated by this library are compatible with Pydantic validation response, so it appears the same to the consumer of API.

Installation

pip install gcm_diagnostics

# or

poetry add gcm_diagnostics

Usage

from gcm_diagnostics import DiagnosticCollector
from gcm_diagnostics.errors import EntityNotFound, EntityAlreadyExists

with DiagnosticCollector(prefix=["body"]) as diag:
    diag.append(EntityNotFound(loc=[1, "id"], id=1))
    diag.append(EntityAlreadyExists(loc=[2, "name"], entity="Hello"))

# Here, DiagnosticException is raised, containing both errors.

print(diag.errors)
# [
#     {
#         "loc": ["body", 1, "id"],
#         "msg": "Entity with id 1 not found",
#         "type": "entity_not_found",
#         "id": 1
#     },
#     {
#         "loc": ["body", 2, "name"],
#         "msg": "Entity already exists",
#         "type": "entity_already_exists",
#         "entity": "Hello"
#     }
# ]

Usage with FastAPI

from fastapi import FastAPI
from gcm_diagnostics import install_exception_handler, DiagnosticCollector, diagnostic_schema
from gcm_diagnostics.errors import EntityNotFound
from starlette.testclient import TestClient

app = FastAPI()

# Install exception handler for the DiagnosticError exception.  
install_exception_handler(app)

@app.get(
    "/",
    # This builds responses structure containing documentation for the specified
    # error types this endpoint produces instead of generic error description.
    # Usage of diagnostic_schema() is not required, it is just a convenient
    # helper to build better documentation even for error states.
    responses=diagnostic_schema([EntityNotFound])
)
def diagnostics(id: int) -> None:
    # Instantiate DiagnosticCollector(), that will collect all diagnostic data from
    # validation logic.
    # In this case, all errors will be prefixed with "query" prefix.
    with DiagnosticCollector(prefix=["query"]) as diag:
        # Append new error to the collector. This does not raise exception immediately,
        # it allows to collect multiple errors at one shot, and present it all to the user.
        # Resulting loc for this error will be ["query", "id"].
        diag.append(EntityNotFound(loc=["id"], id=id))
        
        # If you want to raise exception in the middle of validation logic, you can
        # use raise_if_errors() as follows:
        #diag.raise_if_errors()


with TestClient(app, raise_server_exceptions=False) as client:
    response = client.get("/", params={"id": 123})
    
    # Error code is gathered from diagnostics. If there are only diagnostics of one
    # type (with one status code), this status code is returned.
    # If there are multiple different status codes, generic 422 is returned.
    assert response.status_code == 404
    
    # Response is generated from collected diagnostics. All errors are present in the detail array.
    assert response.json() == {
        "detail": [
            {
                "loc": ["query", "id"],
                "msg": "Entity does not exists.",
                "type": "entity_not_found",
                "id": 123
            }
        ]
    }

For more usage details, please consult the documentation in the code, mainly the documentation of class DiagnosticCollector.

Maturity

This library is currently used in various production applications and is considered stable.

Contributing

Contributions are always welcome. Just open an issue or a merge request.