fastapi-service

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

FastAPI Service

FastAPI 服务

Build production-grade FastAPI services with JWT authentication, structured logging via structlog, and Prometheus metrics.

构建具备JWT认证、基于structlog的结构化日志以及Prometheus指标监控的生产级FastAPI服务。

Quick Start

快速开始

python

import os
import sys
import logging
from contextlib import asynccontextmanager
from typing import Dict

import structlog
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from jose import JWTError, jwt
from prometheus_fastapi_instrumentator import Instrumentator

python

import os
import sys
import logging
from contextlib import asynccontextmanager
from typing import Dict

import structlog
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from jose import JWTError, jwt
from prometheus_fastapi_instrumentator import Instrumentator

-------------------------

Logging configuration

-------------------------

def configure_logging() -> None: log_level_name = os.getenv("LOG_LEVEL", "INFO").upper() log_level = getattr(logging, log_level_name, logging.INFO) json_logs = os.getenv("LOG_JSON", "false").lower() == "true" service_name = os.getenv("SERVICE_NAME", "fastapi-app")

logging.basicConfig(level=log_level, stream=sys.stdout)

processors = [
    structlog.contextvars.merge_contextvars,
    structlog.processors.add_log_level,
    structlog.processors.TimeStamper(fmt="iso"),
    structlog.processors.StackInfoRenderer(),
    structlog.processors.format_exc_info,
]
if json_logs:
    processors.append(structlog.processors.JSONRenderer())
else:
    processors.append(structlog.dev.ConsoleRenderer())

structlog.configure(
    processors=processors,
    wrapper_class=structlog.make_filtering_bound_logger(log_level),
    logger_factory=structlog.PrintLoggerFactory(),
    cache_logger_on_first_use=True,
)

structlog.contextvars.bind_contextvars(service=service_name)

configure_logging() log = structlog.get_logger()

logging.basicConfig(level=log_level, stream=sys.stdout)

processors = [
    structlog.contextvars.merge_contextvars,
    structlog.processors.add_log_level,
    structlog.processors.TimeStamper(fmt="iso"),
    structlog.processors.StackInfoRenderer(),
    structlog.processors.format_exc_info,
]
if json_logs:
    processors.append(structlog.processors.JSONRenderer())
else:
    processors.append(structlog.dev.ConsoleRenderer())

structlog.configure(
    processors=processors,
    wrapper_class=structlog.make_filtering_bound_logger(log_level),
    logger_factory=structlog.PrintLoggerFactory(),
    cache_logger_on_first_use=True,
)

structlog.contextvars.bind_contextvars(service=service_name)

configure_logging() log = structlog.get_logger()

-------------------------

Auth configuration

-------------------------

JWT_SECRET = os.getenv("AUTH_SECRET", "change-me") JWT_ALG = os.getenv("JWT_ALG", "HS256") security = HTTPBearer(auto_error=True)

def require_claims( credentials: HTTPAuthorizationCredentials = Depends(security), ) -> Dict: token = credentials.credentials try: claims = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALG]) except JWTError: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid or expired token", headers={"WWW-Authenticate": "Bearer"}, ) return claims

JWT_SECRET = os.getenv("AUTH_SECRET", "change-me") JWT_ALG = os.getenv("JWT_ALG", "HS256") security = HTTPBearer(auto_error=True)

-------------------------

App & instrumentation

-------------------------

@asynccontextmanager async def lifespan(app: FastAPI): log.info("service.startup") try: yield finally: log.info("service.shutdown")

app = FastAPI(title=os.getenv("SERVICE_NAME", "fastapi-app"), lifespan=lifespan)

@asynccontextmanager async def lifespan(app: FastAPI): log.info("service.startup") try: yield finally: log.info("service.shutdown")

app = FastAPI(title=os.getenv("SERVICE_NAME", "fastapi-app"), lifespan=lifespan)

Prometheus: exposes /metrics by default

Instrumentator().instrument(app).expose(app)

