pylib Package

This is a py utils lib package, here are log and lots of methods. All methods you can see in file methods.py.

HTTP request metrics

HttpRequestMetricsMiddleware records the medlink_http_request_duration_seconds histogram independently of HttpRequestLoggingMiddleware. Install each middleware separately, and start the metrics-only listener from the application's lifespan:

from contextlib import asynccontextmanager

from fastapi import FastAPI
from pylib.gcp_logging import HttpRequestLoggingMiddleware
from pylib.http_metrics import HttpRequestMetricsMiddleware, start_http_metrics_server


@asynccontextmanager
async def lifespan(app: FastAPI):
    metrics_server = start_http_metrics_server()
    try:
        yield
    finally:
        metrics_server.close()


app = FastAPI(lifespan=lifespan)
app.add_middleware(HttpRequestLoggingMiddleware)
app.add_middleware(HttpRequestMetricsMiddleware, service_tier="api")

The listener uses TCP port 9000 by default and serves only GET /metrics; application traffic, Services, and probes can remain on port 8080. Set PROMETHEUS_MULTIPROC_DIR before importing the module or starting workers. Give each pod a writable emptyDir shared by that pod's workers. The listener returns 503 when this directory is not configured instead of exposing incomplete worker-local data.

Labels use the Starlette route template (or /__unmatched__), a standard method or OTHER, status class, service identity from MEDLINK_SERVICE_ID/SERVICE_ID/SERVICE_NAME, service_tier, and APP_ENV (production normalizes to prod). Successful health probes and /metrics scrapes are excluded; raw paths, query strings, request bodies, and identifiers are never label values. The module uses Starlette ASGI types and the package requires Python >=3.10.

Usage

from pylib.log import log
log.debug('msg')
log.info('msg')
log.warning('msg')
log.error('msg')
log.critical('msg')
log.log('DEBUG', 'msg')
log.log('INFO', 'msg')
log.log('WARNING', 'msg')
log.log('ERROR', 'msg')
log.log('CRITICAL', 'msg', silence=True)

from pylib.methods import Methods
Methods.get_stack_funcs()
data_desensitived = Methods.mask_sensitive_data(dict_data)

Worker services can use the standard logging API with a logger name inferred from the calling module:

from pylib import logger

logger.info("worker_started")

Importing pylib.logger initializes GCP worker logging once for the current process. configure_worker_logging() remains available for explicit setup; repeated calls in the same process are ignored, while a forked child configures independently.