package-events-tracking

Librería compartida de tracking de eventos de producto para microservicios Finkargo. Sigue las mismas convenciones que fk_util_tools (fk_utils): empaquetado Poetry + MIT, configuración vía Config/SETTINGS, credenciales vía AWS Secrets Manager (no env vars planas), logging estándar y type hints consistentes. El módulo Python se sigue importando como fk_tracking.

Instalación

poetry add package-events-tracking

Main Features

• Decorador @track_endpoint: instrumenta cualquier endpoint de FastAPI capturando producto, categoría, usuario, duración, estado y metadata flexible — sin tocar la lógica de negocio.
• track_event / @track_on_success: para tracking condicional o post-ejecución — un helper async standalone y un decorador que solo trackea si la función decorada retorna exitosamente (nunca en error), con metadata_fn(result, kwargs) opcional para metadata dinámica.
• Fire-and-forget: un fallo de tracking (repositorio no configurado, DB caída, etc.) nunca rompe el endpoint decorado — se loguea con logger.warning y se descarta.
• Persistencia desacoplada: interface TrackingRepository + adaptador PostgresTrackingRepository (schema events_tracking, tabla tracking_events, metadata JSONB indexado con GIN).
• Configuración centralizada: TrackingConfig.set_repository()/get_repository() + resolución de credenciales vía AWS Secrets Manager (get_tracking_connection_string()).
• Analítica opcional: router de FastAPI (fk_tracking.fastapi.tracking_analytics_router) con endpoints de consulta por producto, usuarios activos y análisis de metadata.

Uso rápido

from fastapi import FastAPI
from fk_tracking import TrackingConfig, track_endpoint
from fk_tracking.config import get_tracking_connection_string
from fk_tracking.repository.postgres_adapter import PostgresTrackingRepository

app = FastAPI()


@app.on_event("startup")
async def configure_tracking():
    connection_string = get_tracking_connection_string()
    if connection_string:
        TrackingConfig.set_repository(PostgresTrackingRepository(connection_string))


@app.get("/cotizaciones")
@track_endpoint(producto="dmp", category="consulta", metadata={"modulo": "cotizaciones"})
async def get_cotizaciones():
    return {"cotizaciones": []}

Tracking condicional o post-ejecución

Para los casos que @track_endpoint no cubre (trackear solo en éxito, o solo bajo una condición interna que no es "la función completó sin excepción"):

from fk_tracking import track_event, track_on_success

# Decorador: trackea solo si la función retorna exitosamente
@track_on_success(
    producto="dmp",
    category=ImportEventCategory.SHIPMENT_CREATED.value,
    metadata_fn=lambda result, kwargs: {"shipment_id": result.id},
    user_id_kwarg="company_uuid",
)
async def create_shipment(company_uuid: str, ...):
    ...


# Función standalone: el caller decide cuándo trackear tras inspeccionar su propio estado
async def process_import(company_uuid: str, shipment):
    result = do_something(shipment)
    if result.requires_manual_review:
        await track_event(
            producto="dmp",
            category=ImportEventCategory.MANUAL_REVIEW.value,
            endpoint="process_import",
            user_id=company_uuid,
            metadata={"shipment_id": shipment.id},
        )

Desarrollo

poetry install
poetry run pytest
poetry run ruff check .

Migraciones

El schema events_tracking y la tabla tracking_events se gestionan en data-bbdd-cross (no en este repo), como cualquier otra tabla de Finkargo:

• rds_co/Finkargo/changes/000810..000812_..._LIBARDO_CUELLO.json
• rds_mx/Finkargo/changes/000792..000794_..._LIBARDO_CUELLO.json

TrackingEvent (fk_tracking/repository/postgres_adapter.py) debe mantenerse en sync con esos changesets, no al revés — la fuente de verdad del schema es Liquibase en data-bbdd-cross. La PK es event_id UUID DEFAULT gen_random_uuid() (no serial), y los timestamps son TIMESTAMP WITH TIME ZONE, consistente con el resto de tablas de ese repo.