API Key
Authentication
D E P L OY I N G A I I N T O P R O D U C T I O N W I T H FA S TA P I
Matt Eckerle
Software and Data Engineering Leader
Why secure APIs?
Stop unauthorized users
Secure API endpoints with API key
authentication
DEPLOYING AI INTO PRODUCTION WITH FASTAPI
How API keys work
Like a digital password for our API
Sent in request headers
Verified before accessing endpoints
DEPLOYING AI INTO PRODUCTION WITH FASTAPI
Understanding APIKeyHeader
from fastapi import FastAPI
from [Link] import APIKeyHeader
header_scheme = APIKeyHeader(
name="X-API-Key",
auto_error=True
)
DEPLOYING AI INTO PRODUCTION WITH FASTAPI
Authenticating an endpoint
from [Link] import APIKeyHeader ApiKeyHeader
from fastapi import Depends, HTTPException
Depends adds header scheme
header_scheme = APIKeyHeader(name="X-API-Key",
HTTPException for exceptions
auto_error=True)
Defines API key header and secret key
API_SECRET_KEY = "your-secret-key"
Validates API keys with test_api_key
@[Link]("/items/")
def read_items( Raises 403 if the key doesn't match
api_key: str = Depends(header_scheme) API_SECRET_KEY
):
if api_key != API_SECRET_KEY:
raise HTTPException(
status_code=403,
detail="Invalid API key")
return {"api_key": api_key}
DEPLOYING AI INTO PRODUCTION WITH FASTAPI
Authenticating an app
def verify_api_key(api_key: str = Depends(header_scheme)):
if api_key != API_KEY:
raise HTTPException(status_code=403, detail="Invalid API key")
return api_key
app = FastAPI(
dependencies=[Depends(verify_api_key)]
)
@[Link]("/predict")
def predict_sentiment(text: str):
return {
"text": text,
"sentiment": "positive",
"status": "success"
}
DEPLOYING AI INTO PRODUCTION WITH FASTAPI
Testing the endpoint
Command with invalid API key: Invalid key output:
curl -X POST \ {"detail":"Invalid API key"}
[Link] \
-H "X-API-Key: wrong-key" \
-H "Content-Type: application/json" \
-d '{"text": "This product is amazing!"}'
Command with valid API key:
Valid key output:
curl -X POST \
{"text":"This product is amazing!",
[Link] \
"sentiment":"positive",
-H "X-API-Key: your-secret-key" \
"status":"success"}
-H "Content-Type: application/json" \
-d '{"text": "This product is amazing!"}'
DEPLOYING AI INTO PRODUCTION WITH FASTAPI
Let's practice!
D E P L OY I N G A I I N T O P R O D U C T I O N W I T H FA S TA P I
Rate Limiting
D E P L OY I N G A I I N T O P R O D U C T I O N W I T H FA S TA P I
Matt Eckerle
Software and Data Engineering Leader
Introducing rate limiting
Purpose: Controls the frequency of API
requests.
Response: Returns HTTP 429 ("Too Many
Requests") when the limit is exceeded.
DEPLOYING AI INTO PRODUCTION WITH FASTAPI
How rate limiting works
DEPLOYING AI INTO PRODUCTION WITH FASTAPI
Authenticating incoming credentials
DEPLOYING AI INTO PRODUCTION WITH FASTAPI
Rate limiting check
DEPLOYING AI INTO PRODUCTION WITH FASTAPI
Setting up our API
from fastapi import FastAPI, Depends, HTTPException
from [Link] import APIKeyHeader
from pydantic import BaseModel
app = FastAPI()
model = SentimentAnalyzer(pkl_file_path)
API_KEY_HEADER = APIKeyHeader(name="X-API-Key")
API_KEY = "your-secret-key"
DEPLOYING AI INTO PRODUCTION WITH FASTAPI
The rate limiter logic
from datetime import datetime, timedelta
class RateLimiter:
def __init__(self, requests_per_min: int = 10):
self.requests_per_min = requests_per_min
[Link] = defaultdict(list)
def is_rate_limited(
self, api_key: str
) -> tuple[bool, int]:
DEPLOYING AI INTO PRODUCTION WITH FASTAPI
Deleting old requests
from datetime import datetime, timedelta
class RateLimiter:
def __init__(self, requests_per_min: int = 10):
self.requests_per_min = requests_per_min
[Link] = defaultdict(list)
def is_rate_limited(
self, api_key: str
) -> tuple[bool, int]:
now = [Link]()
minute_ago = now - timedelta(minutes=1)
[Link][api_key] = [
req_time for req_time in
[Link][api_key]
if req_time > minute_ago
]
DEPLOYING AI INTO PRODUCTION WITH FASTAPI
Check request count
def is_rate_limited(self, api_key: str) ->
tuple[bool, int]:
now = [Link]()
minute_ago = now - timedelta(minutes=1)
[Link][api_key] = [
req_time for req_time in
[Link][api_key]
if req_time > minute_ago
]
recent_requests = len([Link][api_key])
if recent_requests >= self.requests_per_min:
return True, 0
[Link][api_key].append(now)
return False
DEPLOYING AI INTO PRODUCTION WITH FASTAPI
Add rate limit check
rate_limiter = RateLimiter(requests_per_minute=10)
def test_api_key(api_key: str = Depends(API_KEY_HEADER)):
if api_key != API_KEY:
raise HTTPException(
status_code=403,
detail="Invalid API key"
)
is_limited, _ = rate_limiter.is_rate_limited(api_key)
if is_limited:
raise HTTPException(
status_code=429,
detail="Rate limit exceeded. Please try again later."
)
return api_key
DEPLOYING AI INTO PRODUCTION WITH FASTAPI
Apply rate limit to endpoint
@[Link]("/predict") Send request 11 times:
def predict_sentiment(
request: SentimentRequest, curl -X POST "[Link] \
api_key: str = Depends(test_api_key) -H "Content-Type: application/json" \
): -H "X-API-Key: your-secret-key" \
result = sentiment_model([Link]) -d '{"text": "I love this product"}'
_, requests_remaining =
Output:
rate_limiter.is_rate_limited(api_key)
{"detail":"Rate limit exceeded.
return {
Please try again later."}
"text": [Link],
"sentiment": result[0]["label"].lower(),
"confidence": result[0]["score"],
"requests_remaining": requests_remaining
}
DEPLOYING AI INTO PRODUCTION WITH FASTAPI
Let's practice!
D E P L OY I N G A I I N T O P R O D U C T I O N W I T H FA S TA P I
Asynchronous
processing
D E P L OY I N G A I I N T O P R O D U C T I O N W I T H FA S TA P I
Matt Eckerle
Software and Data Engineering Leader
What is asynchronous processing
Allows handling multiple requests
concurrently
DEPLOYING AI INTO PRODUCTION WITH FASTAPI
Synchronous vs asynchronous requests
DEPLOYING AI INTO PRODUCTION WITH FASTAPI
Turning synchronous endpoints asynchronous
@[Link]("/analyze") import asyncio
def analyze_sync(comment: Comment):
result = sentiment_model([Link]) @[Link]("/analyze")
return {"sentiment": result} async def analyze_async(comment: Comment):
result = await asyncio.to_thread(
sentiment_model, [Link]
)
return {"sentiment": result}
DEPLOYING AI INTO PRODUCTION WITH FASTAPI
Implementing background tasks
from fastapi import BackgroundTasks BackgroundTasks manage comment
from typing import List processing queue
@[Link]("/analyze_batch")
async def analyze_batch(
comments: Comments, background_tasks handles post-response
background_tasks: BackgroundTasks processing.
):
async def process_comments(texts: List[str]):
for text in texts: add_task schedules process_comments
result = await asyncio.to_thread( asynchronously.
sentiment_model, text)
background_tasks.add_task(process_comments,
[Link])
return {"message": "Processing started"}
DEPLOYING AI INTO PRODUCTION WITH FASTAPI
Adding error handling
@[Link]("/analyze_comment")
async def analyze_comment(comment: Comment):
try:
sentiment_model = SentimentAnalyzer()
result = await asyncio.wait_for(
sentiment_model([Link]),
timeout=5.0
)
return {"sentiment": result["label"]}
DEPLOYING AI INTO PRODUCTION WITH FASTAPI
Adding error handling
except [Link]:
raise HTTPException(
status_code=408,
detail="Analysis timed out"
)
except Exception:
raise HTTPException(
status_code=500,
detail="Analysis failed"
)
DEPLOYING AI INTO PRODUCTION WITH FASTAPI
Let's practice!
D E P L OY I N G A I I N T O P R O D U C T I O N W I T H FA S TA P I