FastAPI Core Concepts
FastAPI is a modern, fast (high-performance) Python web framework for building APIs. It is built on standard Python type hints, using Starlette and Pydantic.
Main Features
- High Performance: Comparable to NodeJS and Go, one of the fastest Python frameworks
- Rapid development: Development speed increased by approximately 200%-300%
- Reduce errors: Reduce human errors by about 40%
- Intuitive: Powerful editor support, autocomplete everywhere
- Simple: Easy to use and learn, reducing time spent reading documentation
ASGI and Asynchronous Programming
What is ASGI?
ASGI (Asynchronous Server Gateway Interface) is the standard interface between Python asynchronous web servers and applications.
WSGI vs ASGI comparison:
Example
# WSGI 应用（同步）- 传统 Flask 风格
def wsgi_app(environ, start_response):
status = '200 OK'
response_headers = [('Content-type', 'text/plain')]
start_response(status, response_headers)
return [b'Hello World']
# ASGI application (async) - FastAPI style
async def asgi_app(scope, receive, send):
await send({
'type': 'http.response.start',
'status': 200,
'headers': [(b'content-type', b'text/plain')],
})
await send({
'type': 'http.response.body',
'body': b'Hello World',
})
def wsgi_app(environ, start_response):
status = '200 OK'
response_headers = [('Content-type', 'text/plain')]
start_response(status, response_headers)
return [b'Hello World']
# ASGI application (async) - FastAPI style
async def asgi_app(scope, receive, send):
await send({
'type': 'http.response.start',
'status': 200,
'headers': [(b'content-type', b'text/plain')],
})
await send({
'type': 'http.response.body',
'body': b'Hello World',
})
Core Concepts of Asynchronous Programming
Synchronous vs Asynchronous Execution:
Example
import asyncio
import time
import httpx
# Synchronous mode - blocking execution
def sync_fetch_data():
start_time = time.time()
# Simulate three network requests
time.sleep(1) # First request
time.sleep(1) # Second request
time.sleep(1) # Third request
print(f"Elapsed time for synchronous execution: {time.time() - start_time:.2f} seconds") # about 3 seconds
# Asynchronous mode - concurrent execution
async def async_fetch_data():
start_time = time.time()
# Execute three requests concurrently
await asyncio.gather(
asyncio.sleep(1), # First request
asyncio.sleep(1), # Second request
asyncio.sleep(1), # Third request
)
print(fAsynchronous execution time: {time.time() - start_time:.2f} seconds) # about 1 second
# Run Example
sync_fetch_data() # Output: Synchronous execution time: 3.00 seconds
asyncio.run(async_fetch_data()) # Output: Asynchronous execution time: 1.00 seconds
import time
import httpx
# Synchronous mode - blocking execution
def sync_fetch_data():
start_time = time.time()
# Simulate three network requests
time.sleep(1) # First request
time.sleep(1) # Second request
time.sleep(1) # Third request
print(f"Elapsed time for synchronous execution: {time.time() - start_time:.2f} seconds") # about 3 seconds
# Asynchronous mode - concurrent execution
async def async_fetch_data():
start_time = time.time()
# Execute three requests concurrently
await asyncio.gather(
asyncio.sleep(1), # First request
asyncio.sleep(1), # Second request
asyncio.sleep(1), # Third request
)
print(fAsynchronous execution time: {time.time() - start_time:.2f} seconds) # about 1 second
# Run Example
sync_fetch_data() # Output: Synchronous execution time: 3.00 seconds
asyncio.run(async_fetch_data()) # Output: Asynchronous execution time: 1.00 seconds
Asynchronous path operations in FastAPI:
Example
from fastapi import FastAPI
import asyncio
app = FastAPI()
# Synchronous Path Operation
@app.get("/sync")
def sync_endpoint():
# Synchronous operations block the entire application
time.sleep(2)
return {"message": "Synchronous Response"}
# Asynchronous path operation (recommended)
@app.get("/async")
async def async_endpoint():
# Asynchronous operations do not block other requests
await asyncio.sleep(2)
return {"message": "Asynchronous Response"}
# Asynchronous database operation example
@app.get("/users/{user_id}")
async def get_user(user_id: int):
# Asynchronous database query
user = await database.fetch_one(
"SELECT * FROM users WHERE id = :user_id",
{"user_id": user_id}
)
return user
import asyncio
app = FastAPI()
# Synchronous Path Operation
@app.get("/sync")
def sync_endpoint():
# Synchronous operations block the entire application
time.sleep(2)
return {"message": "Synchronous Response"}
# Asynchronous path operation (recommended)
@app.get("/async")
async def async_endpoint():
# Asynchronous operations do not block other requests
await asyncio.sleep(2)
return {"message": "Asynchronous Response"}
# Asynchronous database operation example
@app.get("/users/{user_id}")
async def get_user(user_id: int):
# Asynchronous database query
user = await database.fetch_one(
"SELECT * FROM users WHERE id = :user_id",
{"user_id": user_id}
)
return user
When to use async?
Scenarios Suitable for Asynchronous Operations:
- Database operation
- Network requests (API calls)
- File I/O operations
- Operations with long wait times
Scenarios Not Suitable for Asynchronous Operations:
- CPU-intensive computation
- Simple data processing
- Operations without I/O wait
Example
# Good Async Usage
@app.get("/weather")
async def get_weather():
async with httpx.AsyncClient() as client:
response = await client.get("https://api.weather.com/data")
return response.json()
# Unnecessary async (no I/O wait)
@app.get("/calculate")
def calculate_sync(): # Just Keep It Synchronous
result = sum(range(1000000))
return {"result": result}
@app.get("/weather")
async def get_weather():
async with httpx.AsyncClient() as client:
response = await client.get("https://api.weather.com/data")
return response.json()
# Unnecessary async (no I/O wait)
@app.get("/calculate")
def calculate_sync(): # Just Keep It Synchronous
result = sum(range(1000000))
return {"result": result}
REST API Design Principles
REST Fundamentals
REST (Representational State Transfer) is a Web API design style that emphasizes:
- Resource-oriented: URLs represent resources
- Stateless: Each request is independent
- Uniform interface: Use standard HTTP methods
- Layered system: Supports caching, load balancing, etc.
RESTful URL design
Examples of Good URL Design:
Example
from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel
from typing import List, Optional
app = FastAPI()
# Resource collections and individual resources
@app.get("/users") # GET /users - Get user list
async def get_users():
return users_db
@app.get("/users/{user_id}") # GET /users/123 - Get a specific user
async def get_user(user_id: int):
return find_user(user_id)
@app.post("/users") # POST /users - Create new user
async def create_user(user: UserCreate):
return create_new_user(user)
@app.put("/users/{user_id}") # PUT /users/123 - Full update user
async def update_user(user_id: int, user: UserUpdate):
return update_existing_user(user_id, user)
@app.patch("/users/{user_id}") # PATCH /users/123 - partially update user
async def patch_user(user_id: int, user: UserPatch):
return patch_existing_user(user_id, user)
@app.delete("/users/{user_id}") # DELETE /users/123 - 删除用户
async def delete_user(user_id: int):
return delete_existing_user(user_id)
# Nested Resources
@app.get("/users/{user_id}/posts") # Get all articles of the user
async def get_user_posts(user_id: int):
return get_posts_by_user(user_id)
@app.post("/users/{user_id}/posts") # Create a new article for the user
async def create_user_post(user_id: int, post: PostCreate):
return create_post_for_user(user_id, post)
from pydantic import BaseModel
from typing import List, Optional
app = FastAPI()
# Resource collections and individual resources
@app.get("/users") # GET /users - Get user list
async def get_users():
return users_db
@app.get("/users/{user_id}") # GET /users/123 - Get a specific user
async def get_user(user_id: int):
return find_user(user_id)
@app.post("/users") # POST /users - Create new user
async def create_user(user: UserCreate):
return create_new_user(user)
@app.put("/users/{user_id}") # PUT /users/123 - Full update user
async def update_user(user_id: int, user: UserUpdate):
return update_existing_user(user_id, user)
@app.patch("/users/{user_id}") # PATCH /users/123 - partially update user
async def patch_user(user_id: int, user: UserPatch):
return patch_existing_user(user_id, user)
@app.delete("/users/{user_id}") # DELETE /users/123 - 删除用户
async def delete_user(user_id: int):
return delete_existing_user(user_id)
# Nested Resources
@app.get("/users/{user_id}/posts") # Get all articles of the user
async def get_user_posts(user_id: int):
return get_posts_by_user(user_id)
@app.post("/users/{user_id}/posts") # Create a new article for the user
async def create_user_post(user_id: int, post: PostCreate):
return create_post_for_user(user_id, post)
URL Design Principles:
好的设计 GET /api/v1/users # 获取用户列表 GET /api/v1/users/123 # 获取用户 123 POST /api/v1/users # 创建用户 GET /api/v1/users/123/orders # 获取用户 123 的订单 # 不好的设计 GET /api/v1/getAllUsers # 动词不应该在 URL 中 GET /api/v1/user/123 # 集合名应该用复数 POST /api/v1/createUser # URL 中包含动作 GET /api/v1/users-orders-123 # 关系不清晰
Hierarchical relationship of resources
Example
# Relationship design between users and articles
class User(BaseModel):
id: int
name: str
email: str
class Post(BaseModel):
id: int
title: str
content: str
author_id: int
# Main resource operation
@app.get("/users")
async def list_users() -> List[User]:
return users
@app.get("/posts")
async def list_posts() -> List[Post]:
return posts
# Related Resource Operations
@app.get("/users/{user_id}/posts")
async def get_user_posts(user_id: int) -> List[Post]:
"""Get all articles for a specific user"""
return [post for post in posts if post.author_id == user_id]
@app.get("/posts/{post_id}/author")
async def get_post_author(post_id: int) -> User:
"""Get the author information of the article"""
post = find_post(post_id)
if not post:
raise HTTPException(status_code=404, detail="Post not found")
return find_user(post.author_id)
class User(BaseModel):
id: int
name: str
email: str
class Post(BaseModel):
id: int
title: str
content: str
author_id: int
# Main resource operation
@app.get("/users")
async def list_users() -> List[User]:
return users
@app.get("/posts")
async def list_posts() -> List[Post]:
return posts
# Related Resource Operations
@app.get("/users/{user_id}/posts")
async def get_user_posts(user_id: int) -> List[Post]:
"""Get all articles for a specific user"""
return [post for post in posts if post.author_id == user_id]
@app.get("/posts/{post_id}/author")
async def get_post_author(post_id: int) -> User:
"""Get the author information of the article"""
post = find_post(post_id)
if not post:
raise HTTPException(status_code=404, detail="Post not found")
return find_user(post.author_id)
HTTP Methods and Status Codes
Semantics of HTTP Methods
Example
from fastapi import FastAPI, status, HTTPException
from fastapi.responses import JSONResponse
app = FastAPI()
# GET - safe and idempotent
@app.get("/users/{user_id}")
async def get_user(user_id: int):
"""Get resource without modifying server state"""
user = find_user(user_id)
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found"
)
return user
# POST - unsafe and non-idempotent
@app.post("/users", status_code=status.HTTP_201_CREATED)
async def create_user(user: UserCreate):
"""Create new resource"""
if user_exists(user.email):
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="User already exists"
)
new_user = create_new_user(user)
return new_user
# PUT - unsafe but idempotent
@app.put("/users/{user_id}")
async def replace_user(user_id: int, user: UserUpdate):
"""Completely replace resource"""
if not user_exists(user_id):
# PUT can create a resource
return create_user_with_id(user_id, user)
return replace_existing_user(user_id, user)
# PATCH - unsafe and usually non-idempotent
@app.patch("/users/{user_id}")
async def update_user(user_id: int, user: UserPatch):
"""Partially update resource"""
existing_user = find_user(user_id)
if not existing_user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found"
)
return patch_user_fields(existing_user, user)
# DELETE - unsafe but idempotent
@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_user(user_id: int):
"""Delete resource"""
if not user_exists(user_id):
# Idempotency: deleting a non-existent resource still returns success
return JSONResponse(status_code=status.HTTP_204_NO_CONTENT)
delete_existing_user(user_id)
return JSONResponse(status_code=status.HTTP_204_NO_CONTENT)
from fastapi.responses import JSONResponse
app = FastAPI()
# GET - safe and idempotent
@app.get("/users/{user_id}")
async def get_user(user_id: int):
"""Get resource without modifying server state"""
user = find_user(user_id)
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found"
)
return user
# POST - unsafe and non-idempotent
@app.post("/users", status_code=status.HTTP_201_CREATED)
async def create_user(user: UserCreate):
"""Create new resource"""
if user_exists(user.email):
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="User already exists"
)
new_user = create_new_user(user)
return new_user
# PUT - unsafe but idempotent
@app.put("/users/{user_id}")
async def replace_user(user_id: int, user: UserUpdate):
"""Completely replace resource"""
if not user_exists(user_id):
# PUT can create a resource
return create_user_with_id(user_id, user)
return replace_existing_user(user_id, user)
# PATCH - unsafe and usually non-idempotent
@app.patch("/users/{user_id}")
async def update_user(user_id: int, user: UserPatch):
"""Partially update resource"""
existing_user = find_user(user_id)
if not existing_user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found"
)
return patch_user_fields(existing_user, user)
# DELETE - unsafe but idempotent
@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_user(user_id: int):
"""Delete resource"""
if not user_exists(user_id):
# Idempotency: deleting a non-existent resource still returns success
return JSONResponse(status_code=status.HTTP_204_NO_CONTENT)
delete_existing_user(user_id)
return JSONResponse(status_code=status.HTTP_204_NO_CONTENT)
Usage of HTTP Status Codes
Example
from fastapi import status
# 2xx success response
@app.post("/users", status_code=status.HTTP_201_CREATED) # 201 Created
@app.get("/users") # 200 OK (default)
@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT) # 204 No Content
# 4xx client errors
@app.get("/users/{user_id}")
async def get_user(user_id: int):
if user_id < 1:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST, # 400 Bad Request
detail="User ID must be positive"
)
user = find_user(user_id)
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, # 404 Not Found
detail="User not found"
)
return user
# Business Logic Error
@app.post("/users/{user_id}/follow")
async def follow_user(user_id: int, current_user_id: int):
if user_id == current_user_id:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, # 422 Unprocessable Entity
detail="Cannot follow yourself"
)
if is_already_following(current_user_id, user_id):
raise HTTPException(
status_code=status.HTTP_409_CONFLICT, # 409 Conflict
detail="Already following this user"
)
# 5xx server error (usually handled by the exception handler)
@app.exception_handler(Exception)
async def general_exception_handler(request, exc):
return JSONResponse(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, # 500 Internal Server Error
content={"detail": "Internal server error"}
)
# 2xx success response
@app.post("/users", status_code=status.HTTP_201_CREATED) # 201 Created
@app.get("/users") # 200 OK (default)
@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT) # 204 No Content
# 4xx client errors
@app.get("/users/{user_id}")
async def get_user(user_id: int):
if user_id < 1:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST, # 400 Bad Request
detail="User ID must be positive"
)
user = find_user(user_id)
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, # 404 Not Found
detail="User not found"
)
return user
# Business Logic Error
@app.post("/users/{user_id}/follow")
async def follow_user(user_id: int, current_user_id: int):
if user_id == current_user_id:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, # 422 Unprocessable Entity
detail="Cannot follow yourself"
)
if is_already_following(current_user_id, user_id):
raise HTTPException(
status_code=status.HTTP_409_CONFLICT, # 409 Conflict
detail="Already following this user"
)
# 5xx server error (usually handled by the exception handler)
@app.exception_handler(Exception)
async def general_exception_handler(request, exc):
return JSONResponse(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, # 500 Internal Server Error
content={"detail": "Internal server error"}
)
Quick Reference for Common Status Codes:
# 成功状态码 200 OK # 请求成功 201 Created # 资源创建成功 204 No Content # 成功但无内容返回 206 Partial Content # 部分内容（分页、断点续传） # 重定向 301 Moved Permanently # 永久重定向 302 Found # 临时重定向 304 Not Modified # 资源未修改（缓存） # 客户端错误 400 Bad Request # 请求格式错误 401 Unauthorized # 未认证 403 Forbidden # 已认证但无权限 404 Not Found # 资源不存在 405 Method Not Allowed # HTTP 方法不允许 409 Conflict # 资源冲突 422 Unprocessable Entity # 数据验证失败 429 Too Many Requests # 请求过于频繁 # 服务器错误 500 Internal Server Error # 服务器内部错误 502 Bad Gateway # 网关错误 503 Service Unavailable # 服务不可用
JSON Data Format
JSON Fundamentals and Best Practices
Example
from datetime import datetime
from decimal import Decimal
from typing import Optional, List
from pydantic import BaseModel, Field
import json
# JSON data format specification
class UserResponse(BaseModel):
id: int
username: str
email: str
full_name: Optional[str] = None
is_active: bool = True
created_at: datetime
updated_at: Optional[datetime] = None
# JSON serialization configuration
class Config:
# Allow field aliases
allow_population_by_field_name = True
# JSON encoder
json_encoders = {
datetime: lambda v: v.isoformat(),
Decimal: lambda v: float(v)
}
# Standardize response format
class APIResponse(BaseModel):
"""Standard API response format"""
success: bool = True
message: str = "Operation Successful"
data: Optional[dict] = None
errors: Optional[List[str]] = None
timestamp: datetime = Field(default_factory=datetime.now)
# Pagination Response Format
class PaginatedResponse(BaseModel):
items: List[dict]
total: int
page: int
size: int
pages: int
@app.get("/users", response_model=PaginatedResponse)
async def get_users(page: int = 1, size: int = 10):
"""Return a paginated list of users"""
users = get_users_paginated(page, size)
total = count_users()
return PaginatedResponse(
items=users,
total=total,
page=page,
size=size,
pages=(total + size - 1) // size
)
from decimal import Decimal
from typing import Optional, List
from pydantic import BaseModel, Field
import json
# JSON data format specification
class UserResponse(BaseModel):
id: int
username: str
email: str
full_name: Optional[str] = None
is_active: bool = True
created_at: datetime
updated_at: Optional[datetime] = None
# JSON serialization configuration
class Config:
# Allow field aliases
allow_population_by_field_name = True
# JSON encoder
json_encoders = {
datetime: lambda v: v.isoformat(),
Decimal: lambda v: float(v)
}
# Standardize response format
class APIResponse(BaseModel):
"""Standard API response format"""
success: bool = True
message: str = "Operation Successful"
data: Optional[dict] = None
errors: Optional[List[str]] = None
timestamp: datetime = Field(default_factory=datetime.now)
# Pagination Response Format
class PaginatedResponse(BaseModel):
items: List[dict]
total: int
page: int
size: int
pages: int
@app.get("/users", response_model=PaginatedResponse)
async def get_users(page: int = 1, size: int = 10):
"""Return a paginated list of users"""
users = get_users_paginated(page, size)
total = count_users()
return PaginatedResponse(
items=users,
total=total,
page=page,
size=size,
pages=(total + size - 1) // size
)
JSON Schema and data validation
Example
from pydantic import BaseModel, validator, Field
from typing import Optional, List
from enum import Enum
class UserRole(str, Enum):
ADMIN = "admin"
USER = "user"
MODERATOR = "moderator"
class UserCreate(BaseModel):
username: str = Field(..., min_length=3, max_length=50, description="Username")
email: str = Field(..., regex=r'^[\w\.-]+@[\w\.-]+\.\w+$', description="Email Address")
password: str = Field(..., min_length=8, description="Password")
full_name: Optional[str] = Field(None, max_length=100, description="Full name")
role: UserRole = Field(UserRole.USER, description="User Role")
age: Optional[int] = Field(None, ge=0, le=150, description="Age")
@validator('username')
def username_must_be_alphanumeric(cls, v):
if not v.isalnum():
raise ValueError('Username can only contain letters and numbers')
return v
@validator('password')
def password_strength(cls, v):
if not any(c.isupper() for c in v):
raise ValueError('Password must contain at least one uppercase letter')
if not any(c.islower() for c in v):
raise ValueError('Password must contain at least one lowercase letter')
if not any(c.isdigit() for c in v):
raise ValueError('Password must contain at least one digit')
return v
class Config:
# Generate JSON Schema example
schema_extra = {
"example": {
"username": "johndoe",
"email": "john@example.com",
"password": "SecurePass123",
"full_name": "John Doe",
"role": "user",
"age": 30
}
}
# Use custom validators
@app.post("/users", response_model=UserResponse)
async def create_user(user: UserCreate):
"""Create a new user, including data validation"""
# Pydantic automatically validates input data
return await create_new_user(user)
from typing import Optional, List
from enum import Enum
class UserRole(str, Enum):
ADMIN = "admin"
USER = "user"
MODERATOR = "moderator"
class UserCreate(BaseModel):
username: str = Field(..., min_length=3, max_length=50, description="Username")
email: str = Field(..., regex=r'^[\w\.-]+@[\w\.-]+\.\w+$', description="Email Address")
password: str = Field(..., min_length=8, description="Password")
full_name: Optional[str] = Field(None, max_length=100, description="Full name")
role: UserRole = Field(UserRole.USER, description="User Role")
age: Optional[int] = Field(None, ge=0, le=150, description="Age")
@validator('username')
def username_must_be_alphanumeric(cls, v):
if not v.isalnum():
raise ValueError('Username can only contain letters and numbers')
return v
@validator('password')
def password_strength(cls, v):
if not any(c.isupper() for c in v):
raise ValueError('Password must contain at least one uppercase letter')
if not any(c.islower() for c in v):
raise ValueError('Password must contain at least one lowercase letter')
if not any(c.isdigit() for c in v):
raise ValueError('Password must contain at least one digit')
return v
class Config:
# Generate JSON Schema example
schema_extra = {
"example": {
"username": "johndoe",
"email": "john@example.com",
"password": "SecurePass123",
"full_name": "John Doe",
"role": "user",
"age": 30
}
}
# Use custom validators
@app.post("/users", response_model=UserResponse)
async def create_user(user: UserCreate):
"""Create a new user, including data validation"""
# Pydantic automatically validates input data
return await create_new_user(user)
Error response format
Example
class ErrorDetail(BaseModel):
field: str
message: str
code: str
class ErrorResponse(BaseModel):
error: str
message: str
details: Optional[List[ErrorDetail]] = None
timestamp: datetime = Field(default_factory=datetime.now)
# Custom exception handling
from fastapi import Request
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
errors = []
for error in exc.errors():
errors.append(ErrorDetail(
field='.'.join(str(x) for x in error['loc']),
message=error['msg'],
code=error['type']
))
return JSONResponse(
status_code=422,
content=ErrorResponse(
error="Validation Error",
message="Request data validation failed",
details=errors
).dict()
)
@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
return JSONResponse(
status_code=exc.status_code,
content=ErrorResponse(
error=f"HTTP {exc.status_code}",
message=exc.detail
).dict()
)
field: str
message: str
code: str
class ErrorResponse(BaseModel):
error: str
message: str
details: Optional[List[ErrorDetail]] = None
timestamp: datetime = Field(default_factory=datetime.now)
# Custom exception handling
from fastapi import Request
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
errors = []
for error in exc.errors():
errors.append(ErrorDetail(
field='.'.join(str(x) for x in error['loc']),
message=error['msg'],
code=error['type']
))
return JSONResponse(
status_code=422,
content=ErrorResponse(
error="Validation Error",
message="Request data validation failed",
details=errors
).dict()
)
@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
return JSONResponse(
status_code=exc.status_code,
content=ErrorResponse(
error=f"HTTP {exc.status_code}",
message=exc.detail
).dict()
)
FastAPI-Specific Concepts
Automatic documentation generation
Example
from fastapi import FastAPI, Query, Path, Body
from fastapi.openapi.utils import get_openapi
app = FastAPI(
title="My API",
description=This is an example API, demonstrating the features of FastAPI,
version="1.0.0",
terms_of_service="http://example.com/terms/",
contact={
"name": "Developer",
"url": "http://example.com/contact/",
"email": "developer@example.com",
},
license_info={
"name": "MIT",
"url": "https://opensource.org/licenses/MIT",
},
)
@app.get(
"/users/{user_id}",
summary=Get User Information,
description="Get detailed user information by user ID",
response_description=User Information Object,
tags=["User Management"]
)
async def get_user(
user_id: int = Path(..., title="User ID", description="User ID to fetch", ge=1),
include_posts: bool = Query(False, title="Include Articles", description=Whether to include the user's article list)
):
"""
Get user information:
- **user_id**: The unique identifier of the user.
**include_posts**: Whether to include the user's post list in the response
Return the user's basic information; if include_posts is True, it will also include the list of articles.
"""
user = find_user(user_id)
if include_posts:
user.posts = get_user_posts(user_id)
return user
# Custom OpenAPI schema
def custom_openapi():
if app.openapi_schema:
return app.openapi_schema
openapi_schema = get_openapi(
title="Custom API documentation",
version="2.5.0",
description="This is a custom OpenAPI schema",
routes=app.routes,
)
# Add custom information
openapi_schema["info"]["x-logo"] = {
"url": "https://example.com/logo.png"
}
app.openapi_schema = openapi_schema
return app.openapi_schema
app.openapi = custom_openapi
from fastapi.openapi.utils import get_openapi
app = FastAPI(
title="My API",
description=This is an example API, demonstrating the features of FastAPI,
version="1.0.0",
terms_of_service="http://example.com/terms/",
contact={
"name": "Developer",
"url": "http://example.com/contact/",
"email": "developer@example.com",
},
license_info={
"name": "MIT",
"url": "https://opensource.org/licenses/MIT",
},
)
@app.get(
"/users/{user_id}",
summary=Get User Information,
description="Get detailed user information by user ID",
response_description=User Information Object,
tags=["User Management"]
)
async def get_user(
user_id: int = Path(..., title="User ID", description="User ID to fetch", ge=1),
include_posts: bool = Query(False, title="Include Articles", description=Whether to include the user's article list)
):
"""
Get user information:
- **user_id**: The unique identifier of the user.
**include_posts**: Whether to include the user's post list in the response
Return the user's basic information; if include_posts is True, it will also include the list of articles.
"""
user = find_user(user_id)
if include_posts:
user.posts = get_user_posts(user_id)
return user
# Custom OpenAPI schema
def custom_openapi():
if app.openapi_schema:
return app.openapi_schema
openapi_schema = get_openapi(
title="Custom API documentation",
version="2.5.0",
description="This is a custom OpenAPI schema",
routes=app.routes,
)
# Add custom information
openapi_schema["info"]["x-logo"] = {
"url": "https://example.com/logo.png"
}
app.openapi_schema = openapi_schema
return app.openapi_schema
app.openapi = custom_openapi
Preview of the Dependency Injection System
Example
from fastapi import Depends
# Simple Dependency
def get_current_user_id() -> int:
# Parse user ID from token
return 123
def get_database_session():
# Create database session
db = DatabaseSession()
try:
yield db
finally:
db.close()
# Using Dependencies
@app.get("/profile")
async def get_profile(
user_id: int = Depends(get_current_user_id),
db = Depends(get_database_session)
):
return get_user_profile(db, user_id)
# dependency chain
def get_current_user(
user_id: int = Depends(get_current_user_id),
db = Depends(get_database_session)
):
return db.query(User).filter(User.id == user_id).first()
@app.get("/dashboard")
async def get_dashboard(
current_user: User = Depends(get_current_user) # Depends on other dependencies
):
return generate_dashboard(current_user)
# Simple Dependency
def get_current_user_id() -> int:
# Parse user ID from token
return 123
def get_database_session():
# Create database session
db = DatabaseSession()
try:
yield db
finally:
db.close()
# Using Dependencies
@app.get("/profile")
async def get_profile(
user_id: int = Depends(get_current_user_id),
db = Depends(get_database_session)
):
return get_user_profile(db, user_id)
# dependency chain
def get_current_user(
user_id: int = Depends(get_current_user_id),
db = Depends(get_database_session)
):
return db.query(User).filter(User.id == user_id).first()
@app.get("/dashboard")
async def get_dashboard(
current_user: User = Depends(get_current_user) # Depends on other dependencies
):
return generate_dashboard(current_user)
The Importance of Type Hints
Example
from typing import List, Optional, Union, Dict, Any
from pydantic import BaseModel
# FastAPI relies entirely on type hints:
# 1. Data validation
# 2. Documentation generation
# 3. Editor Support
@app.get("/items/{item_id}")
async def get_item(
item_id: int, # Path parameter, automatically converted to int
q: Optional[str] = None, # Optional Query Parameters
limit: int = 10 # Query parameters with default values
) -> Dict[str, Any]: # Return Type Hint
"""Type hints tell FastAPI how to handle parameters and responses"""
return {"item_id": item_id, "q": q, "limit": limit}
# Complex Type Hint
@app.post("/items/")
async def create_items(
items: List[ItemCreate] # Receive a list of ItemCreate objects
) -> List[ItemResponse]: # Returns a list of ItemResponse objects
return [create_item(item) for item in items]
# Union type (multiple possible types)
@app.get("/search")
async def search(
q: str,
result_type: str = "json"
) -> Union[List[Dict], str]: # Can return a dictionary, list, or string
if result_type == "json":
return [{"title": "Result 1"}, {"title": "Result 2"}]
else:
return "Result 1, Result 2"
from pydantic import BaseModel
# FastAPI relies entirely on type hints:
# 1. Data validation
# 2. Documentation generation
# 3. Editor Support
@app.get("/items/{item_id}")
async def get_item(
item_id: int, # Path parameter, automatically converted to int
q: Optional[str] = None, # Optional Query Parameters
limit: int = 10 # Query parameters with default values
) -> Dict[str, Any]: # Return Type Hint
"""Type hints tell FastAPI how to handle parameters and responses"""
return {"item_id": item_id, "q": q, "limit": limit}
# Complex Type Hint
@app.post("/items/")
async def create_items(
items: List[ItemCreate] # Receive a list of ItemCreate objects
) -> List[ItemResponse]: # Returns a list of ItemResponse objects
return [create_item(item) for item in items]
# Union type (multiple possible types)
@app.get("/search")
async def search(
q: str,
result_type: str = "json"
) -> Union[List[Dict], str]: # Can return a dictionary, list, or string
if result_type == "json":
return [{"title": "Result 1"}, {"title": "Result 2"}]
else:
return "Result 1, Result 2"
Practical Examples of Core Concepts
Example
from fastapi import FastAPI, HTTPException, status, Depends, Query
from pydantic import BaseModel, Field
from typing import List, Optional
from datetime import datetime
import asyncio
# Comprehensive example: Blog API
app = FastAPI(
title=Blog API,
description="A blog system showcasing FastAPI core concepts",
version="1.0.0"
)
# Data Model
class PostBase(BaseModel):
title: str = Field(..., min_length=1, max_length=200)
content: str = Field(..., min_length=1)
published: bool = Field(True)
class PostCreate(PostBase):
pass
class PostResponse(PostBase):
id: int
author_id: int
created_at: datetime
updated_at: Optional[datetime] = None
class Config:
orm_mode = True
# Mock database
posts_db = []
next_id = 1
# Dependency: get current user (simplified version)
async def get_current_user() -> int:
# In real applications, it will be parsed from the JWT token
return 1
# Asynchronous Path Operation
@app.get("/posts", response_model=List[PostResponse], tags=["Article"])
async def list_posts(
skip: int = Query(0, ge=0, description=Skipped Articles Count),
limit: int = Query(10, ge=1, le=100, description=Returned Articles Count),
published_only: bool = Query(True, description="Only return published articles")
):
"""
Get Article List
Support pagination and filtering
"""
# Simulate asynchronous database query
await asyncio.sleep(0.1)
filtered_posts = posts_db
if published_only:
filtered_posts = [p for p in posts_db if p["published"]]
return filtered_posts[skip:skip + limit]
@app.post("/posts", response_model=PostResponse, status_code=status.HTTP_201_CREATED, tags=["Article"])
async def create_post(
post: PostCreate,
current_user_id: int = Depends(get_current_user)
):
"""
Create new article
Requires User Authentication
"""
global next_id
# Simulate asynchronous database operations
await asyncio.sleep(0.1)
new_post = {
"id": next_id,
"title": post.title,
"content": post.content,
"published": post.published,
"author_id": current_user_id,
"created_at": datetime.now(),
"updated_at": None
}
posts_db.append(new_post)
next_id += 1
return new_post
@app.get("/posts/{post_id}", response_model=PostResponse, tags=["Article"])
async def get_post(post_id: int):
"""
Get Specific Article
Return article details based on article ID
"""
# Simulate asynchronous database query
await asyncio.sleep(0.1)
post = next((p for p in posts_db if p["id"] == post_id), None)
if not post:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Post with id {post_id} not found"
)
return post
# Health Check Endpoint
@app.get("/health", tags=[System])
async def health_check():
"""System health check"""
return {
"status": "healthy",
"timestamp": datetime.now(),
"posts_count": len(posts_db)
}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
from pydantic import BaseModel, Field
from typing import List, Optional
from datetime import datetime
import asyncio
# Comprehensive example: Blog API
app = FastAPI(
title=Blog API,
description="A blog system showcasing FastAPI core concepts",
version="1.0.0"
)
# Data Model
class PostBase(BaseModel):
title: str = Field(..., min_length=1, max_length=200)
content: str = Field(..., min_length=1)
published: bool = Field(True)
class PostCreate(PostBase):
pass
class PostResponse(PostBase):
id: int
author_id: int
created_at: datetime
updated_at: Optional[datetime] = None
class Config:
orm_mode = True
# Mock database
posts_db = []
next_id = 1
# Dependency: get current user (simplified version)
async def get_current_user() -> int:
# In real applications, it will be parsed from the JWT token
return 1
# Asynchronous Path Operation
@app.get("/posts", response_model=List[PostResponse], tags=["Article"])
async def list_posts(
skip: int = Query(0, ge=0, description=Skipped Articles Count),
limit: int = Query(10, ge=1, le=100, description=Returned Articles Count),
published_only: bool = Query(True, description="Only return published articles")
):
"""
Get Article List
Support pagination and filtering
"""
# Simulate asynchronous database query
await asyncio.sleep(0.1)
filtered_posts = posts_db
if published_only:
filtered_posts = [p for p in posts_db if p["published"]]
return filtered_posts[skip:skip + limit]
@app.post("/posts", response_model=PostResponse, status_code=status.HTTP_201_CREATED, tags=["Article"])
async def create_post(
post: PostCreate,
current_user_id: int = Depends(get_current_user)
):
"""
Create new article
Requires User Authentication
"""
global next_id
# Simulate asynchronous database operations
await asyncio.sleep(0.1)
new_post = {
"id": next_id,
"title": post.title,
"content": post.content,
"published": post.published,
"author_id": current_user_id,
"created_at": datetime.now(),
"updated_at": None
}
posts_db.append(new_post)
next_id += 1
return new_post
@app.get("/posts/{post_id}", response_model=PostResponse, tags=["Article"])
async def get_post(post_id: int):
"""
Get Specific Article
Return article details based on article ID
"""
# Simulate asynchronous database query
await asyncio.sleep(0.1)
post = next((p for p in posts_db if p["id"] == post_id), None)
if not post:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Post with id {post_id} not found"
)
return post
# Health Check Endpoint
@app.get("/health", tags=[System])
async def health_check():
"""System health check"""
return {
"status": "healthy",
"timestamp": datetime.now(),
"posts_count": len(posts_db)
}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)