from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def read_root():
return {"message": "Hello, World!"}?from fastapi import FastAPI
app = FastAPI()
@app.get("/async_endpoint")
async def async_endpoint():
result = await some_async_function()
return {"result": result}?from fastapi import Depends, FastAPI
app = FastAPI()
async def get_some_data():
# Asynchronous operations here
return {"data": "Some asynchronous data"}
@app.get("/")
async def read_data(data: dict = Depends(get_some_data)):
return data?from fastapi import BackgroundTasks, FastAPI
app = FastAPI()
def write_log(message: str):
with open("log.txt", mode="a") as log:
log.write(message)
@app.post("/send-notification/{email}")
async def send_notification(
email: str, background_tasks: BackgroundTasks
):
message = f"message to {email}"
background_tasks.add_task(write_log, message)
return {"message": "Message sent in the background"}?from fastapi import FastAPI, WebSocket
app = FastAPI()
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Message text was: {data}")
By embracing asynchronous programming, FastAPI is able to handle a large number of simultaneous connections efficiently, making it suitable for building scalable and performant web applications and APIs.
@get decorator :@get(“/users”)
def list_users():
return [{“username”: “jane”}, {“username”: “joe”}]?http://localhost:8000/docs, and ReDoc at http://localhost:8000/redoc.from fastapi import Depends, FastAPI, HTTPException, Security
from fastapi.security import OAuth2PasswordBearer, SecurityScopes
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
def get_current_user(security_scopes: SecurityScopes, token: str = Depends(oauth2_scheme)):
# decode token and fetch user data here
raise HTTPException(status_code=403, detail="Not authenticated")
app = FastAPI()
@app.get("/items/", dependencies=[Depends(Security(get_current_user, scopes=["items:read"]))])
async def read_items():
return [{"item": "Foo", "value": "Bar"}]?from fastapi import FastAPI, HTTPException
app = FastAPI()
@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc):
return JSONResponse(
status_code=exc.status_code,
content={"message": f"Oops! {exc.detail}"},
)
@app.get("/items/{item_id}")
async def read_item(item_id: str):
if item_id not in items:
raise HTTPException(status_code=404, detail="Item not found")?from fastapi import Depends, FastAPI
app = FastAPI()
def get_db():
# Function to get a database connection
# ...
def get_current_user(token: str = Depends(get_token)):
# Function to get the current user based on the token
# ...
@app.get("/items/")
async def read_items(db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
# Route function that depends on the database and current user
# ...?get_db dependency that provides a database session, and then use it in multiple route functions.HTTP methods with routes in FastAPI by specifying the methods argument with a list of methods when you add a route. For example, if you want to add a route that can be accessed with the GET and POST methods, you would do the following:@app.get(“/my-route”, methods=[“GET”, “POST”])
def my_route():
return “Hello, world!”?SQLAlchemy through the use of Pydantic models and dependency injection. First, a SQLAlchemy model is defined for the database structure. Then, a Pydantic model is created to handle data validation and serialization. FastAPI uses these models in route functions to interact with the database.SQLAlchemy’s ORM is used directly within the route functions. The session provided by the dependency handles transactions, ensuring changes are committed or rolled back appropriately. from fastapi import FastAPI, UploadFile, File
app = FastAPI()
@app.post("/files/")
async def create_file(file: UploadFile = File(...)):
return {"filename": file.filename}?UploadFile = File(...) declares a new parameter of type UploadFile. The File function is a “special function” used to declare it..file, get metadata like filename or content type with .filename and .content_type. from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
@app.get("/home", dependencies=[Depends(limiter.limit("5/minute"))])
async def home(request: Request):
return {"Hello": "World"}?/home” endpoint. The key_func parameter determines how to identify different clients, here we’re using the client’s IP address. FastAPI will automatically create an instance of the BackgroundTasks class and inject it into your route function.from fastapi import FastAPI, BackgroundTasks
app = FastAPI()
def write_log(message: str):
# Simulate a time-consuming task (e.g., writing to a log file)
with open("log.txt", mode="a") as log:
log.write(message)
@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
message = f"message to {email}"
background_tasks.add_task(write_log, message)
return {"message": "Message sent in the background"}?send_notification route function takes an email parameter and a background_tasks parameter of type BackgroundTasks. The function sends a message and schedules the write_log function to run in the background.
add_task method of the BackgroundTasks instance. Each task added will be executed asynchronously after the response has been sent to the client.BackgroundTasks are executed in the order they are added, and they are executed after the response to the client has been sent. This ensures that the client receives a quick response, and any additional, potentially time-consuming tasks are performed in the background.from fastapi import FastAPI, BackgroundTasks
app = FastAPI()
def write_log(message: str):
with open("log.txt", mode="a") as log:
log.write(message)
@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
message = f"message to {email}"
background_tasks.add_task(write_log, message)
return {"message": "Message sent in the background"}?/send-notification/{email}, the response is sent immediately with the message "Message sent in the background", and the write_log function is executed in the background to write to a log file. This ensures a responsive API while still allowing for additional background processing. FastAPI application, you would first install pytest and requests libraries. Then, create a test file in your project directory named ‘test_main.py’. In this file, import the necessary modules including FastAPI’s TestClient, pytest, and your main app.FastAPI application as an argument. This client will be used to simulate HTTP requests in your tests.TestClient and checks the response. Use pytest’s assert statements to verify the status code, headers, and body of the response match what is expected.import uvicorn
from starlette.websockets import WebSocket
async def app(scope, receive, send):
websocket = WebSocket(scope, receive, send)
await websocket.accept()
while True:
data = await websocket.receive_text()
# Process data here...
await websocket.send_text(f"Processed: {data}")
if __name__ == "__main__":
uvicorn.run(app, host="127.0.0.1", port=8000)?FastAPI validates the data against the model’s schema using Python’s built-in typing system. FastAPI automatically sends an error response detailing the issue. For valid data, it serializes into JSON format for HTTP responses. This process also works in reverse for incoming JSON data, deserializing it into Python objects. APIRouter class from FastAPI. This class allows you to define routes, dependencies, and other aspects of your API within a specific router instance.from fastapi import APIRouter
router = APIRouter()?@app.get, @app.post, etc.) that you use at the application level. The difference is that you use the decorators on the router instance instead of the main FastAPI application instance.@router.get("/items/")
def read_items():
return {"message": "Read items"}?app.include_router method. This mounts the routes defined in the router to a specific path in the application.from fastapi import FastAPI
app = FastAPI()
app.include_router(router, prefix="/v1")?router will be accessible under the /v1 path.
from fastapi import APIRouter, Depends
router = APIRouter()
def get_db():
# Function to get a database connection
# ...
@router.get("/items/")
def read_items(db: Session = Depends(get_db)):
return {"message": "Read items"}?APIRouter instances within a router. This allows for hierarchical organization of routes within your application.from fastapi import APIRouter
parent_router = APIRouter()
child_router = APIRouter()
@parent_router.get("/parent/")
def parent_route():
return {"message": "Parent route"}
@child_router.get("/child/")
def child_route():
return {"message": "Child route"}
parent_router.include_router(child_router, prefix="/child")?/child/ route is a sub-route of the /parent/ route.UTF8JSONResponse is a class used in FastAPI to return responses that are encoded in UTF-8 and formatted as JSON. This is useful for APIs that need to support international characters, as UTF-8 is a widely used encoding that can represent most languages.
OpenAPI specifications are important because they provide a standard, language-agnostic way of describing REST APIs. This allows developers to more easily understand how an API works, and also allows for tools to be built that can automatically generate code or documentation based on the OpenAPI specification.
FastAPI include :{} within the route path. They are positional and must be included in the order they appear in the route definition.from fastapi import FastAPI
app = FastAPI()
@app.get("/items/{item_id}")
def read_item(item_id: int):
return {"item_id": item_id}?item_id is a path parameter that is part of the URL path /items/{item_id}.? and are usually in the form of key=value pairs. They are optional and can be provided in any order.from fastapi import FastAPI
app = FastAPI()
@app.get("/items/")
def read_item(query_param: str):
return {"query_param": query_param}?query_param is a query parameter that can be provided in the URL, e.g., /items/?query_param=value.from fastapi import FastAPI
app = FastAPI()
@app.get("/items/")
def read_item(query_param: str = "default_value"):
return {"query_param": query_param}?query_param, the default value "default_value" will be used.from fastapi import FastAPI
app = FastAPI()
# Path parameter
@app.get("/items/{item_id}")
def read_item_by_id(item_id: int):
return {"item_id": item_id}
# Query parameter with a default value
@app.get("/items/")
def read_items_with_query_param(query_param: str = "default_value"):
return {"query_param": query_param}?/items/{item_id} expects a path parameter (item_id) in the URL path, while /items/ expects an optional query parameter (query_param) with a default value. built-in middleware for managing CORS, which can be added to the application instance. To enable it, import CORSMiddleware from fastapi.middleware.cors and add it to your FastAPI app using the .add_middleware() method. You need to specify parameters like allow_origins (a list of origins that are allowed), allow_credentials (whether cookies can be supported), allow_methods (HTTP methods allowed), and allow_headers (which HTTP headers are permitted). Here’s an example:from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
origins = ["http://localhost:3000"]
app.add_middleware(
CORSMiddleware,
allow_origins=origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.get("/")
async def main():
return {"message": "Hello World"}?@validator decorator, enabling more complex checks beyond simple type validation. Furthermore, FastAPI can automatically generate interactive documentation for your API including form parameters, making it easier for users to understand how to interact with your endpoints. StaticFiles class from starlette.staticfiles. First, import FastAPI and StaticFiles from fastapi and starlette.staticfiles respectively. Then create an instance of FastAPI and mount a new instance of StaticFiles to it. The directory parameter in StaticFiles should point to your static files’ location. For example :from fastapi import FastAPI
from starlette.staticfiles import StaticFiles
app = FastAPI()
app.mount("/static", StaticFiles(directory="static"), name="static")?/static” is the path where your app will look for static files. “directory” is the folder with your static files. FROM tiangolo/uvicorn-gunicorn-fastapi:python3.7
COPY ./app /app?docker build -t myimage . and run it with docker run -d --name mycontainer -p 80:80 myimage.app.middleware decorator to register middleware functions for request and response processing.starlette.middleware.base.BaseHTTPMiddleware. This gives you more control over the middleware behavior and allows for more complex middleware logic.from fastapi import FastAPI, Request, Response
from starlette.middleware.base import BaseHTTPMiddleware
app = FastAPI()
class CustomMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
# Pre-processing logic
# e.g., Authentication, Logging, Metrics
response = await call_next(request)
# Post-processing logic
# e.g., Error Handling, Response Transformation
return response
app.add_middleware(CustomMiddleware)?CustomMiddleware is a custom middleware class that subclasses BaseHTTPMiddleware. The dispatch method intercepts incoming requests and outgoing responses, allowing you to perform pre-processing and post-processing tasks as needed. pip install tartiflette[aiohttp] uvicorn fastapi?# schema.graphql
"""
type Query {
hello: String!
}
"""
# schema.py
from tartiflette import Resolver, create_engine
@Resolver("Query.hello")
async def resolve_hello(parent, args, context, info):
return "Hello, World!"?TartifletteApp class provided by tartiflette.from fastapi import FastAPI
from tartiflette import Engine
from schema import schema
app = FastAPI()
@app.post("/graphql")
async def graphql(request: Request):
engine = Engine(schema)
return await engine.handle_graphql_request(request)?uvicorn.uvicorn main:app --reload?/graphql endpoint.curl -X POST -H "Content-Type: application/json" -d '{"query": "{ hello }"}' http://localhost:8000/graphql?status_code parameter is used in FastAPI route functions : 200 OK for successful responses.status_code parameter in the route function decorator.from fastapi import FastAPI, status
app = FastAPI()
@app.get("/items/", status_code=status.HTTP_201_CREATED)
def create_item():
# Logic to create an item
return {"message": "Item created successfully"}?/items/ endpoint returns a status code of 201 Created instead of the default 200 OK.status_code parameter to specify the appropriate HTTP status code for error responses, such as 404 Not Found, 400 Bad Request, or 500 Internal Server Error.from fastapi import FastAPI, HTTPException, status
app = FastAPI()
@app.get("/items/{item_id}")
def read_item(item_id: int):
# Logic to retrieve item from database
item = ...
if not item:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Item not found")
return item?HTTPException with a status code of 404 Not Found.status_code parameter in FastAPI route functions allows you to customize the HTTP status codes returned by your API endpoints, providing clear and consistent communication of the request outcomes to clients. cookies’ parameter in path operation functions. SessionMiddleware to your application, providing a secret key. This middleware uses signed cookies to store session data client-side. The data is cryptographically signed but not encrypted, so user can see contents but cannot modify them without invalidating signature. OAuth2 with Password and Bearer, a standard for user authentication, allowing secure access to resources. This is achieved by using Python-Jose to encrypt and verify tokens. FastAPI also offers HTTPBasicAuth for simpler cases where username and password are required.Cross-Site Scripting (XSS) and SQL Injection attacks by default. It uses Pydantic models which automatically validate incoming JSON requests, preventing malicious code from being executed. from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def read_root():
return {"Hello": "World"}?from pydantic import BaseModel
class Item(BaseModel):
name: str
description: str = None
price: float
tax: float = None?from fastapi import FastAPI
app = FastAPI()
@app.post("/items/")
def create_item(item: Item):
return item?/docs or /redoc endpoints in your browser, you will see the automatically generated API documentation.from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
name: str
description: str = None
price: float
tax: float = None
@app.post("/items/")
def create_item(item: Item):
return item?http://localhost:8000/docs in your browser, you will see the automatically generated API documentation with interactive features for testing endpoints.