🔌 API Development

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

# Resource-oriented URLs
GET    /api/v2/products           # List products
POST   /api/v2/products           # Create product
GET    /api/v2/products/{id}      # Get product
PUT    /api/v2/products/{id}      # Update product
DELETE /api/v2/products/{id}      # Delete product

# Nested resources
GET    /api/v2/products/{id}/reviews     # Product reviews
GET    /api/v2/categories/{id}/products  # Products in category

# Query parameters for filtering, pagination, sorting
GET    /api/v2/products?category=electronics&sort=-price&page=2&limit=20

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

openapi: 3.0.0
info:
  title: Product API
  version: 2.0.0
  description: API for product catalog management

paths:
  /products/{id}:
    get:
      summary: Get product by ID
      parameters:
        - name: id
          in: path
          required: true
          schema:
            type: integer
      responses:
        '200':
          description: Product found
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/Product'
        '404':
          description: Product not found

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

from fastapi import FastAPI, Depends, HTTPException, Security
from fastapi.security import OAuth2PasswordBearer, APIKeyHeader

# Multiple auth strategies
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
api_key_header = APIKeyHeader(name="X-API-Key")

async def get_current_user(token: str = Depends(oauth2_scheme)):
    user = await verify_token(token)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user

async def verify_api_key(api_key: str = Security(api_key_header)):
    if not await is_valid_api_key(api_key):
        raise HTTPException(status_code=403, detail="Invalid API key")
    return api_key

@app.get("/protected")
async def protected_route(user: User = Depends(get_current_user)):
    return {"user": user.email}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

# Response caching
from fastapi_cache import FastAPICache
from fastapi_cache.decorator import cache

@app.get("/products/{id}")
@cache(expire=300)  # Cache for 5 minutes
async def get_product(id: int):
    return await product_service.get(id)

# Async processing
from fastapi import BackgroundTasks

@app.post("/products")
async def create_product(
    product: ProductCreate,
    background_tasks: BackgroundTasks
):
    result = await product_service.create(product)
    background_tasks.add_task(notify_subscribers, result)
    return result

# Connection pooling
from databases import Database

database = Database(
    DATABASE_URL,
    min_size=5,
    max_size=20
)