fk-util-tools

Utility tools and shared functions for FK projects: centralized configuration, logging, middlewares, AWS integration, and OpenTelemetry tracing for Python web frameworks.

Installation

pip install fk-util-tools

Main Features

• Centralized Configuration: Config class for global settings management.
• Cache: Redis integration.
• Environment Variables & AWS: Access to AWS Parameter Store and Secrets Manager.
• Custom Exceptions: HTTP error handling with custom codes.
• Structured Logging: Logging for FastAPI and other frameworks.
• Middlewares: SQL Printer, error handling, internationalization, and language support for Flask, Django, and FastAPI.
• OpenTelemetry Tracing: Instrumentation for Flask, Django, FastAPI, and AWS Lambda.
• Type Hints: PEP 561 support.

Usage Examples

Configuration

from fk_utils import SETTINGS
print(SETTINGS.ENV)

Structured Logging in FastAPI

from fastapi import FastAPI
from fk_utils.logging.fastapi.logging import setup_logging

app = FastAPI()
setup_logging(app)

SQL Printer Middleware

Flask

from flask import Flask
from fk_utils.middlewares.flask.sql_middleware import SqlPrintingMiddleware

app = Flask(__name__)
app.config['DEBUG'] = True
SqlPrintingMiddleware(app)

Django

from fk_utils.middlewares.django.sql_middleware import SqlPrintingMiddleware

SqlPrintingMiddleware()

FastAPI

from fk_utils.middlewares.fastapi.sql_middleware import SqlPrintingMiddleware

app = FastAPI()
SqlPrintingMiddleware(app)

OpenTelemetry Instrumentation

from fk_utils.traces.opentelemetry.fastapi.trace import instrument_app

app = FastAPI()
instrument_app(app)

AWS Lambda Instrumentation

from fk_utils.traces.opentelemetry.fastapi.aws_lambda.trace import instrument_app

app = FastAPI()
is_lambda = instrument_app(app, instrument_lambda=True)
print("Lambda instrumented:", is_lambda)

Health check endpoint

fk_utils.health expone un /health estandarizado para Django, FastAPI y Flask:

from fk_utils.health.django import DjangoHealthCheck
from fk_utils.health import DjangoPostgreSQLCheck, MongoDBCheck

health = DjangoHealthCheck(
    environment=settings.ENV,
    country="CO",
    version=settings.VERSION,
    checks=[DjangoPostgreSQLCheck(), MongoDBCheck(mongo_client)],
)
urlpatterns = [path("health", health.view)]

Respuesta: {"status": "healthy|unhealthy", "environment", "country", "version", "databases": {"postgresql": "healthy", "mongodb": "unhealthy: ..."}}.

Los checks corren en paralelo y la respuesta está acotada en tiempo (2 s por defecto). Si una dependencia se cuelga (p. ej. pymongo reintentando autenticación durante 20 s), el check se reporta como unhealthy: timeout after 2s y el endpoint igual responde: los probes de Kubernetes y el blackbox de monitoreo nunca ven un timeout HTTP por culpa de una dependencia. El presupuesto se ajusta con el parámetro timeout= o la variable FK_HEALTH_CHECK_TIMEOUT; timeout=0 vuelve al comportamiento secuencial sin límite.

Checks propios: subclasear fk_utils.health.base.HealthCheck (ver docstring).

License

MIT