Используя FastAPI в качестве бэкэнда, я в настоящее время сталкиваюсь с проблемой CORS (совместное использование ресурсо ⇐ Python

[Anonymous]

Код: Выделить всё

# main.py
from fastapi import FastAPI
from fastapi.routing import APIRoute
from fastapi.middleware.cors import CORSMiddleware

from app.api.main import api_router
from app.core.config import settings
from app.core.middleware import JWTMiddleware
from app.logging_config import setup_logging

# app = FastAPI()

setup_logging()

def custom_generate_unique_id(route: APIRoute) -> str:
return f"{route.tags[0]}-{route.name}"

if settings.ENVIRONMENT == 'test':
openapi_url = f"{settings.API_V1_STR}/openapi.json"
docs_url = f"{settings.API_V1_STR}/docs"
redoc_url = f"{settings.API_V1_STR}/redoc"
else:
openapi_url = None
docs_url = None
redoc_url = None

app = FastAPI(
title=settings.PROJECT_NAME,
openapi_url=openapi_url,
generate_unique_id_function=custom_generate_unique_id,
docs_url=docs_url,
redoc_url=redoc_url,
)

origins = [
"http://localhost",
"http://localhost:5173",
"https://localhost",
"https://localhost:5173",
"http://localhost:8086",
"https://localhost:8086"
]

if origins:
app.add_middleware(
CORSMiddleware,
allow_origins=origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
max_age=0
)

app.add_middleware(JWTMiddleware)

app.include_router(api_router, prefix=settings.API_V1_STR)

Код: Выделить всё

# middleware.py
import logging
import jwt
from fastapi.responses import JSONResponse, Response
from datetime import datetime
from starlette.middleware.base import BaseHTTPMiddleware

from app.api.deps import get_redis, get_db
from app.core.config import settings
from app.core.security import ALGORITHM
from app.schemas.response_schemas import Code, Message, RestResult
from app.utils import parse_token
from app.constant import WhiteList

logger = logging.getLogger(__name__)

class JWTMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
white_list = WhiteList.white_list.value

if request.url.path in white_list:
response = await call_next(request)
return response

if request.method == "OPTIONS":
logger.info("Handling preflight request for %s", request.url.path)
headers = {
'Access-Control-Allow-Methods': ', '.join(['GET', 'POST', 'PUT', 'DELETE', 'OPTIONS']),
'Access-Control-Allow-Headers': ', '.join(['Authorization', 'Content-Type']),
'Access-Control-Allow-Origin': request.headers.get('Origin', ''),
}
return Response(content=None, status_code=204, headers=headers)

cache = True
if request.url.path == "/api/v1/logout":
cache = False

authorization: str = request.headers.get("Authorization")
if authorization and authorization.startswith("Bearer "):
token = authorization.split(" ")[1]

async with get_redis() as redis, get_db() as session:
request.state.mysql_session = session
request.state.redis_session = redis

if await redis.exists(token):
logger.info(Message.INVALID_TOKEN)
resp_data = RestResult(code=Code.FAIL, msg=Message.INVALID_TOKEN).__dict__
return JSONResponse(resp_data)

try:
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=ALGORITHM)

if datetime.fromtimestamp(payload["exp"]) <  datetime.utcnow():
logger.info(Message.INVALID_TOKEN)
resp_data = RestResult(code=Code.FAIL, msg=Message.INVALID_TOKEN).__dict__
return JSONResponse(resp_data)

res = await parse_token(session, redis, payload, cache)
if not res:
logger.info(f"The parsing result is empty: {res}")
resp_data = RestResult(code=Code.FAIL, msg=Message.INVALID_TOKEN).__dict__
return JSONResponse(resp_data)

request.state.user_payload = res

except jwt.InvalidTokenError:
logger.info(Message.INVALID_TOKEN)
resp_data = RestResult(code=Code.FAIL, msg=Message.INVALID_TOKEN).__dict__
return JSONResponse(resp_data)
else:
logger.info(Message.INVALID_TOKEN)
resp_data = RestResult(code=Code.FAIL, msg=Message.ILLEGAL_REQUEST).__dict__
return JSONResponse(resp_data)

response = await call_next(request)
return response

Код: Выделить всё

# mine_api.py
import logging
import traceback

from fastapi import APIRouter, Request, Form
from fastapi.responses import JSONResponse
from sqlalchemy.exc import SQLAlchemyError, IntegrityError

from app.schemas.response_schemas import RestResult, Code, Message
from app.dao.user_center.mine_dao import update_user_info_dao

router = APIRouter()
@router.get(
"/user_account",
)
async def my_account(
request: Request
) -> JSONResponse:
return True

Разработчик внешнего интерфейса использует запрос GET для вызова конечной точки API /api/v1/user_account и сталкивается с ошибкой CORS (совместное использование ресурсов между источниками). Я убедился, что происхождение пользователя внешнего интерфейса указано в настройках.BACKEND_CORS_ORIGINS. Вот скриншот запроса:
введите здесь описание изображения

Подробнее здесь: https://stackoverflow.com/questions/791 ... n-resource