0% found this document useful (0 votes)
3 views32 pages

Chapter 4

Uploaded by

Uzair Waheed
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as PDF, TXT or read online on Scribd
0% found this document useful (0 votes)
3 views32 pages

Chapter 4

Uploaded by

Uzair Waheed
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as PDF, TXT or read online on Scribd

API versioning and

documentation
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 API versioning?

API endpoints as menu items

Keep old customers happy


Some customers want new options

Iterate without impacting existing


customers

DEPLOYING AI INTO PRODUCTION WITH FASTAPI


API for cloud AI jobs
from pydantic import BaseModel

class AIJobV1(BaseModel):
job_name: str
data: bytes

class AIJobV1(BaseModel):
job_name: str
data: bytes
config: dict

DEPLOYING AI INTO PRODUCTION WITH FASTAPI


Versioned endpoints
from pydantic import BaseModel from fastapi import FastAPI

class AIJobV1(BaseModel): app = FastAPI()


job_name: str
data: bytes @[Link]("/v1/ai-job")
def ai_job_v1(job: AIJobV1):
class AIJobV2(BaseModel): ...
job_name: str
data: bytes @[Link]("/v2/ai-job")
config: dict def ai_job_v2(job: AIJobV2):
...

DEPLOYING AI INTO PRODUCTION WITH FASTAPI


Reasons to update endpoint version

Breaking change in schema

Change in underlying function


Updated model code

Updated model training set

Updated pre/post processing

DEPLOYING AI INTO PRODUCTION WITH FASTAPI


Iteration with optional fields
from pydantic import BaseModel

Versioning is not always required to iterate


class AIJobV1(BaseModel):
Optional fields can support additional data job_name: str
without breaking schemas data: bytes

from typing import Optional

class AIJobV1(BaseModel):
job_name: str
data: bytes
config: Optional[dict]

DEPLOYING AI INTO PRODUCTION WITH FASTAPI


Documenting APIs with Swagger

Standard tool for API documentation


Keeps track of endpoints and versions

Built on OpenAPI standard metadata

DEPLOYING AI INTO PRODUCTION WITH FASTAPI


Swagger UI

DEPLOYING AI INTO PRODUCTION WITH FASTAPI


Swagger UI for an endpoint

DEPLOYING AI INTO PRODUCTION WITH FASTAPI


Using FastAPI's description field
from fastapi import FastAPI

app = FastAPI(
description="AI Job API"
)

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
Advanced input
validation and error
handling
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 we need advanced input

API for restaurant orders

Variable number of items

class Order(BaseModel):
item1: str
item2: str
item3: str

DEPLOYING AI INTO PRODUCTION WITH FASTAPI


Nested Pydantic models
from pydantic import BaseModel from pydantic import BaseModel
from typing import List
class Foo(BaseModel):
count: int class OrderItem(BaseModel):
name: str
class Bar(BaseModel): quantity: int
foo: Foo
class RestaurantOrder(BaseModel):
>>> m = Bar(foo={'count': 4}) customer_name: str

>>> print(m) items: List[OrderItem]


foo=Foo(count=4)

DEPLOYING AI INTO PRODUCTION WITH FASTAPI


Custom model validators
from fastapi import FastAPI class RestaurantOrder(BaseModel):
from [Link] import ( customer_name: str
RequestValidationError items: List[OrderItem]
)
from pydantic import ( @model_validator(mode="after")
BaseModel, def validate_after(self):
model_validator, if len([Link]) == 0:
) raise RequestValidationError(
from typing import List "No items in order!"
)
class OrderItem(BaseModel): return self
name: str
quantity: int {"detail":"No items in order!"}

DEPLOYING AI INTO PRODUCTION WITH FASTAPI


Global exception handlers
from fastapi import FastAPI
from [Link] import RequestValidationError
from [Link] import PlainTextResponse

app = FastAPI()

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
msg = "Input validation error. See the documentation: [Link]
return PlainTextResponse(msg, status_code=422)

Input validation error. See the documentation: [Link]

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
Monitoring and
logging
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 monitoring and logging?

Can't debug in production

App supervisor needs a simple health


check

Logging key metrics over time

DEPLOYING AI INTO PRODUCTION WITH FASTAPI


Setting up custom logging
from fastapi import FastAPI
import logging

logger = [Link](
Load the uvicorn error logger
'[Link]'
)
app = FastAPI()
Add custom logs to app startup
[Link]("App is running!")

@[Link]('/')
Add custom logs to endpoints async def main():
[Link]('GET /')
return 'ok'

DEPLOYING AI INTO PRODUCTION WITH FASTAPI


Logging a when a model is loaded
from fastapi import FastAPI
import logging
import joblib

logger = [Link]('[Link]')

model = [Link]('penguin_classifier.pkl')
[Link]("Penguin classifier loaded successfully.")

app = FastAPI()

DEPLOYING AI INTO PRODUCTION WITH FASTAPI


Logging process time with middleware
from fastapi import FastAPI, Request
import logging
import time
logger = [Link]('[Link]')
app = FastAPI()

@[Link]("http")
async def log_process_time(request: Request, call_next):
start_time = time.perf_counter()
response = await call_next(request)
process_time = time.perf_counter() - start_time
[Link](f"Process time was {process_time} seconds.")
return response

1 [Link]

DEPLOYING AI INTO PRODUCTION WITH FASTAPI


Setting the logging level

uvicorn main:app --log-level debug

DEPLOYING AI INTO PRODUCTION WITH FASTAPI


Monitoring

from fastapi import FastAPI

app = FastAPI()
@[Link]("/health") "I'm ok!"
async def get_health():
return {"status": "OK"}

DEPLOYING AI INTO PRODUCTION WITH FASTAPI


Sharing model parameters with monitoring
from fastapi import FastAPI
import joblib

model = [Link](
'penguin_classifier.pkl'
)
app = FastAPI()
@[Link]("/health")
async def get_health(): "I'm ok!"
params = model.get_params()
"Here are some fun facts about me!"
return {"status": "OK",
"params": params}

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
Wrap-up
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
Introduction to FastAPI for Model Deployment
Chapter 1
Basic GET and POST requests

Loading a pre-trained model

Running the uvicorn server

Pydantic models for requests and responses

DEPLOYING AI INTO PRODUCTION WITH FASTAPI


Integrating AI models
Chapter 2
More structured input types

Loading a pre-trained model in the app

Structured prediction results

DEPLOYING AI INTO PRODUCTION WITH FASTAPI


Securing and optimizing the API
Chapter 3
API key authentication

Rate limiting

Async processing

DEPLOYING AI INTO PRODUCTION WITH FASTAPI


API versioning, monitoring and logging
Chapter 4
API versioning and documentation

Advanced input validation and error handling

Monitoring and logging

DEPLOYING AI INTO PRODUCTION WITH FASTAPI


Congratulations!
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

You might also like