89 lines
2.4 KiB
Python
89 lines
2.4 KiB
Python
from fastapi import FastAPI, Depends, HTTPException, Security
|
|
from fastapi.security.api_key import APIKeyHeader
|
|
|
|
from fastapi.openapi.models import APIKey
|
|
from fastapi.openapi.utils import get_openapi
|
|
|
|
|
|
from pydantic import BaseModel
|
|
from typing import Dict
|
|
|
|
|
|
app = FastAPI()
|
|
|
|
|
|
|
|
API_KEY = "your-secret-api-key"
|
|
API_KEY_NAME = "X-API-Key"
|
|
|
|
api_key_header = APIKeyHeader(name=API_KEY_NAME, auto_error=False)
|
|
|
|
def get_api_key(api_key: str = Depends(api_key_header)):
|
|
if api_key == API_KEY:
|
|
return api_key
|
|
else:
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="Invalid or missing API Key",
|
|
)
|
|
|
|
|
|
def custom_openapi():
|
|
if app.openapi_schema:
|
|
return app.openapi_schema
|
|
openapi_schema = get_openapi(
|
|
title="FastAPI with API Key Authentication",
|
|
version="1.0.0",
|
|
description="API documentation with API Key authentication",
|
|
routes=app.routes,
|
|
)
|
|
openapi_schema["components"]["securitySchemes"] = {
|
|
"APIKeyHeader": {
|
|
"type": "apiKey",
|
|
"name": API_KEY_NAME,
|
|
"in": "header",
|
|
}
|
|
}
|
|
openapi_schema["security"] = [{"APIKeyHeader": []}]
|
|
app.openapi_schema = openapi_schema
|
|
return app.openapi_schema
|
|
|
|
app.openapi = custom_openapi
|
|
|
|
|
|
|
|
|
|
# In-memory "database"
|
|
items_db: Dict[int, dict] = {}
|
|
|
|
class Item(BaseModel):
|
|
name: str
|
|
description: str = None
|
|
price: float
|
|
|
|
@app.get("/items3/{item_id}", response_model=Item, dependencies=[Depends(get_api_key)])
|
|
async def get_item(item_id: int):
|
|
item = items_db.get(item_id)
|
|
if not item:
|
|
raise HTTPException(status_code=404, detail="Item not found")
|
|
return item
|
|
|
|
@app.post("/items/", response_model=Item, status_code=201)
|
|
async def create_item(item_id: int, item: Item):
|
|
if item_id in items_db:
|
|
raise HTTPException(status_code=400, detail="Item ID already exists")
|
|
items_db[item_id] = item.dict()
|
|
return item
|
|
|
|
@app.put("/items/{item_id}", response_model=Item)
|
|
async def update_item(item_id: int, item: Item):
|
|
if item_id not in items_db:
|
|
raise HTTPException(status_code=404, detail="Item not found")
|
|
items_db[item_id] = item.dict()
|
|
return item
|
|
|
|
@app.delete("/items/{item_id}", status_code=204)
|
|
async def delete_item(item_id: int):
|
|
if item_id not in items_db:
|
|
raise HTTPException(status_code=404, detail="Item not found")
|
|
del items_db[item_id] |