目录

API_KEY

方法一

from fastapi import Security, HTTPException, status
from fastapi.security.api_key import APIKeyHeader


API_KEY="123456"
API_KEY_NAME = "X-API-KEY"
api_key_header = APIKeyHeader(name=API_KEY_NAME, auto_error=True)

async def get_api_key(api_key: str = Security(api_key_header)):
    if api_key != API_KEY:
        raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,
             detail="Invalid API Key"
        )


@app.get('/index', dependencies=[Security(get_api_key)])
async def index():
    return {"message": "Hello World"}

APIKeyHeader 的源码

class APIKeyHeader(APIKeyBase):
    def __init__(
        self,
        *,
        name: str,
        scheme_name: Optional[str] = None,
        description: Optional[str] = None,
        auto_error: bool = True,
    ):
        self.model: APIKey = APIKey(
            **{"in": APIKeyIn.header},  # type: ignore[arg-type]
            name=name,
            description=description,
        )
        self.scheme_name = scheme_name or self.__class__.__name__
        self.auto_error = auto_error

    async def __call__(self, request: Request) -> Optional[str]:
        api_key = request.headers.get(self.model.name)
        if not api_key:
            if self.auto_error:
                raise HTTPException(
                    status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
                )
            else:
                return None
        return api_key

如果 auto_error=True（默认），那么当 api_key 不存在时，会抛出 HTTPException 异常，返回 403 状态码。

如果 auto_error=False，那么当 api_key 不存在时，会返回 None。那么自定义函数 get_api_key 才会被调用。

方法二

from fastapi import Security, HTTPException, status
from fastapi.security.api_key import APIKeyHeader, APIKey


API_KEY="123456"
API_KEY_NAME = "X-API-KEY"
api_key_header = APIKeyHeader(name=API_KEY_NAME, auto_error=True)

async def get_api_key(api_key: str = Security(api_key_header)):
    if api_key != API_KEY:
        raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,
             detail="Invalid API Key"
        )


@app.get('/index')
async def index(api_key: APIKey = Security(get_api_key)):
    return {"message": "Hello World"}

参考资料