一个功能完整、生产就绪的企业级Python应用模板,集成了现代企业应用所需的所有核心功能。
- 🔐 安全管理: API密钥管理、RBAC权限控制、JWT认证
- ⚡ 性能优化: 连接池管理、缓存策略、异步处理
- 📊 可观测性: 分布式追踪、指标收集、智能告警
- 🛡️ 数据治理: 数据质量管理、血缘追踪、合规检查
- 🔄 高可用性: 负载均衡、故障转移、健康检查
- 🔗 企业集成: API网关、消息队列、第三方服务集成
- 框架: FastAPI, SQLAlchemy, Pydantic
- 数据库: PostgreSQL, Redis, MongoDB
- 监控: OpenTelemetry, Prometheus, Jaeger
- 消息队列: RabbitMQ, Apache Kafka
- 缓存: Redis, Memcached
- 部署: Docker, Kubernetes, Helm
Templates/
├── templates/ # 核心模板模块
│ ├── auth/ # 认证授权模块
│ │ ├── __init__.py
│ │ ├── jwt_auth.py # JWT认证实现
│ │ └── oauth.py # OAuth集成
│ ├── security/ # 安全管理模块
│ │ ├── __init__.py
│ │ ├── api_keys.py # API密钥管理
│ │ └── rbac.py # 基于角色的访问控制
│ ├── performance/ # 性能优化模块
│ │ ├── __init__.py
│ │ └── connection_pool.py # 连接池管理
│ ├── observability/ # 可观测性模块
│ │ ├── __init__.py
│ │ ├── tracing.py # 分布式追踪
│ │ ├── metrics.py # 指标收集
│ │ └── alerting.py # 告警系统
│ ├── data_governance/ # 数据治理模块
│ │ ├── __init__.py
│ │ └── data_quality.py # 数据质量管理
│ ├── high_availability/ # 高可用性模块
│ │ ├── __init__.py
│ │ ├── load_balancer.py # 负载均衡
│ │ └── failover.py # 故障转移
│ ├── enterprise_integration/ # 企业集成模块
│ │ ├── __init__.py
│ │ └── api_gateway.py # API网关
│ ├── database/ # 数据库模块
│ │ ├── __init__.py
│ │ ├── models.py # 数据模型
│ │ ├── connection.py # 数据库连接
│ │ └── migrations/ # 数据库迁移
│ ├── api/ # API模块
│ │ ├── __init__.py
│ │ ├── routes/ # 路由定义
│ │ ├── middleware/ # 中间件
│ │ └── dependencies.py # 依赖注入
│ ├── core/ # 核心功能
│ │ ├── __init__.py
│ │ ├── config.py # 配置管理
│ │ ├── logging.py # 日志配置
│ │ └── exceptions.py # 异常处理
│ └── utils/ # 工具函数
│ ├── __init__.py
│ ├── helpers.py # 辅助函数
│ └── validators.py # 验证器
├── examples/ # 使用示例
│ ├── basic_example.py # 基础使用示例
│ ├── advanced_example.py # 高级功能示例
│ └── enterprise_example.py # 企业级功能演示
├── tests/ # 测试代码
│ ├── unit/ # 单元测试
│ ├── integration/ # 集成测试
│ └── performance/ # 性能测试
├── docs/ # 文档
│ ├── enterprise-features.md # 企业级功能文档
│ ├── api-reference.md # API参考文档
│ └── deployment-guide.md # 部署指南
├── docker/ # Docker配置
│ ├── Dockerfile
│ ├── docker-compose.yml
│ └── docker-compose.prod.yml
├── k8s/ # Kubernetes配置
│ ├── deployment.yaml
│ ├── service.yaml
│ └── ingress.yaml
├── requirements.txt # Python依赖
├── requirements-dev.txt # 开发依赖
├── pyproject.toml # 项目配置
└── README.md # 项目说明
- Python 3.11+
- PostgreSQL 13+
- Redis 6+
- Docker (可选)
- Kubernetes (生产环境)
# 克隆项目
git clone <repository-url>
cd Templates
# 创建虚拟环境
python -m venv venv
source venv/bin/activate # Linux/Mac
# 或
venv\Scripts\activate # Windows
# 安装依赖
pip install -r requirements.txt
# 开发环境额外依赖
pip install -r requirements-dev.txt
- 环境变量配置:
# 复制环境变量模板
cp .env.example .env
# 编辑环境变量
vim .env
- 数据库初始化:
# 创建数据库
createdb myapp_db
# 运行迁移
alembic upgrade head
- Redis配置:
# 启动Redis服务
redis-server
# 或使用Docker
docker run -d -p 6379:6379 redis:alpine
# 开发模式
uvicorn main:app --reload --host 0.0.0.0 --port 8000
# 生产模式
gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker
# 健康检查
curl http://localhost:8000/health
# API文档
open http://localhost:8000/docs
# 指标监控
curl http://localhost:8000/metrics
from templates.core import create_app
from templates.auth import JWTAuth
from templates.database import DatabaseManager
# 创建应用实例
app = create_app()
# 初始化认证
auth = JWTAuth(secret_key="your-secret-key")
# 初始化数据库
db = DatabaseManager("postgresql://user:pass@localhost/db")
# 启动应用
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
from templates.security import APIKeyManager, RBACManager
from templates.performance import ConnectionPoolManager
from templates.observability import TracingManager, MetricsManager
from templates.high_availability import LoadBalancerManager
# 安全管理
api_key_manager = APIKeyManager()
rbac_manager = RBACManager()
# 性能优化
pool_manager = ConnectionPoolManager()
# 可观测性
tracing_manager = TracingManager()
metrics_manager = MetricsManager()
# 高可用性
lb_manager = LoadBalancerManager()
# 集成到FastAPI应用
from fastapi import FastAPI, Depends
app = FastAPI()
@app.middleware("http")
async def add_tracing(request, call_next):
with tracing_manager.start_trace("http_request") as trace:
trace.set_attribute("http.method", request.method)
trace.set_attribute("http.url", str(request.url))
response = await call_next(request)
trace.set_attribute("http.status_code", response.status_code)
return response
@app.get("/api/users")
async def get_users(
current_user = Depends(rbac_manager.get_current_user)
):
# 检查权限
rbac_manager.check_permission(current_user.id, "users:read")
# 使用连接池获取数据
async with pool_manager.get_connection("main_db") as conn:
users = await conn.fetch("SELECT * FROM users")
# 记录指标
metrics_manager.increment_counter("api_requests", {
"endpoint": "/api/users",
"method": "GET"
})
return users
from templates.data_governance import DataQualityManager
# 初始化数据质量管理器
dq_manager = DataQualityManager()
# 定义数据质量规则
completeness_rule = dq_manager.create_rule(
"email_completeness",
"completeness",
"email",
threshold=0.95,
description="Email字段完整性检查"
)
uniqueness_rule = dq_manager.create_rule(
"user_id_uniqueness",
"uniqueness",
"user_id",
threshold=1.0,
description="用户ID唯一性检查"
)
# 执行数据质量检查
import pandas as pd
data = pd.read_sql("SELECT * FROM users", connection)
results = dq_manager.run_checks("users", data)
# 生成质量报告
report = dq_manager.generate_report("users")
print(f"数据质量得分: {report.overall_score}")
# 构建镜像
docker build -t enterprise-app .
# 运行容器
docker run -p 8000:8000 enterprise-app
# 使用docker-compose
docker-compose up -d
# 生产环境部署
docker-compose -f docker-compose.prod.yml up -d
# 扩展服务
docker-compose -f docker-compose.prod.yml up -d --scale app=3
# 应用配置
kubectl apply -f k8s/
# 检查部署状态
kubectl get pods -l app=enterprise-app
# 查看服务
kubectl get services
# 查看日志
kubectl logs -f deployment/enterprise-app
访问 http://localhost:8000/metrics
查看Prometheus格式的指标数据。
主要指标包括:
http_requests_total
: HTTP请求总数http_request_duration_seconds
: 请求响应时间database_connections_active
: 活跃数据库连接数cache_hits_total
: 缓存命中次数api_key_validations_total
: API密钥验证次数
使用Jaeger查看分布式追踪数据:
# 启动Jaeger
docker run -d --name jaeger \
-p 16686:16686 \
-p 14268:14268 \
jaegertracing/all-in-one:latest
# 访问Jaeger UI
open http://localhost:16686
应用使用结构化日志,支持多种日志后端:
from templates.core.logging import get_logger
logger = get_logger("my_module")
logger.info("用户登录", extra={
"user_id": "123",
"ip_address": "192.168.1.100",
"user_agent": "Mozilla/5.0..."
})
# 运行所有测试
pytest
# 运行单元测试
pytest tests/unit/
# 运行集成测试
pytest tests/integration/
# 生成覆盖率报告
pytest --cov=templates --cov-report=html
# 使用locust进行负载测试
locust -f tests/performance/locustfile.py --host=http://localhost:8000
# 数据库性能测试
python tests/performance/db_benchmark.py
# 代码格式化
black templates/
isort templates/
# 代码检查
flake8 templates/
mypy templates/
# 安全检查
bandit -r templates/
# 更新依赖
pip-compile requirements.in
pip-compile requirements-dev.in
# 安全审计
safety check
# 许可证检查
pip-licenses
- Fork项目
- 创建功能分支 (
git checkout -b feature/amazing-feature
) - 提交更改 (
git commit -m 'Add some amazing feature'
) - 推送到分支 (
git push origin feature/amazing-feature
) - 创建Pull Request
- 遵循PEP 8代码风格
- 编写单元测试
- 更新相关文档
- 确保所有测试通过
本项目采用MIT许可证 - 查看 LICENSE 文件了解详情。
如果您遇到问题或有疑问:
- GraphQL API支持
- 机器学习模型集成
- 更多数据库支持
- 高级缓存策略
- 微服务架构支持
- 服务网格集成
- 云原生功能增强
- AI驱动的运维
感谢所有为这个项目做出贡献的开发者和社区成员。
特别感谢以下开源项目:
- FastAPI
- SQLAlchemy
- OpenTelemetry
- Prometheus
- Redis
- PostgreSQL
企业级Python模板项目 - 让您的应用从开发到生产一步到位!