alien_py_cloud/alien_gateway/main.py

import logging
import os
from typing import List
import httpx
from fastapi import FastAPI, Request, Response, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from starlette.status import HTTP_502_BAD_GATEWAY
from alien_gateway.config import settings
from alien_util.redis_client import check_redis_connection

app = FastAPI(
    title=f"{settings.PROJECT_NAME} - Gateway & Auth Service",
    version="1.0.0",
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

LOG_DIR = os.path.join("common", "logs", "alien_gateway")
os.makedirs(LOG_DIR, exist_ok=True)


def _init_logger():
    logger = logging.getLogger("alien_gateway")
    if logger.handlers:
        return logger
    logger.setLevel(logging.INFO)
    fmt = logging.Formatter("%(asctime)s [%(levelname)s] %(name)s %(message)s")

    info_handler = logging.FileHandler(os.path.join(LOG_DIR, "info.log"), encoding="utf-8")
    info_handler.setLevel(logging.INFO)
    info_handler.setFormatter(fmt)

    error_handler = logging.FileHandler(os.path.join(LOG_DIR, "error.log"), encoding="utf-8")
    error_handler.setLevel(logging.ERROR)
    error_handler.setFormatter(fmt)

    console_handler = logging.StreamHandler()
    console_handler.setFormatter(fmt)

    logger.addHandler(info_handler)
    logger.addHandler(error_handler)
    logger.addHandler(console_handler)
    return logger


logger = _init_logger()


@app.get("/health")
async def health():
    return {"service": "alien_gateway", "status": "ok", "env": settings.APP_ENV}


@app.get("/health/redis")
async def redis_health():
    try:
        return check_redis_connection()
    except Exception as exc:
        logger.error("redis health check failed err=%s", exc)
        raise HTTPException(status_code=HTTP_502_BAD_GATEWAY, detail="Redis unavailable")


# 此模块未来将承担 JWT 签发、权限校验中间件、路由聚合等核心功能
@app.post("/auth/login")
async def login():
    return {"message": "Auth logic here"}


HOP_BY_HOP_HEADERS: List[str] = [
    "connection",
    "keep-alive",
    "proxy-authenticate",
    "proxy-authorization",
    "te",
    "trailers",
    "transfer-encoding",
    "upgrade",
]


def _clean_headers(headers):
    return {k: v for k, v in headers.items() if k.lower() not in HOP_BY_HOP_HEADERS}


async def _proxy(request: Request, target_url: str, service_tag: str) -> Response:
    """通用反向代理：转发请求体、头部、查询参数并回写下游响应。"""
    client_ip = request.client.host if request.client else "-"
    body = await request.body()
    headers = _clean_headers(dict(request.headers))

    try:
        async with httpx.AsyncClient(timeout=30.0) as client:
            resp = await client.request(
                request.method,
                target_url,
                content=body,
                headers=headers,
                params=request.query_params,
            )
    except Exception as exc:
        logger.error(
            "proxy to %s failed ip=%s url=%s err=%s",
            service_tag, client_ip, target_url, exc,
        )
        raise HTTPException(status_code=HTTP_502_BAD_GATEWAY, detail="Upstream unavailable")

    return Response(
        content=resp.content,
        status_code=resp.status_code,
        headers=_clean_headers(resp.headers),
        media_type=resp.headers.get("content-type"),
    )


@app.api_route("/api/store/{full_path:path}", methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"])
async def proxy_to_store(full_path: str, request: Request):
    target_url = f"{settings.STORE_BASE_URL}/api/store/{full_path}"
    return await _proxy(request, target_url, "store")


@app.api_route("/api/contract/{full_path:path}", methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"])
async def proxy_to_contract(full_path: str, request: Request):
    target_url = f"{settings.CONTRACT_BASE_URL}/api/contract/{full_path}"
    return await _proxy(request, target_url, "contract")


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=settings.GATEWAY_PORT)