YouTipFastapi Core
FastAPI is a modern, fast (high-performance) Python web framework for building APIs. It is based on standard Python type hints and built on Starlette and Pydantic.
### Key Features
1. **High Performance**: On par with NodeJS and Go, one of the fastest Python frameworks
2. **Fast Development**: Increases development speed by approximately 200%-300%
3. **Fewer Bugs**: Reduces about 40% of human errors
4. **Intuitive**: Great editor support, autocompletion everywhere
5. **Simple**: Easy to use and learn, less 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 application (synchronous) - Traditional Flask style
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 (asynchronous) - 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 approach - 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"Synchronous execution time: {time.time() - start_time:.2f}seconds")# About 3 seconds
# Asynchronous approach - Concurrent execution
async def async_fetch_data():
start_time =time.time()
# Three requests executed concurrently
await asyncio.gather(
asyncio.sleep(1),# First request
asyncio.sleep(1),# Second request
asyncio.sleep(1),# Third request
)
print(f"Asynchronous execution time: {time.time() - start_time:.2f}seconds")# About 1 second
# Run examples
sync_fetch_data()# Output: Synchronous execution time: 3.00seconds
asyncio.run(async_fetch_data())# Output: Asynchronous execution time: 1.00seconds
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 will block the entire application
time.sleep(2)
return{"message": "Synchronous response"}
# Asynchronous path operation (recommended)
@app.get("/async")
async def async_endpoint():
# Asynchronous operations will 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 asynchronous?
**Scenarios suitable for asynchronous:**
* Database operations
* Network requests (API calls)
* File I/O operations
* Long-waiting operations
**Scenarios not suitable for asynchronous:**
* CPU-intensive computations
* Simple data processing
* Operations without I/O waiting
## Example
# Good asynchronous 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 asynchronous (no I/O waiting)
@app.get("/calculate")
def calculate_sync(): # Keep it synchronous
result =sum(range(1000000))
return{"result": result}
* * *
## REST API Design Principles
### Basic REST Concepts
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
**Good URL design examples:**
## Example
from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel
from typing import List, Optional
app = FastAPI()
# Resource collection and single resource
@app.get("/users")# GET /users - Get user list
async def get_users():
return users_db
@app.get("/users/{user_id}")# GET /users/123 - Get 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 of 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 - Partial update of 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 - Delete user
async def delete_user(user_id: int):
return delete_existing_user(user_id)
# Nested resources
@app.get("/users/{user_id}/posts")# Get all posts of a 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 post for a user
async def create_user_post(user_id: int, post: PostCreate):
return create_post_for_user(user_id, post)
URL design principles:
Good design GET /api/v1/users # Get user list GET /api/v1/users/123 # Get user 123 POST /api/v1/users # Create user GET /api/v1/users/123/orders # Get orders for user 123# Bad design GET /api/v1/getAllUsers # Verbs should not be in the URL GET /api/v1/user/123 # Collection names should be plural POST /api/v1/createUser # URL contains an action GET /api/v1/users-orders-123 # Unclear relationship
### Resource Hierarchy
## Example
# User and post relationship design
class User(BaseModel):
id: int
name: str
email: str
class Post(BaseModel):
id: int
title: str
content: str
author_id: int
# Main resource operations
@app.get("/users")
async def list_users() -> List:
return users
@app.get("/posts")
async def list_posts() -> List:
return posts
# Associated resource operations
@app.get("/users/{user_id}/posts")
async def get_user_posts(user_id: int) -> List:
"""Get all posts 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 a post"""
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):
"""Retrieve a resource, does not modify 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 a 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 a 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 a 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 a 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