a47b4b5dd0
- Introduced a comprehensive API contract for the WonderQ-MiniAPP, detailing endpoints for site configuration, product listings, and lead submissions. - Defined data types for various entities including HeroSlide, Destination, Theme, CtaBanner, PublicProduct, and more. - Specified request and response formats, including error handling guidelines. chore: Update requirements to include python-multipart - Added python-multipart dependency to requirements.txt for handling file uploads. test: Implement API contract tests - Created test suite for API contracts, validating serializers and endpoints for public products and leads. - Included tests for destination and product serializers, ensuring correct data handling and validation. test: Add configuration tests for OSS settings - Implemented tests to verify that OSS settings are correctly loaded from environment variables.
46 lines
1.8 KiB
Python
46 lines
1.8 KiB
Python
import logging
|
|
from fastapi import FastAPI, HTTPException, Request
|
|
from fastapi.exceptions import RequestValidationError
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.responses import JSONResponse
|
|
from .config import get_settings
|
|
from .routers import admin, public
|
|
|
|
|
|
def create_app() -> FastAPI:
|
|
settings = get_settings()
|
|
logging.basicConfig(level=settings.log_level.upper())
|
|
|
|
app = FastAPI(title="WonderQ Admin API")
|
|
origins = ["*"] if settings.cors_origins == "*" else [item.strip() for item in settings.cors_origins.split(",") if item.strip()]
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=origins,
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
@app.exception_handler(RequestValidationError)
|
|
async def validation_exception_handler(_request: Request, exc: RequestValidationError):
|
|
first = exc.errors()[0] if exc.errors() else {}
|
|
return JSONResponse(status_code=400, content={"message": first.get("msg", "请求参数不正确")})
|
|
|
|
@app.exception_handler(HTTPException)
|
|
async def http_exception_handler(_request: Request, exc: HTTPException):
|
|
if isinstance(exc.detail, dict) and "message" in exc.detail:
|
|
return JSONResponse(status_code=exc.status_code, content=exc.detail)
|
|
return JSONResponse(status_code=exc.status_code, content={"message": exc.detail})
|
|
|
|
@app.exception_handler(Exception)
|
|
async def unhandled_exception_handler(request: Request, exc: Exception):
|
|
request.app.logger.exception(exc) if hasattr(request.app, "logger") else logging.exception(exc)
|
|
return JSONResponse(status_code=500, content={"message": "服务暂时不可用"})
|
|
|
|
app.include_router(public.router)
|
|
app.include_router(admin.router)
|
|
return app
|
|
|
|
|
|
app = create_app()
|