software

为企业 FastAPI 应用构建 10 个 Python 包：我的收获

发布: (2025年12月17日 GMT+8 08:20)
6 min read
原文: Dev.to

Source: Dev.to – Building 10 Python Packages for Enterprise FastAPI Apps: What I Learned

“构建 10 个用于企业 FastAPI 应用的 Python 包：我的收获”封面图片

Daniel Garza

在构建企业平台的一年里，我不断遇到相同的问题：

  • 如何处理能够真正扩展的基于 RBAC 的 JWT 认证？
  • 为什么每个项目都需要自己的日志配置？
  • 如何在 Azure Key Vault 与本地开发之间管理机密？
  • 管理数据库连接池的最简洁方式是什么？

于是，我将这些模式提炼为 10 个可复用的包 并开源。

Netrun 服务库

所有包均采用 MIT 许可证，并可在 PyPI 上获取：

pip install netrun-auth netrun-logging netrun-config

基础层

netrun-logging – 结构化日志记录，不再糟糕

  • 基于 structlog，相较于标准库提升约 2.9 倍性能。
  • 关键特性： 自动脱敏敏感字段。
from netrun_logging import get_logger

logger = get_logger(__name__)

# The password field is automatically redacted
logger.info(
    "user_login",
    username="daniel",
    password="secret123",   # Logged as "password": "[REDACTED]"
    ip_address="192.168.1.1",
)

netrun-errors – 与 HTTP 对应的异常层级

  • 使用类型化错误，取代通用的 Exceptions。
  • 每个错误类对应相应的 HTTP 状态码。
from netrun_errors import NotFoundError, ValidationError
from fastapi import FastAPI

app = FastAPI()

@app.get("/users/{user_id}")
async def get_user(user_id: str):
    user = await db.get_user(user_id)
    if not user:
        # Raises a 404 response automatically
        raise NotFoundError("User", user_id)
    return user

Source:

认证层

netrun-auth – JWT + RBAC + 多租户隔离

这是最难实现的组件。它使用 Casbin 进行基于策略的访问控制。

from netrun_auth import JWTAuthenticator, require_permission

auth = JWTAuthenticator(
    secret_key="your-secret",
    algorithm="HS256"
)

@app.get("/admin/users")
@require_permission("users:read")
async def list_users(user=Depends(auth.get_current_user)):
    # Only users with the `users:read` permission can access this endpoint
    return await get_users()

多租户隔离确保用户只能访问属于其自身租户的数据：

@require_tenant_access
async def get_tenant_data(tenant_id: str, user=Depends(auth.get_current_user)):
    # Automatically validates: user.tenant_id == tenant_id
    return await db.get_tenant_data(tenant_id)
  • JWTAuthenticator – 处理令牌的创建/验证。
  • require_permission – 检查所需 Casbin 权限的装饰器。
  • require_tenant_access – 强制租户级别隔离的装饰器。

配置层

netrun-config – Azure Key Vault 本地回退

在生产环境中使用 Azure Key Vault，在本地使用 .env 文件时均可工作：

from netrun_config import AzureKeyVaultConfig

config = AzureKeyVaultConfig(
    vault_url="https://my-vault.vault.azure.net",
    cache_ttl=300,  # Cache secrets for 5 minutes
)

# In production: fetches from Key Vault
# Locally: falls back to environment variables
db_password = await config.get_secret("database-password")
  • 生产环境 – 密钥直接从 Azure Key Vault 检索。
  • 本地开发 – 如果无法访问金库，配置会回退到环境变量（例如，从 .env 文件加载的变量）。

LLM 层

netrun-llm – 多提供商编排

netrun-llm 抽象化了不同 LLM 提供商之间的差异，使您能够在它们之间无缝切换或回退。

from netrun_llm import LLMOrchestrator

llm = LLMOrchestrator(
    providers=["azure-openai", "ollama", "claude"],
    fallback_enabled=True,
)

# 自动在主提供商宕机时切换
response = await llm.complete(
    prompt="总结此文档",
    model="gpt-4",
    fallback_model="llama2",  # 如果 Azure 宕机，使用 Ollama
)

数据层

netrun-db-pool – 异步 SQLAlchemy，处理连接风暴

带自动健康检查的连接池

from netrun_db_pool import DatabasePool

pool = DatabasePool(
    url="postgresql+asyncpg://...",
    pool_size=20,
    max_overflow=10,
    health_check_interval=30,
)

async with pool.session() as session:
    result = await session.execute(query)
  • url – 数据库连接字符串。
  • pool_size – 永久保持打开的连接数量。
  • max_overflow – 当连接池耗尽时可以额外创建的连接数。
  • health_check_interval – 自动健康检查 ping 的间隔秒数，用于保持连接活跃。

设计理念

软依赖

每个包都可以独立工作，但当安装多个包时，它们会自动相互集成：

包结果
netrun-auth + netrun-logging认证事件会自动记录
netrun-config + netrun-logging密钥访问会被审计
netrun-db-pool + netrun-errors连接错误会被类型化

集成在运行时通过一个简单的检测模式完成：

try:
    from netrun_logging import get_logger
    logger = get_logger(__name__)
except ImportError:                     # Fallback to the standard library
    import logging
    logger = logging.getLogger(__name__)

零配置默认

每个包都附带合理的默认设置，因此您可以立即使用而无需任何配置：

from netrun_auth import JWTAuthenticator

# Works out‑of‑the‑box
auth = JWTAuthenticator()

# Full customisation when you need it
auth = JWTAuthenticator(
    secret_key="…",
    algorithm="RS256",
    issuer="my-app",
    audience=["api", "web"]
)

测试优先

netrun-pyte – 一个测试优先的工具，随套件一起提供（内容为简洁起见已省略）。

…此处将继续补充关于测试方法的详细信息…

st‑fixtures – 统一测试夹具

# conftest.py
pytest_plugins = ["netrun_pytest_fixtures"]

# Your tests automatically get:
# - mock_auth:   Pre‑configured auth bypass
# - mock_db:     In‑memory SQLite
# - mock_config: Environment‑based config
# - mock_llm:    Deterministic LLM responses

接下来

我正在积极维护这些包。当前优先事项：

  • 更好的文档 – README 是一个开始，但完整的文档站点即将推出。
  • 更多 LLM 提供商 – 添加 Anthropic Claude API、Google Gemini 等。
  • OpenTelemetry 跟踪 – 在所有包中实现统一跟踪。

试一试

pip install netrun-auth netrun-logging netrun-config

GitHub: https://github.com/your‑org/st‑fixtures
PyPI: https://pypi.org/project/st‑fixtures/

MIT 许可证。欢迎提交问题和拉取请求！

在你的 FastAPI 应用中，你使用哪些跨切关注点的模式？我很想了解我遗漏了什么。

Back to Blog

相关文章

阅读更多 »