Build high-performance APIs with FastAPI framework, automatic documentation, and modern Python features
FastAPI is a modern, fast web framework for building APIs with Python based on standard Python type hints. It's designed to be easy to use, fast to code, and production-ready with automatic API documentation.
Why FastAPI?
What You'll Build:
A complete e-commerce API with products, orders, authentication, file uploads, and real-time features - showcasing FastAPI's enterprise capabilities.
Performance Benefits:
FastAPI is built on top of Starlette and Pydantic for high performance and data validation:
Core Components:
Basic FastAPI Application:
1from fastapi import FastAPI, HTTPException
2from pydantic import BaseModel
3from typing import List, Optional
4
5app = FastAPI(
6 title="E-commerce API",
7 description="High-performance e-commerce API built with FastAPI",
8 version="1.0.0"
9)
10
11class Product(BaseModel):
12 id: Optional[int] = None
13 name: str
14 price: float
15 description: Optional[str] = None
16 category_id: int
17
18@app.get("/")
19async def root():
20 return {"message": "Welcome to FastAPI E-commerce API"}
21
22@app.get("/products/", response_model=List[Product])
23async def get_products():
24 return [
25 Product(id=1, name="Laptop", price=999.99, category_id=1),
26 Product(id=2, name="Mouse", price=29.99, category_id=2)
27 ]
28
29@app.post("/products/", response_model=Product)
30async def create_product(product: Product):
31 # Automatic validation of request body
32 return productKey Features:
Pydantic provides powerful data validation using Python type hints:
Advanced Pydantic Models:
1from pydantic import BaseModel, validator, Field
2from typing import Optional, List
3from datetime import datetime
4from enum import Enum
5
6class CategoryEnum(str, Enum):
7 electronics = "electronics"
8 clothing = "clothing"
9 books = "books"
10
11class Product(BaseModel):
12 id: Optional[int] = None
13 name: str = Field(..., min_length=1, max_length=100)
14 price: float = Field(..., gt=0, description="Price must be positive")
15 description: Optional[str] = Field(None, max_length=500)
16 category: CategoryEnum
17 in_stock: bool = True
18 created_at: Optional[datetime] = None
19
20 @validator('name')
21 def validate_name(cls, v):
22 if not v.strip():
23 raise ValueError('Name cannot be empty')
24 return v.title()
25
26 @validator('price')
27 def validate_price(cls, v):
28 if v > 10000:
29 raise ValueError('Price cannot exceed $10,000')
30 return round(v, 2)
31
32 class Config:
33 schema_extra = {
34 "example": {
35 "name": "MacBook Pro",
36 "price": 1999.99,
37 "description": "Professional laptop",
38 "category": "electronics"
39 }
40 }Validation Features:
FastAPI's dependency injection system enables clean, testable code:
Dependency Injection:
1from fastapi import Depends, HTTPException, status
2from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
3
4security = HTTPBearer()
5
6async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
7 # Validate JWT token
8 token = credentials.credentials
9 user = validate_jwt_token(token)
10 if not user:
11 raise HTTPException(
12 status_code=status.HTTP_401_UNAUTHORIZED,
13 detail="Invalid authentication credentials"
14 )
15 return user
16
17async def get_db():
18 db = SessionLocal()
19 try:
20 yield db
21 finally:
22 db.close()
23
24@app.get("/protected/")
25async def protected_route(
26 current_user: User = Depends(get_current_user),
27 db: Session = Depends(get_db)
28):
29 return {"user": current_user.username, "data": "protected"}Middleware:
1from fastapi.middleware.cors import CORSMiddleware
2from fastapi.middleware.gzip import GZipMiddleware
3import time
4
5# CORS Middleware
6app.add_middleware(
7 CORSMiddleware,
8 allow_origins=["https://yourdomain.com"],
9 allow_credentials=True,
10 allow_methods=["*"],
11 allow_headers=["*"],
12)
13
14# Compression Middleware
15app.add_middleware(GZipMiddleware, minimum_size=1000)
16
17# Custom Middleware
18@app.middleware("http")
19async def add_process_time_header(request: Request, call_next):
20 start_time = time.time()
21 response = await call_next(request)
22 process_time = time.time() - start_time
23 response.headers["X-Process-Time"] = str(process_time)
24 return responseBuild a production-ready e-commerce API with FastAPI featuring products, orders, authentication, file uploads, and real-time notifications. Demonstrates FastAPI best practices and enterprise patterns.
1# š COMPLETE FASTAPI E-COMMERCE API
2# High-performance API with automatic documentation and validation
3
4from fastapi import FastAPI, HTTPException, Depends, status, File, UploadFile
5from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
6from fastapi.middleware.cors import CORSMiddleware
7from fastapi.middleware.gzip import GZipMiddleware
8from pydantic import BaseModel, validator, Field
9from typing import List, Optional, Dict, Any
10from datetime import datetime, timedelta
11from enum import Enum
12import asyncio
13import json
14
15# FastAPI Application Setup
16app = FastAPI(
17 title="E-commerce API",
18 description="High-performance e-commerce API built with FastAPI",
19 version="2.0.0",
20 docs_url="/docs",
21 redoc_url="/redoc",
22 openapi_url="/openapi.json"
23)
24
25# Middleware Configuration
26app.add_middleware(
27 CORSMiddleware,
28 allow_origins=["https://yourdomain.com", "http://localhost:3000"],
29 allow_credentials=True,
30 allow_methods=["*"],
31 allow_headers=["*"],
32)
33
34app.add_middleware(GZipMiddleware, minimum_size=1000)
35
36# Pydantic Models with Advanced Validation
37class CategoryEnum(str, Enum):
38 electronics = "electronics"
39 clothing = "clothing"
40 books = "books"
41 home = "home"
42 sports = "sports"
43
44class OrderStatus(str, Enum):
45 pending = "pending"
46 processing = "processing"
47 shipped = "shipped"
48 delivered = "delivered"
49 cancelled = "cancelled"
50
51class ProductBase(BaseModel):
52 name: str = Field(..., min_length=1, max_length=100, description="Product name")
53 price: float = Field(..., gt=0, le=10000, description="Product price in USD")
54 description: Optional[str] = Field(None, max_length=1000)
55 category: CategoryEnum
56 in_stock: bool = True
57 stock_quantity: int = Field(..., ge=0, description="Available stock")
58
59 @validator('name')
60 def validate_name(cls, v):
61 if not v.strip():
62 raise ValueError('Name cannot be empty')
63 return v.title()
64
65 @validator('price')
66 def validate_price(cls, v):
67 return round(v, 2)
68
69class Product(ProductBase):
70 id: int
71 created_at: datetime
72 updated_at: datetime
73
74 class Config:
75 schema_extra = {
76 "example": {
77 "id": 1,
78 "name": "MacBook Pro 16"",
79 "price": 2499.99,
80 "description": "Professional laptop with M1 chip",
81 "category": "electronics",
82 "in_stock": True,
83 "stock_quantity": 50,
84 "created_at": "2024-01-01T00:00:00Z",
85 "updated_at": "2024-01-01T00:00:00Z"
86 }
87 }
88
89class ProductCreate(ProductBase):
90 pass
91
92class ProductUpdate(BaseModel):
93 name: Optional[str] = Field(None, min_length=1, max_length=100)
94 price: Optional[float] = Field(None, gt=0, le=10000)
95 description: Optional[str] = Field(None, max_length=1000)
96 category: Optional[CategoryEnum] = None
97 in_stock: Optional[bool] = None
98 stock_quantity: Optional[int] = Field(None, ge=0)
99
100class User(BaseModel):
101 id: int
102 username: str
103 email: str
104 full_name: Optional[str] = None
105 is_active: bool = True
106 created_at: datetime
107
108class UserCreate(BaseModel):
109 username: str = Field(..., min_length=3, max_length=50)
110 email: str = Field(..., description="Valid email address")
111 password: str = Field(..., min_length=8, description="Minimum 8 characters")
112 full_name: Optional[str] = Field(None, max_length=100)
113
114 @validator('email')
115 def validate_email(cls, v):
116 if '@' not in v or '.' not in v:
117 raise ValueError('Invalid email format')
118 return v.lower()
119
120class OrderItem(BaseModel):
121 product_id: int
122 quantity: int = Field(..., gt=0, le=100)
123 price: float = Field(..., gt=0)
124
125class Order(BaseModel):
126 id: int
127 user_id: int
128 items: List[OrderItem]
129 total_amount: float
130 status: OrderStatus
131 shipping_address: str
132 created_at: datetime
133 updated_at: datetime
134
135class OrderCreate(BaseModel):
136 items: List[OrderItem] = Field(..., min_items=1)
137 shipping_address: str = Field(..., min_length=10, max_length=500)
138
139# Authentication & Security
140security = HTTPBearer()
141
142class AuthManager:
143 """Authentication and authorization management"""
144
145 @staticmethod
146 def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
147 """Create JWT access token"""
148 to_encode = data.copy()
149 if expires_delta:
150 expire = datetime.utcnow() + expires_delta
151 else:
152 expire = datetime.utcnow() + timedelta(minutes=15)
153 to_encode.update({"exp": expire})
154 return f"mock_jwt_token_{data.get('sub', 'user')}" # Mock token for demo
155
156 @staticmethod
157 def verify_token(token: str) -> Optional[Dict[str, Any]]:
158 """Verify JWT token"""
159 if token.startswith("mock_jwt_token_"):
160 username = token.replace("mock_jwt_token_", "")
161 return {"sub": username, "exp": datetime.utcnow() + timedelta(hours=1)}
162 return None
163
164# Dependency Injection
165async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)) -> User:
166 """Get current authenticated user"""
167 token_data = AuthManager.verify_token(credentials.credentials)
168 if not token_data:
169 raise HTTPException(
170 status_code=status.HTTP_401_UNAUTHORIZED,
171 detail="Invalid authentication credentials",
172 headers={"WWW-Authenticate": "Bearer"},
173 )
174
175 # Mock user for demonstration
176 return User(
177 id=1,
178 username=token_data["sub"],
179 email=f"{token_data['sub']}@example.com",
180 full_name="Demo User",
181 created_at=datetime.utcnow()
182 )
183
184async def get_admin_user(current_user: User = Depends(get_current_user)) -> User:
185 """Require admin privileges"""
186 if current_user.username != "admin":
187 raise HTTPException(
188 status_code=status.HTTP_403_FORBIDDEN,
189 detail="Admin privileges required"
190 )
191 return current_user
192
193# Mock Database
194class MockDatabase:
195 """Mock database for demonstration"""
196 def __init__(self):
197 self.products = [
198 Product(
199 id=1,
200 name="MacBook Pro 16"",
201 price=2499.99,
202 description="Professional laptop with M1 chip",
203 category=CategoryEnum.electronics,
204 in_stock=True,
205 stock_quantity=50,
206 created_at=datetime.utcnow(),
207 updated_at=datetime.utcnow()
208 ),
209 Product(
210 id=2,
211 name="iPhone 14 Pro",
212 price=999.99,
213 description="Latest iPhone with advanced camera",
214 category=CategoryEnum.electronics,
215 in_stock=True,
216 stock_quantity=100,
217 created_at=datetime.utcnow(),
218 updated_at=datetime.utcnow()
219 ),
220 Product(
221 id=3,
222 name="Python Programming Book",
223 price=49.99,
224 description="Complete guide to Python programming",
225 category=CategoryEnum.books,
226 in_stock=True,
227 stock_quantity=25,
228 created_at=datetime.utcnow(),
229 updated_at=datetime.utcnow()
230 )
231 ]
232 self.orders = []
233 self.next_product_id = 4
234 self.next_order_id = 1
235
236 async def get_products(self, category: Optional[CategoryEnum] = None, skip: int = 0, limit: int = 100):
237 products = self.products
238 if category:
239 products = [p for p in products if p.category == category]
240 return products[skip:skip + limit]
241
242 async def get_product(self, product_id: int):
243 return next((p for p in self.products if p.id == product_id), None)
244
245 async def create_product(self, product: ProductCreate):
246 new_product = Product(
247 id=self.next_product_id,
248 **product.dict(),
249 created_at=datetime.utcnow(),
250 updated_at=datetime.utcnow()
251 )
252 self.products.append(new_product)
253 self.next_product_id += 1
254 return new_product
255
256 async def create_order(self, order: OrderCreate, user_id: int):
257 total_amount = sum(item.price * item.quantity for item in order.items)
258 new_order = Order(
259 id=self.next_order_id,
260 user_id=user_id,
261 items=order.items,
262 total_amount=total_amount,
263 status=OrderStatus.pending,
264 shipping_address=order.shipping_address,
265 created_at=datetime.utcnow(),
266 updated_at=datetime.utcnow()
267 )
268 self.orders.append(new_order)
269 self.next_order_id += 1
270 return new_order
271
272# Database dependency
273db = MockDatabase()
274
275async def get_database() -> MockDatabase:
276 return db
277
278# API Routes
279@app.get("/", tags=["Root"])
280async def root():
281 """Welcome endpoint with API information"""
282 return {
283 "message": "Welcome to FastAPI E-commerce API",
284 "version": "2.0.0",
285 "docs": "/docs",
286 "features": [
287 "High-performance async API",
288 "Automatic validation",
289 "Interactive documentation",
290 "Type safety",
291 "Dependency injection"
292 ]
293 }
294
295@app.get("/health", tags=["Health"])
296async def health_check():
297 """Health check endpoint"""
298 return {
299 "status": "healthy",
300 "timestamp": datetime.utcnow(),
301 "version": "2.0.0"
302 }
303
304# Product Routes
305@app.get("/products/", response_model=List[Product], tags=["Products"])
306async def get_products(
307 category: Optional[CategoryEnum] = None,
308 skip: int = 0,
309 limit: int = 100,
310 database: MockDatabase = Depends(get_database)
311):
312 """Get all products with optional category filtering"""
313 return await database.get_products(category, skip, limit)
314
315@app.get("/products/{product_id}", response_model=Product, tags=["Products"])
316async def get_product(
317 product_id: int,
318 database: MockDatabase = Depends(get_database)
319):
320 """Get a specific product by ID"""
321 product = await database.get_product(product_id)
322 if not product:
323 raise HTTPException(status_code=404, detail="Product not found")
324 return product
325
326@app.post("/products/", response_model=Product, tags=["Products"])
327async def create_product(
328 product: ProductCreate,
329 database: MockDatabase = Depends(get_database),
330 current_user: User = Depends(get_admin_user)
331):
332 """Create a new product (admin only)"""
333 return await database.create_product(product)
334
335@app.put("/products/{product_id}", response_model=Product, tags=["Products"])
336async def update_product(
337 product_id: int,
338 product_update: ProductUpdate,
339 database: MockDatabase = Depends(get_database),
340 current_user: User = Depends(get_admin_user)
341):
342 """Update a product (admin only)"""
343 existing_product = await database.get_product(product_id)
344 if not existing_product:
345 raise HTTPException(status_code=404, detail="Product not found")
346
347 # Update fields
348 update_data = product_update.dict(exclude_unset=True)
349 for field, value in update_data.items():
350 setattr(existing_product, field, value)
351 existing_product.updated_at = datetime.utcnow()
352
353 return existing_product
354
355# Order Routes
356@app.post("/orders/", response_model=Order, tags=["Orders"])
357async def create_order(
358 order: OrderCreate,
359 database: MockDatabase = Depends(get_database),
360 current_user: User = Depends(get_current_user)
361):
362 """Create a new order"""
363 # Validate products exist and are in stock
364 for item in order.items:
365 product = await database.get_product(item.product_id)
366 if not product:
367 raise HTTPException(status_code=404, detail=f"Product {item.product_id} not found")
368 if not product.in_stock or product.stock_quantity < item.quantity:
369 raise HTTPException(status_code=400, detail=f"Insufficient stock for product {item.product_id}")
370
371 return await database.create_order(order, current_user.id)
372
373@app.get("/orders/", response_model=List[Order], tags=["Orders"])
374async def get_user_orders(
375 current_user: User = Depends(get_current_user),
376 database: MockDatabase = Depends(get_database)
377):
378 """Get current user's orders"""
379 return [order for order in database.orders if order.user_id == current_user.id]
380
381# File Upload
382@app.post("/upload/", tags=["Files"])
383async def upload_file(
384 file: UploadFile = File(...),
385 current_user: User = Depends(get_current_user)
386):
387 """Upload a file"""
388 if not file.content_type.startswith('image/'):
389 raise HTTPException(status_code=400, detail="Only image files are allowed")
390
391 contents = await file.read()
392 return {
393 "filename": file.filename,
394 "content_type": file.content_type,
395 "size": len(contents),
396 "message": "File uploaded successfully"
397 }
398
399# WebSocket for Real-time Updates
400@app.websocket("/ws")
401async def websocket_endpoint(websocket):
402 """WebSocket endpoint for real-time updates"""
403 await websocket.accept()
404 try:
405 while True:
406 # Send periodic updates
407 await asyncio.sleep(5)
408 await websocket.send_json({
409 "type": "heartbeat",
410 "timestamp": datetime.utcnow().isoformat(),
411 "message": "Connection active"
412 })
413 except Exception as e:
414 print(f"WebSocket error: {e}")
415
416# Demonstration
417print("š FastAPI E-commerce API")
418print("=" * 50)
419
420print("\nā
Features Implemented:")
421features = [
422 "High-performance async API endpoints",
423 "Automatic request/response validation",
424 "Interactive API documentation",
425 "JWT authentication with dependencies",
426 "Role-based access control",
427 "File upload handling",
428 "WebSocket real-time communication",
429 "CORS and compression middleware",
430 "Comprehensive error handling",
431 "Type-safe development"
432]
433
434for i, feature in enumerate(features, 1):
435 print(f"{i:2d}. {feature}")
436
437print("\nš API Performance Benefits:")
438print("ā¢ 2-3x faster than Django REST Framework")
439print("ā¢ Automatic OpenAPI documentation")
440print("ā¢ Built-in data validation")
441print("ā¢ Async/await support for high concurrency")
442print("ā¢ Type hints for IDE support")
443print("ā¢ Minimal boilerplate code")
444
445print("\nš§ Available Endpoints:")
446endpoints = [
447 "GET /docs - Interactive API documentation",
448 "GET /products/ - List products with filtering",
449 "POST /products/ - Create product (admin)",
450 "GET /products/{id} - Get specific product",
451 "POST /orders/ - Create new order",
452 "GET /orders/ - Get user orders",
453 "POST /upload/ - Upload files",
454 "WS /ws - WebSocket connection"
455]
456
457for endpoint in endpoints:
458 print(f" ā¢ {endpoint}")
459
460print("\nšÆ FastAPI Advantages:")
461print("ā¢ Automatic validation based on type hints")
462print("ā¢ Fast development with less code")
463print("ā¢ Production-ready performance")
464print("ā¢ Modern Python features (async/await)")
465print("ā¢ Excellent developer experience")
466print("ā¢ Built-in security features")
467print("ā¢ Easy testing and debugging")
468print("ā¢ Comprehensive documentation")š FastAPI Architecture Breakdown:
1. Application Setup:
FastAPI(): Main application instance with metadata2. Pydantic Models:
BaseModel: Type-safe data validation3. Dependency Injection:
Depends(): Clean, testable dependency management4. Async Support:
5. Security Features:
6. API Documentation:
7. Error Handling:
This FastAPI implementation demonstrates modern API development with high performance and developer productivity!
Your progress is saved in this browser, on this device only.