@app.get("/healthz", tags=["health"]) def healthz(): return {"status": "ok"}

@app.get("/whoami", tags=["auth"]) def whoami(claims: Dict = Depends(require_claims)): return {"claims": claims}

if name == "main": import uvicorn

host = os.getenv("HOST", "0.0.0.0")
port = int(os.getenv("PORT", "8000"))
uvicorn.run("main:app", host=host, port=port, reload=False)

undefined

Instrumentator().instrument(app).expose(app)

@app.get("/healthz", tags=["health"]) def healthz(): return {"status": "ok"}

@app.get("/whoami", tags=["auth"]) def whoami(claims: Dict = Depends(require_claims)): return {"claims": claims}

if name == "main": import uvicorn

host = os.getenv("HOST", "0.0.0.0")
port = int(os.getenv("PORT", "8000"))
uvicorn.run("main:app", host=host, port=port, reload=False)

undefined

Dependencies

依赖项

bash

uv add fastapi "uvicorn[standard]" "python-jose[cryptography]" prometheus-fastapi-instrumentator structlog

bash

uv add fastapi "uvicorn[standard]" "python-jose[cryptography]" prometheus-fastapi-instrumentator structlog

Environment

环境配置

Create

.env.server

export AUTH_SECRET="your-strong-shared-secret"
export LOG_LEVEL=INFO
export LOG_JSON=false

创建

.env.server

export AUTH_SECRET="your-strong-shared-secret"
export LOG_LEVEL=INFO
export LOG_JSON=false

Makefile

SHELL := /bin/bash

define setup_env
        $(eval ENV_FILE := $(1))
        $(eval include $(1))
        $(eval export)
endef

run-server:
    $(call setup_env, .env.server)
    uv run uvicorn main:app --host 0.0.0.0 --port 8000

Makefile

SHELL := /bin/bash

define setup_env
        $(eval ENV_FILE := $(1))
        $(eval include $(1))
        $(eval export)
endef

run-server:
    $(call setup_env, .env.server)
    uv run uvicorn main:app --host 0.0.0.0 --port 8000

Testing Auth

认证测试

Generate a test token:

bash

uv run python - <<'PY'
from jose import jwt
print(jwt.encode({"sub":"alice","role":"admin"}, "your-strong-shared-secret", algorithm="HS256"))
PY

Call the authenticated endpoint:

bash

curl -H "Authorization: Bearer <token>" http://localhost:8000/whoami

生成测试令牌：

bash

uv run python - <<'PY'
from jose import jwt
print(jwt.encode({"sub":"alice","role":"admin"}, "your-strong-shared-secret", algorithm="HS256"))
PY

调用需认证的接口：

bash

curl -H "Authorization: Bearer <token>" http://localhost:8000/whoami

Key Patterns

核心模式

Lifespan: Use
```
@asynccontextmanager
```
for startup/shutdown, not deprecated
```
on_event
```
Dependency injection: Use
```
Depends()
```
for auth, DB connections, and shared resources
Structured logging: Configure structlog once at module level; bind context vars per-request
Metrics:
```
prometheus-fastapi-instrumentator
```
auto-instruments all routes
Health check: Always expose
```
/healthz
```
for k8s readiness/liveness probes

Lifespan: 使用
```
@asynccontextmanager
```
处理启动/关闭逻辑，而非已弃用的
```
on_event
```
Dependency injection: 使用
```
Depends()
```
实现认证、数据库连接及共享资源的依赖注入
Structured logging: 在模块级别统一配置structlog；为每个请求绑定上下文变量
Metrics:
```
prometheus-fastapi-instrumentator
```
会自动为所有路由添加监控埋点
Health check: 始终暴露
```
/healthz
```
接口，用于K8s的就绪/存活探针

Bookkeeping

后续维护

After modifying the server, update the adjacent README.md to reflect the current API surface.

修改服务器代码后，请更新相邻的README.md文件，以反映当前的API接口情况。