-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathmain.py
More file actions
118 lines (92 loc) · 3.14 KB
/
Copy pathmain.py
File metadata and controls
118 lines (92 loc) · 3.14 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
import uvicorn
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager
from loguru import logger
from app.core.config import settings, get_settings
from app.core.logging import setup_logging
from app.core.cli import (
parse_args,
apply_cli_args_to_env,
merge_cli_args_with_defaults,
print_startup_info,
)
from app.middleware.error_handler import error_handler_middleware
from app.middleware.logging_middleware import logging_middleware
from app.core.redis import redis_client
from app.api.v1.router import api_router
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Application lifespan manager"""
# Startup
logger.info(f"Starting {settings.app_name} v{settings.app_version}")
logger.info(f"Environment: {settings.app_env}")
# Initialize Redis
await redis_client.connect()
yield
# Shutdown
logger.info("Shutting down application")
await redis_client.disconnect()
def create_app() -> FastAPI:
"""Application factory"""
# Setup logging
setup_logging()
# Create FastAPI app
app = FastAPI(
title=settings.app_name,
version=settings.app_version,
debug=settings.debug,
lifespan=lifespan,
docs_url="/docs" if settings.is_development else None,
redoc_url="/redoc" if settings.is_development else None,
)
# Custom middlewares (order matters: last added = first executed)
app.middleware("http")(error_handler_middleware)
app.middleware("http")(logging_middleware)
# CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"] if settings.is_development else [],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Include API routers
app.include_router(api_router)
# Health check endpoint
@app.get("/health")
async def health_check():
# Check Redis health
redis_healthy = await redis_client.health_check()
overall_status = "healthy" if redis_healthy else "degraded"
return {
"status": overall_status,
"checks": {"redis": "ok" if redis_healthy else "error"},
"app": settings.app_name,
"version": settings.app_version,
"environment": settings.app_env,
}
logger.info("Application created successfully")
return app
app = create_app()
if __name__ == "__main__":
# 解析命令行参数
cli_args = parse_args()
# 应用命令行参数到环境变量
apply_cli_args_to_env(cli_args)
# 重新加载设置（应用命令行参数后）
get_settings.cache_clear()
current_settings = get_settings()
# 合并配置
config = merge_cli_args_with_defaults(cli_args)
# 打印启动信息
print_startup_info(config, current_settings)
# 启动服务
uvicorn.run(
"main:app",
host=cli_args.host or current_settings.host,
port=cli_args.port or current_settings.port,
workers=config.get("workers", 1),
reload=config.get("reload", False),
log_level=config.get("log_level", "INFO").lower(),
)