This guide provides structured answers to the fundamental questions of API endpoint development, from initial design through production deployment.
Table of Contents
- Creating API Endpoints: Tasks, Patterns, Architecture, and Progression
- JSON Formatting with Database Tables and Progression
- Rites of Passage for Showcasing API Development
- API Endpoint Building Templates for Reference
1. Creating API Endpoints: Tasks, Patterns, Architecture, and Progression
Core Tasks in Endpoint Creation
Building API endpoints requires mastering five fundamental areas:
1.1 Resource Modeling
- Identify Nouns: Map business entities (Users, Products, Orders) to URL paths
- URL Structure: Use plural nouns (e.g.,
/api/v1/products,/api/v1/users) - Avoid Verbs: Let HTTP methods define actions, not the URL
- Example:
- ✅
GET /api/v1/products - ❌
GET /api/v1/getProducts
- ✅
1.2 Action Mapping
Map CRUD operations to HTTP verbs:
| Operation | HTTP Method | Endpoint Example | Description |
|---|---|---|---|
| Create | POST | /api/v1/products | Create new resource |
| Read All | GET | /api/v1/products | Retrieve collection |
| Read One | GET | /api/v1/products/{id} | Retrieve single resource |
| Update (Full) | PUT | /api/v1/products/{id} | Replace entire resource |
| Update (Partial) | PATCH | /api/v1/products/{id} | Update specific fields |
| Delete | DELETE | /api/v1/products/{id} | Remove resource |
1.3 Data Serialization & Validation
- Define Schemas: Use JSON schemas or Pydantic models to enforce structure
- Validate Input: Sanitize all incoming data to prevent SQL injection and XSS
- Type Safety: Enforce data types (string, integer, boolean, etc.)
- Required vs Optional: Clearly distinguish mandatory and optional fields
1.4 Security Implementation
- Authentication: Implement JWT (JSON Web Tokens) or OAuth2
- Authorization: Role-based access control (RBAC)
- HTTPS Only: Enforce secure transport layer
- Rate Limiting: Prevent API abuse with throttling
1.5 Performance Controls
- Pagination: Implement offset-based or cursor-based pagination
- Example:
GET /api/v1/products?page=1&limit=20
- Example:
- Filtering: Allow query parameters for searching
- Example:
GET /api/v1/products?category=electronics&price_max=500
- Example:
- Sorting: Enable ordering by fields
- Example:
GET /api/v1/products?sort=price&order=desc
- Example:
Architectural Patterns & Variations
Different use cases demand different architectural approaches:
Pattern 1: RESTful API (Standard)
Best For: Standard CRUD operations, public APIs, microservices
Characteristics:
- Stateless communication
- Resource-oriented URLs
- Standard HTTP status codes
- JSON or XML responses
Example Structure:
GET /api/v1/products → List all products
POST /api/v1/products → Create product
GET /api/v1/products/123 → Get product by ID
PUT /api/v1/products/123 → Update product
DELETE /api/v1/products/123 → Delete productBashStatus Code Usage:
200 OK– Successful GET, PUT, PATCH201 Created– Successful POST204 No Content– Successful DELETE400 Bad Request– Invalid input401 Unauthorized– Missing/invalid auth403 Forbidden– Insufficient permissions404 Not Found– Resource doesn’t exist429 Too Many Requests– Rate limit exceeded500 Internal Server Error– Server error
Pattern 2: GraphQL
Best For: Complex data requirements, mobile apps, reducing over-fetching
Characteristics:
- Single endpoint (typically
/graphql) - Client specifies exactly what data it needs
- Strongly typed schema
- Reduces multiple round trips
Example Query:
query {
product(id: "123") {
name
price
category {
name
}
}
}GraphQLWhen to Use:
- Multiple clients with different data needs
- Complex, nested data structures
- Need to minimize network requests
- Frontend-driven data requirements
Pattern 3: gRPC
Best For: Internal microservices, high-performance requirements
Characteristics:
- Protocol Buffers (binary format)
- HTTP/2 for transport
- Strongly typed contracts
- Bi-directional streaming support
When to Use:
- Service-to-service communication
- Performance is critical
- Type safety is paramount
- Real-time streaming needed
Pattern 4: Event-Driven / Webhooks
Best For: Asynchronous notifications, real-time updates
Characteristics:
- Server pushes to client
- Event-based triggers
- Callback URLs
- Asynchronous processing
Example Events:
{
"event": "order.created",
"timestamp": "2026-01-31T10:30:00Z",
"data": {
"orderId": "ORD-123",
"total": 99.99
}
}JSONWhen to Use:
- Real-time notifications needed
- Reduce polling overhead
- Third-party integrations
- Decoupled architectures
Gradual Progression: From PoC to Production
Phase 1: Basic CRUD (Proof of Concept)
Goal: Validate technical feasibility
Characteristics:
- Single resource endpoint
- Simple GET and POST operations
- In-memory or basic file storage
- No authentication
- Minimal validation
Example:
# Simple Flask endpoint (PoC)
@app.route('/products', methods=['GET'])
def get_products():
return jsonify(products_list)
@app.route('/products', methods=['POST'])
def create_product():
data = request.json
products_list.append(data)
return jsonify(data), 201PythonDeliverable: Working demo showing core functionality
Phase 2: Structured Design
Goal: Establish professional API structure
Additions:
- URL versioning (
/api/v1/) - Proper HTTP status codes
- Input validation with error messages
- Consistent JSON response format
- Basic error handling
Example Response Structure:
{
"success": true,
"data": {
"id": 1,
"name": "Product Name",
"price": 29.99
},
"meta": {
"timestamp": "2026-01-31T10:30:00Z"
}
}JSONError Response:
{
"success": false,
"error": {
"code": 400,
"message": "Validation failed",
"details": [
{
"field": "price",
"issue": "must be a positive number"
}
]
}
}JSONPhase 3: Security & Business Logic
Goal: Production-ready security and filtering
Additions:
- JWT or OAuth2 authentication
- Role-based authorization
- Filtering and search parameters
- Sorting capabilities
- Input sanitization
- CORS configuration
Example with Authentication:
@app.route('/api/v1/products', methods=['GET'])
@jwt_required()
def get_products():
# Extract query parameters
category = request.args.get('category')
sort_by = request.args.get('sort', 'created_at')
# Apply filters
filtered_products = filter_products(category)
sorted_products = sort_products(filtered_products, sort_by)
return jsonify({
"success": true,
"data": sorted_products
})PythonPhase 4: Optimization & Scaling
Goal: Handle production traffic efficiently
Additions:
- Caching (Redis, Memcached)
- Rate limiting per user/IP
- Database query optimization
- Pagination for large datasets
- Response compression (gzip)
- CDN integration for static responses
Pagination Example:
GET /api/v1/products?page=2&limit=20
Response:
{
"success": true,
"data": [...],
"pagination": {
"page": 2,
"limit": 20,
"total": 150,
"totalPages": 8,
"hasNext": true,
"hasPrev": true
}
}BashRate Limiting Headers:
X-RateLimit-Limit: 1000
X-RateLimit-Remaining: 987
X-RateLimit-Reset: 1643630400BashPhase 5: Advanced Architecture
Goal: Enterprise-grade features
Additions:
- Webhooks for event notifications
- gRPC for internal services
- GraphQL endpoints for flexible queries
- API versioning strategy (deprecation notices)
- Comprehensive monitoring and logging
- Auto-scaling infrastructure
- Multi-region deployment
Webhook Registration Example:
POST /api/v1/webhooks
{
"url": "https://client.com/webhook",
"events": ["order.created", "order.updated"],
"secret": "webhook_secret_key"
}Bash2. JSON Formatting with Database Tables and Progression
Fundamental Principles
2.1 Basic Mapping
Transform database columns to JSON keys:
Database Table: products
id | name | price | created_at
------|---------------|--------|--------------------
1 | Laptop | 999.99 | 2026-01-15 10:30:00
2 | Mouse | 29.99 | 2026-01-20 14:00:00BashJSON Output:
[
{
"id": 1,
"name": "Laptop",
"price": 999.99,
"createdAt": "2026-01-15T10:30:00Z"
},
{
"id": 2,
"name": "Mouse",
"price": 29.99,
"createdAt": "2026-01-20T14:00:00Z"
}
]JSON2.2 Naming Conventions
- Database:
snake_case(e.g.,user_id,created_at) - JSON:
camelCase(e.g.,userId,createdAt) - Consistency: Choose one convention and stick to it
2.3 Data Type Mapping
| Database Type | JSON Type | Notes |
|---|---|---|
INTEGER | number | No quotes |
DECIMAL/FLOAT | number | Preserve precision |
VARCHAR/TEXT | string | Use quotes |
BOOLEAN | boolean | true/false, not 1/0 |
DATE/TIMESTAMP | string | ISO 8601 format |
NULL | null | Decide if to include or omit |
JSON/JSONB | object | Direct mapping |
Gradual Progression of JSON Formatting
Progression Phase 1: Flat Mapping (Basic SELECT)
Goal: Simple row-to-JSON conversion
SQL Query:
SELECT id, name, email FROM users;SQLJSON Output:
[
{
"id": 1,
"name": "John Doe",
"email": "john@example.com"
},
{
"id": 2,
"name": "Jane Smith",
"email": "jane@example.com"
}
]JSONImplementation (Python with SQLAlchemy):
@app.route('/api/v1/users', methods=['GET'])
def get_users():
users = db.session.query(User).all()
return jsonify([{
'id': user.id,
'name': user.name,
'email': user.email
} for user in users])PythonProgression Phase 2: Single Record Retrieval
Goal: Access individual resources by ID
Endpoint: GET /api/v1/users/1
SQL Query:
SELECT id, name, email FROM users WHERE id = 1;SQLJSON Output:
{
"id": 1,
"name": "John Doe",
"email": "john@example.com"
}JSONError Handling:
// If ID doesn't exist
{
"success": false,
"error": {
"code": 404,
"message": "User not found"
}
}JSONProgression Phase 3: Input Validation (POST/PUT)
Goal: Accept and validate JSON input for creating/updating records
Endpoint: POST /api/v1/users
Request Body:
{
"name": "Alice Johnson",
"email": "alice@example.com",
"age": 28
}JSONValidation Rules:
name: Required, min 2 characters, max 100email: Required, valid email formatage: Optional, must be positive integer
SQL Insert:
INSERT INTO users (name, email, age)
VALUES ('Alice Johnson', 'alice@example.com', 28);SQLSuccess Response (201 Created):
{
"success": true,
"data": {
"id": 3,
"name": "Alice Johnson",
"email": "alice@example.com",
"age": 28,
"createdAt": "2026-01-31T10:30:00Z"
}
}JSONValidation Error (400 Bad Request):
{
"success": false,
"error": {
"code": 400,
"message": "Validation failed",
"details": [
{
"field": "email",
"issue": "Invalid email format"
}
]
}
}JSONProgression Phase 4: Nested Relationships (JOINs)
Goal: Transform relational data into hierarchical JSON
Database Tables:
-- users table
id | name | email
---|-----------|------------------
1 | John Doe | john@example.com
-- addresses table
id | user_id | type | street
---|---------|----------|------------------
1 | 1 | billing | 123 Main St
2 | 1 | shipping | 456 Oak AveBashSQL Query with JOIN:
SELECT
u.id, u.name, u.email,
json_agg(
json_build_object(
'id', a.id,
'type', a.type,
'street', a.street
)
) as addresses
FROM users u
LEFT JOIN addresses a ON u.id = a.user_id
WHERE u.id = 1
GROUP BY u.id, u.name, u.email;SQLNested JSON Output:
{
"id": 1,
"name": "John Doe",
"email": "john@example.com",
"addresses": [
{
"id": 1,
"type": "billing",
"street": "123 Main St"
},
{
"id": 2,
"type": "shipping",
"street": "456 Oak Ave"
}
]
}JSONAlternative: Separate Endpoints
GET /api/v1/users/1 → User data
GET /api/v1/users/1/addresses → User's addressesBashProgression Phase 5: Partial Updates (PATCH)
Goal: Update specific fields without replacing entire resource
Endpoint: PATCH /api/v1/users/1
Request Body (Update only email):
{
"email": "newemail@example.com"
}JSONSQL Update:
UPDATE users
SET email = 'newemail@example.com',
updated_at = NOW()
WHERE id = 1;SQLResponse (200 OK):
{
"success": true,
"data": {
"id": 1,
"name": "John Doe",
"email": "newemail@example.com",
"updatedAt": "2026-01-31T11:00:00Z"
}
}JSONProgression Phase 6: Advanced JSONB Operations (PostgreSQL)
Goal: Store and query native JSON data efficiently
Database Schema:
CREATE TABLE products (
id SERIAL PRIMARY KEY,
name VARCHAR(255),
metadata JSONB -- Store flexible attributes
);SQLInsert with JSON:
INSERT INTO products (name, metadata)
VALUES (
'Laptop',
'{"brand": "Dell", "specs": {"ram": "16GB", "storage": "512GB SSD"}}'
);SQLQuery JSON Fields:
-- Find products with specific brand
SELECT * FROM products
WHERE metadata->>'brand' = 'Dell';
-- Find products with 16GB RAM
SELECT * FROM products
WHERE metadata->'specs'->>'ram' = '16GB';SQLAdd GIN Index for Performance:
CREATE INDEX idx_products_metadata
ON products USING GIN (metadata);SQLAPI Response:
{
"id": 1,
"name": "Laptop",
"metadata": {
"brand": "Dell",
"specs": {
"ram": "16GB",
"storage": "512GB SSD"
}
}
}JSONDatabase-Level JSON Serialization
PostgreSQL: Using JSON Functions
-- Convert rows to JSON array
SELECT json_agg(
json_build_object(
'id', id,
'name', name,
'price', price
)
) FROM products;
-- Convert single row to JSON
SELECT row_to_json(products.*)
FROM products
WHERE id = 1;SQLSQL Server: FOR JSON Clause
-- Convert to JSON array
SELECT id, name, price
FROM products
FOR JSON PATH;
-- Nested structure
SELECT
u.id, u.name,
(SELECT a.type, a.street
FROM addresses a
WHERE a.user_id = u.id
FOR JSON PATH) AS addresses
FROM users u
FOR JSON PATH;SQLBest Practices for JSON Formatting
- Consistency: Always use the same naming convention (camelCase recommended)
- Omit Null vs Include Null: Decide project-wide whether to include
nullvalues - Date Formatting: Use ISO 8601 format (
2026-01-31T10:30:00Z) - Avoid Deep Nesting: Limit to 2-3 levels; consider separate endpoints instead
- Use DTOs: Don’t expose raw database schema; use Data Transfer Objects
- Field Selection: Allow clients to specify fields (
?fields=id,name,email)
3. Rites of Passage for Showcasing API Development
To professionally demonstrate your API development skills, follow this comprehensive lifecycle approach:
Rite 1: The Contract (Planning & Design)
1.1 Design-First Approach
Why: Establish the API contract before writing code
Required Artifacts:
- OpenAPI Specification (Swagger): YAML/JSON file defining all endpoints
- Use Case Documentation: Clear description of what the API solves
- Data Models: Entity-relationship diagrams
Example OpenAPI Spec:
openapi: 3.0.0
info:
title: Product API
version: 1.0.0
description: API for managing product catalog
paths:
/api/v1/products:
get:
summary: List all products
parameters:
- name: page
in: query
schema:
type: integer
default: 1
- name: limit
in: query
schema:
type: integer
default: 20
responses:
'200':
description: Successful response
content:
application/json:
schema:
type: object
properties:
success:
type: boolean
data:
type: array
items:
$ref: '#/components/schemas/Product'
components:
schemas:
Product:
type: object
required:
- name
- price
properties:
id:
type: integer
readOnly: true
name:
type: string
minLength: 2
maxLength: 255
price:
type: number
minimum: 0
createdAt:
type: string
format: date-time
readOnly: trueJSON1.2 Mock Validation
Tools: Postman Mock Server, Prism, Swagger Codegen
Process:
- Generate mock server from OpenAPI spec
- Create test requests in Postman
- Validate responses match specification
- Get stakeholder approval before coding
Rite 2: Implementation Best Practices
2.1 Professional Code Structure
Demonstrate:
- Separation of Concerns: Models, Views/Controllers, Serializers
- DRY Principle: Reusable components and utilities
- Error Handling: Comprehensive exception management
- Logging: Structured logging for debugging
Project Structure Example:
api_project/
├── app/
│ ├── models/ # Database models
│ ├── serializers/ # JSON validation
│ ├── views/ # Business logic
│ ├── middleware/ # Auth, logging, CORS
│ └── utils/ # Helper functions
├── tests/
│ ├── unit/
│ ├── integration/
│ └── e2e/
├── docs/
│ ├── openapi.yaml
│ └── README.md
├── .env.example
├── requirements.txt
└── docker-compose.ymlBash2.2 Versioning Strategy
Why: Maintain backward compatibility
Approaches:
- URL Versioning:
/api/v1/products,/api/v2/products - Header Versioning:
Accept: application/vnd.api+json; version=1 - Query Parameter:
/api/products?version=1
Deprecation Notice Example:
{
"success": true,
"data": [...],
"meta": {
"deprecation": {
"version": "v1",
"sunset": "2026-12-31",
"message": "This endpoint will be retired. Please migrate to /api/v2/products"
}
}
}JSON2.3 Input Sanitization
Demonstrate Security Awareness:
from bleach import clean
import re
def sanitize_input(data):
"""Remove potential XSS and SQL injection vectors"""
if isinstance(data, str):
# Remove HTML tags
data = clean(data, tags=[], strip=True)
# Remove SQL keywords (basic example)
sql_keywords = ['DROP', 'DELETE', 'INSERT', 'UPDATE', '--', ';']
for keyword in sql_keywords:
data = re.sub(f'\\b{keyword}\\b', '', data, flags=re.IGNORECASE)
return dataPython2.4 Status Code Mastery
Show Semantic Understanding:
from flask import jsonify, make_response
# 200 OK - Successful GET, PUT, PATCH
@app.route('/api/v1/products/<int:id>', methods=['GET'])
def get_product(id):
product = Product.query.get(id)
if not product:
return jsonify({"error": "Not found"}), 404
return jsonify({"data": product.to_dict()}), 200
# 201 Created - Successful POST
@app.route('/api/v1/products', methods=['POST'])
def create_product():
data = request.json
product = Product(**data)
db.session.add(product)
db.session.commit()
response = make_response(jsonify({"data": product.to_dict()}), 201)
response.headers['Location'] = f'/api/v1/products/{product.id}'
return response
# 204 No Content - Successful DELETE
@app.route('/api/v1/products/<int:id>', methods=['DELETE'])
def delete_product(id):
product = Product.query.get_or_404(id)
db.session.delete(product)
db.session.commit()
return '', 204
# 400 Bad Request - Validation error
# 401 Unauthorized - Missing auth
# 403 Forbidden - Insufficient permissions
# 422 Unprocessable Entity - Semantic errorsPythonRite 3: Rigorous Testing (The Proof)
3.1 Test-Driven Development (TDD)
Red-Green-Refactor Workflow:
import pytest
from app import create_app, db
from app.models import Product
@pytest.fixture
def client():
app = create_app('testing')
with app.test_client() as client:
with app.app_context():
db.create_all()
yield client
with app.app_context():
db.drop_all()
# Test: GET /api/v1/products
def test_get_products_empty(client):
"""Test GET request returns empty array when no products"""
response = client.get('/api/v1/products')
assert response.status_code == 200
assert response.json['data'] == []
# Test: POST /api/v1/products
def test_create_product_success(client):
"""Test successful product creation"""
payload = {
"name": "Test Product",
"price": 29.99
}
response = client.post('/api/v1/products', json=payload)
assert response.status_code == 201
assert response.json['data']['name'] == "Test Product"
# Test: Validation error
def test_create_product_invalid_price(client):
"""Test product creation with negative price fails"""
payload = {
"name": "Invalid Product",
"price": -10
}
response = client.post('/api/v1/products', json=payload)
assert response.status_code == 400
assert 'error' in response.json
# Test: Not found
def test_get_product_not_found(client):
"""Test GET request for non-existent product returns 404"""
response = client.get('/api/v1/products/999')
assert response.status_code == 404Python3.2 Edge Case Coverage
Demonstrate Robustness:
- Boundary Testing: Min/max values, empty strings, very long strings
- Null Handling: Missing fields, null values
- Type Validation: Send string where number expected
- SQL Injection Attempts:
'; DROP TABLE users; -- - XSS Attempts:
<script>alert('XSS')</script> - Large Payloads: Test payload size limits
- Concurrent Requests: Race conditions, deadlocks
3.3 Postman Collection
Create Comprehensive Test Suite:
{
"info": {
"name": "Product API Test Suite",
"description": "Complete test coverage for Product API"
},
"item": [
{
"name": "Products",
"item": [
{
"name": "Get All Products",
"request": {
"method": "GET",
"url": "{{base_url}}/api/v1/products"
},
"tests": [
"pm.test('Status is 200', () => {",
" pm.response.to.have.status(200);",
"});",
"pm.test('Response has data array', () => {",
" pm.expect(pm.response.json()).to.have.property('data');",
"});"
]
},
{
"name": "Create Product - Success",
"request": {
"method": "POST",
"url": "{{base_url}}/api/v1/products",
"body": {
"mode": "raw",
"raw": "{\n \"name\": \"New Product\",\n \"price\": 49.99\n}"
}
},
"tests": [
"pm.test('Status is 201', () => {",
" pm.response.to.have.status(201);",
"});",
"pm.test('Response has ID', () => {",
" pm.expect(pm.response.json().data).to.have.property('id');",
"});"
]
},
{
"name": "Create Product - Validation Error",
"request": {
"method": "POST",
"url": "{{base_url}}/api/v1/products",
"body": {
"mode": "raw",
"raw": "{\n \"name\": \"X\",\n \"price\": -5\n}"
}
},
"tests": [
"pm.test('Status is 400', () => {",
" pm.response.to.have.status(400);",
"});"
]
}
]
}
]
}JSONRite 4: Documentation & Deployment
4.1 Interactive Documentation
Host Auto-Generated Docs:
Swagger UI Integration (FastAPI):
from fastapi import FastAPI
app = FastAPI(
title="Product API",
description="Comprehensive product management API",
version="1.0.0",
docs_url="/docs", # Swagger UI
redoc_url="/redoc" # ReDoc
)PythonAccess: Navigate to https://api.example.com/docs
4.2 Professional README
Must Include:
# Product API
## Overview
RESTful API for managing product catalog with full CRUD operations.
## Features
- JWT authentication
- Pagination & filtering
- Input validation
- Comprehensive error handling
- Auto-generated OpenAPI docs
## Tech Stack
- Python 3.11
- FastAPI
- PostgreSQL
- Redis (caching)
- Docker
## Quick Start
### Prerequisites
- Docker & Docker Compose
- Python 3.11+
### Installation
```bash
# Clone repository
git clone https://github.com/yourusername/product-api.git
cd product-api
# Set environment variables
cp .env.example .env
# Start services
docker-compose up -d
# Run migrations
docker-compose exec api alembic upgrade headMarkdownAuthentication
Obtain JWT token:
POST /api/v1/auth/login
{
"username": "demo@example.com",
"password": "demo123"
}BashUse token in requests:
Authorization: Bearer <your_token_here>BashAPI Endpoints
Products
GET /api/v1/products– List all productsPOST /api/v1/products– Create productGET /api/v1/products/{id}– Get product by IDPUT /api/v1/products/{id}– Update productDELETE /api/v1/products/{id}– Delete product
Testing
# Run tests
pytest
# With coverage
pytest --cov=app --cov-report=htmlBashDocumentation
- Swagger UI: https://api.example.com/docs
- ReDoc: https://api.example.com/redoc
- Postman Collection: Download
License
MIT
#### 4.3 Live Deployment
**Cloud Options:**
1. **Heroku (Easiest):**
```bash
heroku create my-product-api
git push heroku main
heroku run alembic upgrade headMarkdown- AWS (Scalable):
- EC2 with Nginx reverse proxy
- RDS for PostgreSQL
- ElastiCache for Redis
- Route 53 for DNS
- Certificate Manager for SSL
- DigitalOcean App Platform (Balanced):
- Push to GitHub
- Connect repository
- Auto-deploy on push
Deployment Checklist:
- ✅ HTTPS enabled (Let’s Encrypt)
- ✅ Environment variables secured
- ✅ Database migrations automated
- ✅ Logging configured (CloudWatch, LogDNA)
- ✅ Monitoring setup (New Relic, Datadog)
- ✅ Rate limiting active
- ✅ CORS properly configured
- ✅ Health check endpoint (
/health)
Rite 5: Monitoring & Observability
5.1 Logging
Structured Logging Example:
import logging
import json
from datetime import datetime
class JSONFormatter(logging.Formatter):
def format(self, record):
log_data = {
"timestamp": datetime.utcnow().isoformat(),
"level": record.levelname,
"message": record.getMessage(),
"module": record.module,
"function": record.funcName
}
return json.dumps(log_data)
# Configure logger
logger = logging.getLogger(__name__)
handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())
logger.addHandler(handler)
logger.setLevel(logging.INFO)
# Use in endpoints
@app.route('/api/v1/products', methods=['POST'])
def create_product():
logger.info("Product creation requested", extra={
"user_id": get_current_user_id(),
"ip_address": request.remote_addr
})
# ... rest of logicPython5.2 Metrics & Health Checks
from datetime import datetime
import psutil
@app.route('/health', methods=['GET'])
def health_check():
"""Health check endpoint for load balancers"""
try:
# Check database connection
db.session.execute('SELECT 1')
db_status = "healthy"
except Exception as e:
db_status = "unhealthy"
logger.error(f"Database health check failed: {e}")
return jsonify({
"status": "healthy" if db_status == "healthy" else "degraded",
"timestamp": datetime.utcnow().isoformat(),
"version": "1.0.0",
"services": {
"database": db_status,
"cache": check_redis_connection()
},
"system": {
"cpu_percent": psutil.cpu_percent(),
"memory_percent": psutil.virtual_memory().percent
}
}), 200 if db_status == "healthy" else 503Python4. API Endpoint Building Templates for Reference
Template A: URL Structure & Resource Naming
Base URL Pattern
https://{domain}/api/{version}/{resource}/{identifier}BashExamples:
https://api.example.com/api/v1/productshttps://api.example.com/api/v1/products/123https://api.example.com/api/v1/users/456/orders
Naming Rules
| Rule | ✅ Correct | ❌ Incorrect |
|---|---|---|
| Use plural nouns | /products | /product |
| Lowercase only | /products | /Products |
| Hyphenate multi-word | /product-categories | /productCategories |
| No verbs | /products | /getProducts |
| No file extensions | /products | /products.json |
Nesting Guidelines
Maximum 2 levels deep:
- ✅
/users/123/orders - ✅
/orders/456/items - ❌
/users/123/orders/456/items/789/reviews(too deep)
Alternative for deep relationships:
GET /order-items?order_id=456&item_id=789BashTemplate B: HTTP Methods Standard
| Method | Action | Request Body | Response Body | Idempotent | Safe |
|---|---|---|---|---|---|
GET | Retrieve | No | Yes | Yes | Yes |
POST | Create | Yes | Yes | No | No |
PUT | Replace | Yes | Yes | Yes | No |
PATCH | Partial Update | Yes | Yes | No | No |
DELETE | Remove | No | Optional | Yes | No |
OPTIONS | Get allowed methods | No | Yes | Yes | Yes |
Idempotent: Multiple identical requests have same effect as single request
Safe: Request doesn’t modify server state
Template C: Request/Response Structure
Standard Success Response
{
"success": true,
"data": {
// Actual resource data
},
"meta": {
"timestamp": "2026-01-31T10:30:00Z",
"version": "1.0.0"
}
}JSONStandard Error Response
{
"success": false,
"error": {
"code": 400,
"message": "Human-readable error message",
"details": [
{
"field": "email",
"issue": "Invalid format",
"value": "notanemail"
}
]
},
"meta": {
"timestamp": "2026-01-31T10:30:00Z",
"requestId": "req_abc123xyz"
}
}JSONPagination Response
{
"success": true,
"data": [...],
"pagination": {
"page": 2,
"limit": 20,
"total": 150,
"totalPages": 8,
"links": {
"first": "/api/v1/products?page=1&limit=20",
"prev": "/api/v1/products?page=1&limit=20",
"self": "/api/v1/products?page=2&limit=20",
"next": "/api/v1/products?page=3&limit=20",
"last": "/api/v1/products?page=8&limit=20"
}
}
}JSONTemplate D: OpenAPI Specification Starter
openapi: 3.0.3
info:
title: Your API Name
description: Comprehensive API description
version: 1.0.0
contact:
name: API Support
email: support@example.com
license:
name: MIT
servers:
- url: https://api.example.com/api/v1
description: Production server
- url: https://staging-api.example.com/api/v1
description: Staging server
- url: http://localhost:8000/api/v1
description: Development server
tags:
- name: Products
description: Product management
- name: Users
description: User operations
paths:
/products:
get:
tags:
- Products
summary: List all products
description: Retrieve paginated list of products with optional filtering
operationId: listProducts
parameters:
- name: page
in: query
description: Page number
schema:
type: integer
minimum: 1
default: 1
- name: limit
in: query
description: Items per page
schema:
type: integer
minimum: 1
maximum: 100
default: 20
- name: category
in: query
description: Filter by category
schema:
type: string
- name: sort
in: query
description: Sort field
schema:
type: string
enum: [name, price, created_at]
default: created_at
- name: order
in: query
description: Sort order
schema:
type: string
enum: [asc, desc]
default: desc
responses:
'200':
description: Successful response
content:
application/json:
schema:
type: object
properties:
success:
type: boolean
example: true
data:
type: array
items:
$ref: '#/components/schemas/Product'
pagination:
$ref: '#/components/schemas/Pagination'
'400':
$ref: '#/components/responses/BadRequest'
'500':
$ref: '#/components/responses/InternalError'
post:
tags:
- Products
summary: Create new product
operationId: createProduct
security:
- bearerAuth: []
requestBody:
required: true
content:
application/json:
schema:
$ref: '#/components/schemas/ProductCreate'
responses:
'201':
description: Product created successfully
headers:
Location:
description: URL of created product
schema:
type: string
content:
application/json:
schema:
type: object
properties:
success:
type: boolean
data:
$ref: '#/components/schemas/Product'
'400':
$ref: '#/components/responses/BadRequest'
'401':
$ref: '#/components/responses/Unauthorized'
/products/{id}:
parameters:
- name: id
in: path
required: true
description: Product ID
schema:
type: integer
get:
tags:
- Products
summary: Get product by ID
operationId: getProduct
responses:
'200':
description: Successful response
content:
application/json:
schema:
type: object
properties:
success:
type: boolean
data:
$ref: '#/components/schemas/Product'
'404':
$ref: '#/components/responses/NotFound'
put:
tags:
- Products
summary: Update product
operationId: updateProduct
security:
- bearerAuth: []
requestBody:
required: true
content:
application/json:
schema:
$ref: '#/components/schemas/ProductUpdate'
responses:
'200':
description: Product updated
content:
application/json:
schema:
type: object
properties:
success:
type: boolean
data:
$ref: '#/components/schemas/Product'
'404':
$ref: '#/components/responses/NotFound'
delete:
tags:
- Products
summary: Delete product
operationId: deleteProduct
security:
- bearerAuth: []
responses:
'204':
description: Product deleted successfully
'404':
$ref: '#/components/responses/NotFound'
components:
securitySchemes:
bearerAuth:
type: http
scheme: bearer
bearerFormat: JWT
schemas:
Product:
type: object
required:
- id
- name
- price
properties:
id:
type: integer
readOnly: true
example: 1
name:
type: string
minLength: 2
maxLength: 255
example: "Laptop"
description:
type: string
maxLength: 1000
example: "High-performance laptop"
price:
type: number
format: float
minimum: 0
example: 999.99
category:
type: string
example: "Electronics"
inStock:
type: boolean
example: true
createdAt:
type: string
format: date-time
readOnly: true
updatedAt:
type: string
format: date-time
readOnly: true
ProductCreate:
type: object
required:
- name
- price
properties:
name:
type: string
minLength: 2
maxLength: 255
description:
type: string
price:
type: number
format: float
minimum: 0
category:
type: string
inStock:
type: boolean
default: true
ProductUpdate:
type: object
properties:
name:
type: string
description:
type: string
price:
type: number
format: float
minimum: 0
category:
type: string
inStock:
type: boolean
Pagination:
type: object
properties:
page:
type: integer
limit:
type: integer
total:
type: integer
totalPages:
type: integer
links:
type: object
properties:
first:
type: string
prev:
type: string
self:
type: string
next:
type: string
last:
type: string
Error:
type: object
properties:
success:
type: boolean
example: false
error:
type: object
properties:
code:
type: integer
message:
type: string
details:
type: array
items:
type: object
properties:
field:
type: string
issue:
type: string
responses:
BadRequest:
description: Invalid request
content:
application/json:
schema:
$ref: '#/components/schemas/Error'
Unauthorized:
description: Authentication required
content:
application/json:
schema:
$ref: '#/components/schemas/Error'
NotFound:
description: Resource not found
content:
application/json:
schema:
$ref: '#/components/schemas/Error'
InternalError:
description: Internal server error
content:
application/json:
schema:
$ref: '#/components/schemas/Error'JSONTemplate E: Django REST Framework (Full Stack)
Project Structure
django_project/
├── config/
│ ├── settings/
│ │ ├── base.py
│ │ ├── development.py
│ │ └── production.py
│ ├── urls.py
│ └── wsgi.py
├── apps/
│ └── products/
│ ├── models.py
│ ├── serializers.py
│ ├── views.py
│ ├── urls.py
│ ├── permissions.py
│ └── tests/
│ ├── test_models.py
│ ├── test_views.py
│ └── test_serializers.py
├── requirements/
│ ├── base.txt
│ ├── development.txt
│ └── production.txt
├── .env.example
├── docker-compose.yml
└── manage.pyBashModels (apps/products/models.py)
from django.db import models
from django.core.validators import MinValueValidator
from django.utils import timezone
class Product(models.Model):
"""Product model for catalog management"""
name = models.CharField(max_length=255, db_index=True)
description = models.TextField(blank=True)
price = models.DecimalField(
max_digits=10,
decimal_places=2,
validators=[MinValueValidator(0)]
)
category = models.CharField(max_length=100, db_index=True)
in_stock = models.BooleanField(default=True)
created_at = models.DateTimeField(default=timezone.now, editable=False)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
ordering = ['-created_at']
indexes = [
models.Index(fields=['category', '-created_at']),
]
def __str__(self):
return self.namePythonSerializers (apps/products/serializers.py)
from rest_framework import serializers
from .models import Product
class ProductSerializer(serializers.ModelSerializer):
"""Serializer for Product model"""
class Meta:
model = Product
fields = [
'id', 'name', 'description', 'price',
'category', 'in_stock', 'created_at', 'updated_at'
]
read_only_fields = ['id', 'created_at', 'updated_at']
def validate_price(self, value):
"""Ensure price is positive"""
if value < 0:
raise serializers.ValidationError("Price must be positive")
return value
def validate_name(self, value):
"""Ensure name is not too short"""
if len(value) < 2:
raise serializers.ValidationError("Name must be at least 2 characters")
return value
class ProductCreateSerializer(serializers.ModelSerializer):
"""Serializer for creating products"""
class Meta:
model = Product
fields = ['name', 'description', 'price', 'category', 'in_stock']PythonViews (apps/products/views.py)
from rest_framework import viewsets, filters, status
from rest_framework.decorators import action
from rest_framework.response import Response
from rest_framework.permissions import IsAuthenticatedOrReadOnly
from django_filters.rest_framework import DjangoFilterBackend
from .models import Product
from .serializers import ProductSerializer, ProductCreateSerializer
from .permissions import IsOwnerOrReadOnly
class ProductViewSet(viewsets.ModelViewSet):
"""
ViewSet for Product CRUD operations
list: GET /api/v1/products/
create: POST /api/v1/products/
retrieve: GET /api/v1/products/{id}/
update: PUT /api/v1/products/{id}/
partial_update: PATCH /api/v1/products/{id}/
destroy: DELETE /api/v1/products/{id}/
"""
queryset = Product.objects.all()
serializer_class = ProductSerializer
permission_classes = [IsAuthenticatedOrReadOnly]
filter_backends = [DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter]
filterset_fields = ['category', 'in_stock']
search_fields = ['name', 'description']
ordering_fields = ['price', 'created_at', 'name']
ordering = ['-created_at']
def get_serializer_class(self):
"""Use different serializer for create"""
if self.action == 'create':
return ProductCreateSerializer
return ProductSerializer
def list(self, request, *args, **kwargs):
"""Override to add custom response structure"""
queryset = self.filter_queryset(self.get_queryset())
page = self.paginate_queryset(queryset)
if page is not None:
serializer = self.get_serializer(page, many=True)
return self.get_paginated_response({
'success': True,
'data': serializer.data
})
serializer = self.get_serializer(queryset, many=True)
return Response({
'success': True,
'data': serializer.data
})
@action(detail=False, methods=['get'])
def categories(self, request):
"""Custom endpoint: GET /api/v1/products/categories/"""
categories = Product.objects.values_list('category', flat=True).distinct()
return Response({
'success': True,
'data': list(categories)
})PythonURLs (config/urls.py)
from django.contrib import admin
from django.urls import path, include
from rest_framework.routers import DefaultRouter
from rest_framework_simplejwt.views import (
TokenObtainPairView,
TokenRefreshView,
)
from drf_yasg.views import get_schema_view
from drf_yasg import openapi
from apps.products.views import ProductViewSet
# Router for ViewSets
router = DefaultRouter()
router.register(r'products', ProductViewSet, basename='product')
# Swagger/OpenAPI documentation
schema_view = get_schema_view(
openapi.Info(
title="Product API",
default_version='v1',
description="Comprehensive product management API",
contact=openapi.Contact(email="api@example.com"),
license=openapi.License(name="MIT License"),
),
public=True,
)
urlpatterns = [
path('admin/', admin.site.urls),
# API endpoints
path('api/v1/', include(router.urls)),
# Authentication
path('api/v1/auth/token/', TokenObtainPairView.as_view(), name='token_obtain'),
path('api/v1/auth/token/refresh/', TokenRefreshView.as_view(), name='token_refresh'),
# Documentation
path('swagger/', schema_view.with_ui('swagger', cache_timeout=0), name='swagger-ui'),
path('redoc/', schema_view.with_ui('redoc', cache_timeout=0), name='redoc'),
path('swagger.json', schema_view.without_ui(cache_timeout=0), name='schema-json'),
]PythonTemplate F: FastAPI (Modern Python)
Project Structure
fastapi_project/
├── app/
│ ├── api/
│ │ └── v1/
│ │ ├── endpoints/
│ │ │ ├── products.py
│ │ │ └── auth.py
│ │ └── router.py
│ ├── core/
│ │ ├── config.py
│ │ ├── security.py
│ │ └── database.py
│ ├── models/
│ │ └── product.py
│ ├── schemas/
│ │ └── product.py
│ ├── crud/
│ │ └── product.py
│ └── main.py
├── tests/
│ ├── test_products.py
│ └── conftest.py
├── alembic/
│ └── versions/
├── .env.example
├── requirements.txt
└── docker-compose.ymlBashMain Application (app/main.py)
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.api.v1.router import api_router
from app.core.config import settings
app = FastAPI(
title=settings.PROJECT_NAME,
version=settings.VERSION,
description="Comprehensive Product Management API",
docs_url="/docs",
redoc_url="/redoc",
openapi_url="/openapi.json"
)
# CORS
app.add_middleware(
CORSMiddleware,
allow_origins=settings.ALLOWED_ORIGINS,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Include API router
app.include_router(api_router, prefix="/api/v1")
@app.get("/", tags=["Root"])
def read_root():
"""Root endpoint"""
return {
"message": "Product API",
"version": settings.VERSION,
"docs": "/docs"
}
@app.get("/health", tags=["Health"])
def health_check():
"""Health check endpoint"""
return {
"status": "healthy",
"version": settings.VERSION
}PythonSchemas (app/schemas/product.py)
from pydantic import BaseModel, Field, validator
from datetime import datetime
from typing import Optional
class ProductBase(BaseModel):
"""Base product schema"""
name: str = Field(..., min_length=2, max_length=255)
description: Optional[str] = Field(None, max_length=1000)
price: float = Field(..., gt=0, description="Price must be positive")
category: str = Field(..., max_length=100)
in_stock: bool = Field(default=True)
@validator('name')
def name_must_not_be_empty(cls, v):
if not v.strip():
raise ValueError('Name cannot be empty')
return v.strip()
class ProductCreate(ProductBase):
"""Schema for creating products"""
pass
class ProductUpdate(BaseModel):
"""Schema for updating products (all fields optional)"""
name: Optional[str] = Field(None, min_length=2, max_length=255)
description: Optional[str] = Field(None, max_length=1000)
price: Optional[float] = Field(None, gt=0)
category: Optional[str] = Field(None, max_length=100)
in_stock: Optional[bool] = None
class ProductResponse(ProductBase):
"""Schema for product responses"""
id: int
created_at: datetime
updated_at: datetime
class Config:
orm_mode = True
class ProductListResponse(BaseModel):
"""Schema for paginated product list"""
success: bool = True
data: list[ProductResponse]
pagination: dictPythonEndpoints (app/api/v1/endpoints/products.py)
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy.orm import Session
from typing import List, Optional
from app.core.database import get_db
from app.schemas.product import (
ProductCreate,
ProductUpdate,
ProductResponse,
ProductListResponse
)
from app.crud import product as crud_product
from app.core.security import get_current_user
router = APIRouter()
@router.get("/", response_model=ProductListResponse)
def list_products(
db: Session = Depends(get_db),
page: int = Query(1, ge=1),
limit: int = Query(20, ge=1, le=100),
category: Optional[str] = None,
search: Optional[str] = None,
sort_by: str = Query("created_at", regex="^(name|price|created_at)$"),
order: str = Query("desc", regex="^(asc|desc)$")
):
"""
Retrieve products with pagination and filtering
- **page**: Page number (default: 1)
- **limit**: Items per page (default: 20, max: 100)
- **category**: Filter by category
- **search**: Search in name and description
- **sort_by**: Sort field (name, price, created_at)
- **order**: Sort order (asc, desc)
"""
skip = (page - 1) * limit
products, total = crud_product.get_products(
db=db,
skip=skip,
limit=limit,
category=category,
search=search,
sort_by=sort_by,
order=order
)
total_pages = (total + limit - 1) // limit
return {
"success": True,
"data": products,
"pagination": {
"page": page,
"limit": limit,
"total": total,
"totalPages": total_pages,
"hasNext": page < total_pages,
"hasPrev": page > 1
}
}
@router.post("/", response_model=ProductResponse, status_code=status.HTTP_201_CREATED)
def create_product(
product: ProductCreate,
db: Session = Depends(get_db),
current_user = Depends(get_current_user)
):
"""
Create new product
Requires authentication.
"""
return crud_product.create_product(db=db, product=product)
@router.get("/{product_id}", response_model=ProductResponse)
def get_product(
product_id: int,
db: Session = Depends(get_db)
):
"""
Get product by ID
"""
product = crud_product.get_product(db=db, product_id=product_id)
if not product:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Product not found"
)
return product
@router.put("/{product_id}", response_model=ProductResponse)
def update_product(
product_id: int,
product_update: ProductUpdate,
db: Session = Depends(get_db),
current_user = Depends(get_current_user)
):
"""
Update product
Requires authentication.
"""
product = crud_product.update_product(
db=db,
product_id=product_id,
product_update=product_update
)
if not product:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Product not found"
)
return product
@router.delete("/{product_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_product(
product_id: int,
db: Session = Depends(get_db),
current_user = Depends(get_current_user)
):
"""
Delete product
Requires authentication.
"""
success = crud_product.delete_product(db=db, product_id=product_id)
if not success:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Product not found"
)
return NonePythonSummary & Quick Reference
Development Workflow Checklist
- Phase 1: Design
- Define use cases and requirements
- Create OpenAPI specification
- Design database schema
- Generate mock server for validation
- Phase 2: Implementation
- Set up project structure
- Implement models/schemas
- Create CRUD endpoints
- Add input validation
- Implement authentication
- Phase 3: Testing
- Write unit tests (models, serializers)
- Write integration tests (endpoints)
- Create Postman collection
- Test edge cases and errors
- Achieve >80% code coverage
- Phase 4: Documentation
- Generate interactive API docs
- Write comprehensive README
- Document authentication flow
- Create usage examples
- Phase 5: Deployment
- Configure environment variables
- Set up database migrations
- Enable HTTPS
- Configure CORS
- Set up monitoring/logging
- Deploy to cloud platform
Key Metrics for Success
| Metric | Target | Description |
|---|---|---|
| Response Time | < 200ms | Average API response time |
| Availability | > 99.9% | Uptime percentage |
| Error Rate | < 0.1% | Percentage of 5xx errors |
| Test Coverage | > 80% | Code covered by tests |
| Documentation Coverage | 100% | All endpoints documented |
| Security Score | A+ | SSL Labs rating |
Resources & Tools
Development:
- Frameworks: Django REST Framework, FastAPI, Express.js, Spring Boot
- Databases: PostgreSQL, MySQL, MongoDB
- ORM: SQLAlchemy, Django ORM, Prisma
- Validation: Pydantic, Marshmallow, Joi
Testing:
- Unit Tests: pytest, Jest, JUnit
- API Testing: Postman, Insomnia, curl
- Load Testing: Apache JMeter, Locust, k6
- Mocking: Postman Mock, Prism
Documentation:
- Specification: OpenAPI 3.0, Swagger
- Generators: Swagger UI, ReDoc, Stoplight
- API Design: Stoplight Studio, Apicurio
Deployment:
- Platforms: AWS, Heroku, DigitalOcean, Railway
- Containerization: Docker, Kubernetes
- CI/CD: GitHub Actions, GitLab CI, Jenkins
- Monitoring: Datadog, New Relic, Sentry
Complete Project Templates
Django REST Framework – Complete Production Template
Complete File Structure
django_api_template/
├── .env.example
├── .gitignore
├── docker-compose.yml
├── Dockerfile
├── requirements.txt
├── README.md
├── manage.py
├── pytest.ini
├── config/
│ ├── __init__.py
│ ├── asgi.py
│ ├── wsgi.py
│ ├── settings/
│ │ ├── __init__.py
│ │ ├── base.py
│ │ ├── development.py
│ │ ├── production.py
│ │ └── testing.py
│ └── urls.py
└── apps/
├── __init__.py
├── products/
│ ├── __init__.py
│ ├── admin.py
│ ├── apps.py
│ ├── models.py
│ ├── serializers.py
│ ├── views.py
│ ├── urls.py
│ ├── permissions.py
│ ├── filters.py
│ ├── pagination.py
│ ├── migrations/
│ │ └── __init__.py
│ └── tests/
│ ├── __init__.py
│ ├── test_models.py
│ ├── test_views.py
│ └── test_serializers.py
└── users/
├── __init__.py
├── admin.py
├── apps.py
├── models.py
├── serializers.py
├── views.py
├── urls.py
└── tests/
└── __init__.pyBashFile: requirements.txt
# Django Core
Django==5.0.1
djangorestframework==3.14.0
django-environ==0.11.2
# Database
psycopg2-binary==2.9.9
dj-database-url==2.1.0
# Authentication & Security
djangorestframework-simplejwt==5.3.1
django-cors-headers==4.3.1
argon2-cffi==23.1.0
# Filtering & Pagination
django-filter==23.5
# API Documentation
drf-yasg==1.21.7
# Development & Testing
pytest==7.4.3
pytest-django==4.7.0
pytest-cov==4.1.0
factory-boy==3.3.0
Faker==22.0.0
# Production
gunicorn==21.2.0
whitenoise==6.6.0
# Utilities
python-dotenv==1.0.0
celery==5.3.4
redis==5.0.1BashFile: .env.example
# Django Settings
DJANGO_SETTINGS_MODULE=config.settings.development
SECRET_KEY=your-secret-key-here-change-in-production
DEBUG=True
ALLOWED_HOSTS=localhost,127.0.0.1
# Database
DATABASE_URL=postgresql://postgres:postgres@localhost:5432/django_api_db
# Redis
REDIS_URL=redis://localhost:6379/0
# JWT Settings
JWT_SECRET_KEY=your-jwt-secret-key
JWT_ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=30
REFRESH_TOKEN_EXPIRE_DAYS=7
# CORS
CORS_ALLOWED_ORIGINS=http://localhost:3000,http://localhost:8080
# API Settings
API_VERSION=v1
API_TITLE=Product APIBashFile: docker-compose.yml
version: '3.8'
services:
db:
image: postgres:16-alpine
environment:
POSTGRES_DB: django_api_db
POSTGRES_USER: postgres
POSTGRES_PASSWORD: postgres
ports:
- "5432:5432"
volumes:
- postgres_data:/var/lib/postgresql/data
redis:
image: redis:7-alpine
ports:
- "6379:6379"
volumes:
- redis_data:/data
web:
build: .
command: python manage.py runserver 0.0.0.0:8000
volumes:
- .:/app
ports:
- "8000:8000"
environment:
- DJANGO_SETTINGS_MODULE=config.settings.development
depends_on:
- db
- redis
env_file:
- .env
volumes:
postgres_data:
redis_data:YAMLFile: Dockerfile
FROM python:3.11-slim
ENV PYTHONUNBUFFERED=1
ENV PYTHONDONTWRITEBYTECODE=1
WORKDIR /app
# Install system dependencies
RUN apt-get update && apt-get install -y \
postgresql-client \
build-essential \
&& rm -rf /var/lib/apt/lists/*
# Install Python dependencies
COPY requirements.txt /app/
RUN pip install --upgrade pip && pip install -r requirements.txt
# Copy project
COPY . /app/
# Collect static files
RUN python manage.py collectstatic --noinput || true
EXPOSE 8000
CMD ["gunicorn", "config.wsgi:application", "--bind", "0.0.0.0:8000"]DockerfileFile: config/settings/base.py
import os
from pathlib import Path
import environ
# Build paths
BASE_DIR = Path(__file__).resolve().parent.parent.parent
# Environment variables
env = environ.Env(
DEBUG=(bool, False)
)
environ.Env.read_env(os.path.join(BASE_DIR, '.env'))
# Security
SECRET_KEY = env('SECRET_KEY')
DEBUG = env('DEBUG')
ALLOWED_HOSTS = env.list('ALLOWED_HOSTS', default=[])
# Application definition
INSTALLED_APPS = [
'django.contrib.admin',
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.messages',
'django.contrib.staticfiles',
# Third party apps
'rest_framework',
'rest_framework_simplejwt',
'corsheaders',
'django_filters',
'drf_yasg',
# Local apps
'apps.products',
'apps.users',
]
MIDDLEWARE = [
'django.middleware.security.SecurityMiddleware',
'whitenoise.middleware.WhiteNoiseMiddleware',
'corsheaders.middleware.CorsMiddleware',
'django.contrib.sessions.middleware.SessionMiddleware',
'django.middleware.common.CommonMiddleware',
'django.middleware.csrf.CsrfViewMiddleware',
'django.contrib.auth.middleware.AuthenticationMiddleware',
'django.contrib.messages.middleware.MessageMiddleware',
'django.middleware.clickjacking.XFrameOptionsMiddleware',
]
ROOT_URLCONF = 'config.urls'
TEMPLATES = [
{
'BACKEND': 'django.template.backends.django.DjangoTemplates',
'DIRS': [],
'APP_DIRS': True,
'OPTIONS': {
'context_processors': [
'django.template.context_processors.debug',
'django.template.context_processors.request',
'django.contrib.auth.context_processors.auth',
'django.contrib.messages.context_processors.messages',
],
},
},
]
WSGI_APPLICATION = 'config.wsgi.application'
# Database
DATABASES = {
'default': env.db('DATABASE_URL')
}
# Password validation
AUTH_PASSWORD_VALIDATORS = [
{'NAME': 'django.contrib.auth.password_validation.UserAttributeSimilarityValidator'},
{'NAME': 'django.contrib.auth.password_validation.MinimumLengthValidator'},
{'NAME': 'django.contrib.auth.password_validation.CommonPasswordValidator'},
{'NAME': 'django.contrib.auth.password_validation.NumericPasswordValidator'},
]
# Internationalization
LANGUAGE_CODE = 'en-us'
TIME_ZONE = 'UTC'
USE_I18N = True
USE_TZ = True
# Static files
STATIC_URL = '/static/'
STATIC_ROOT = os.path.join(BASE_DIR, 'staticfiles')
STATICFILES_STORAGE = 'whitenoise.storage.CompressedManifestStaticFilesStorage'
# Media files
MEDIA_URL = '/media/'
MEDIA_ROOT = os.path.join(BASE_DIR, 'media')
# Default primary key field type
DEFAULT_AUTO_FIELD = 'django.db.models.BigAutoField'
# REST Framework
REST_FRAMEWORK = {
'DEFAULT_AUTHENTICATION_CLASSES': [
'rest_framework_simplejwt.authentication.JWTAuthentication',
],
'DEFAULT_PERMISSION_CLASSES': [
'rest_framework.permissions.IsAuthenticatedOrReadOnly',
],
'DEFAULT_PAGINATION_CLASS': 'rest_framework.pagination.PageNumberPagination',
'PAGE_SIZE': 20,
'DEFAULT_FILTER_BACKENDS': [
'django_filters.rest_framework.DjangoFilterBackend',
'rest_framework.filters.SearchFilter',
'rest_framework.filters.OrderingFilter',
],
'DEFAULT_RENDERER_CLASSES': [
'rest_framework.renderers.JSONRenderer',
],
'DEFAULT_PARSER_CLASSES': [
'rest_framework.parsers.JSONParser',
],
'EXCEPTION_HANDLER': 'rest_framework.views.exception_handler',
'DATETIME_FORMAT': '%Y-%m-%dT%H:%M:%SZ',
}
# JWT Settings
from datetime import timedelta
SIMPLE_JWT = {
'ACCESS_TOKEN_LIFETIME': timedelta(minutes=env.int('ACCESS_TOKEN_EXPIRE_MINUTES', default=30)),
'REFRESH_TOKEN_LIFETIME': timedelta(days=env.int('REFRESH_TOKEN_EXPIRE_DAYS', default=7)),
'ROTATE_REFRESH_TOKENS': True,
'BLACKLIST_AFTER_ROTATION': True,
'UPDATE_LAST_LOGIN': True,
'ALGORITHM': env('JWT_ALGORITHM', default='HS256'),
'SIGNING_KEY': env('JWT_SECRET_KEY', default=SECRET_KEY),
'AUTH_HEADER_TYPES': ('Bearer',),
}
# CORS Settings
CORS_ALLOWED_ORIGINS = env.list('CORS_ALLOWED_ORIGINS', default=[])
CORS_ALLOW_CREDENTIALS = TruePythonFile: config/settings/development.py
from .base import *
DEBUG = True
# Development-specific settings
INSTALLED_APPS += [
'django_extensions',
]
# Allow all hosts in development
ALLOWED_HOSTS = ['*']
# Enable browsable API in development
REST_FRAMEWORK['DEFAULT_RENDERER_CLASSES'] = [
'rest_framework.renderers.JSONRenderer',
'rest_framework.renderers.BrowsableAPIRenderer',
]
# Logging
LOGGING = {
'version': 1,
'disable_existing_loggers': False,
'handlers': {
'console': {
'class': 'logging.StreamHandler',
},
},
'root': {
'handlers': ['console'],
'level': 'INFO',
},
}PythonFile: apps/products/models.py
from django.db import models
from django.core.validators import MinValueValidator
from django.utils import timezone
class Category(models.Model):
"""Product category model"""
name = models.CharField(max_length=100, unique=True)
description = models.TextField(blank=True)
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
verbose_name_plural = 'Categories'
ordering = ['name']
def __str__(self):
return self.name
class Product(models.Model):
"""Product model for catalog management"""
name = models.CharField(max_length=255, db_index=True)
description = models.TextField(blank=True)
price = models.DecimalField(
max_digits=10,
decimal_places=2,
validators=[MinValueValidator(0)]
)
category = models.ForeignKey(
Category,
on_delete=models.SET_NULL,
null=True,
related_name='products'
)
sku = models.CharField(max_length=50, unique=True)
in_stock = models.BooleanField(default=True)
stock_quantity = models.IntegerField(
default=0,
validators=[MinValueValidator(0)]
)
created_at = models.DateTimeField(default=timezone.now, editable=False)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
ordering = ['-created_at']
indexes = [
models.Index(fields=['sku']),
models.Index(fields=['category', '-created_at']),
models.Index(fields=['in_stock', '-created_at']),
]
def __str__(self):
return self.name
def save(self, *args, **kwargs):
"""Update in_stock based on stock_quantity"""
self.in_stock = self.stock_quantity > 0
super().save(*args, **kwargs)PythonFile: apps/products/serializers.py
from rest_framework import serializers
from .models import Product, Category
class CategorySerializer(serializers.ModelSerializer):
"""Serializer for Category model"""
product_count = serializers.SerializerMethodField()
class Meta:
model = Category
fields = ['id', 'name', 'description', 'product_count', 'created_at']
read_only_fields = ['id', 'created_at']
def get_product_count(self, obj):
return obj.products.count()
class ProductListSerializer(serializers.ModelSerializer):
"""Lightweight serializer for product lists"""
category_name = serializers.CharField(source='category.name', read_only=True)
class Meta:
model = Product
fields = [
'id', 'name', 'price', 'category_name',
'sku', 'in_stock', 'created_at'
]
read_only_fields = ['id', 'created_at']
class ProductDetailSerializer(serializers.ModelSerializer):
"""Detailed serializer for single product view"""
category = CategorySerializer(read_only=True)
category_id = serializers.PrimaryKeyRelatedField(
queryset=Category.objects.all(),
source='category',
write_only=True,
required=False
)
class Meta:
model = Product
fields = [
'id', 'name', 'description', 'price',
'category', 'category_id', 'sku',
'in_stock', 'stock_quantity',
'created_at', 'updated_at'
]
read_only_fields = ['id', 'in_stock', 'created_at', 'updated_at']
def validate_price(self, value):
"""Ensure price is positive"""
if value < 0:
raise serializers.ValidationError("Price must be positive")
return value
def validate_sku(self, value):
"""Ensure SKU is unique"""
if self.instance and self.instance.sku == value:
return value
if Product.objects.filter(sku=value).exists():
raise serializers.ValidationError("Product with this SKU already exists")
return value
class ProductCreateSerializer(serializers.ModelSerializer):
"""Serializer for creating products"""
class Meta:
model = Product
fields = [
'name', 'description', 'price',
'category', 'sku', 'stock_quantity'
]
def validate(self, data):
"""Validate product data"""
if data.get('price', 0) <= 0:
raise serializers.ValidationError({"price": "Price must be positive"})
if not data.get('name', '').strip():
raise serializers.ValidationError({"name": "Name cannot be empty"})
return dataPythonFile: apps/products/views.py
from rest_framework import viewsets, filters, status
from rest_framework.decorators import action
from rest_framework.response import Response
from rest_framework.permissions import IsAuthenticatedOrReadOnly, IsAuthenticated
from django_filters.rest_framework import DjangoFilterBackend
from django.db.models import Q
from .models import Product, Category
from .serializers import (
ProductListSerializer,
ProductDetailSerializer,
ProductCreateSerializer,
CategorySerializer
)
from .filters import ProductFilter
from .pagination import CustomPageNumberPagination
class ProductViewSet(viewsets.ModelViewSet):
"""
ViewSet for Product CRUD operations
list: GET /api/v1/products/
create: POST /api/v1/products/
retrieve: GET /api/v1/products/{id}/
update: PUT /api/v1/products/{id}/
partial_update: PATCH /api/v1/products/{id}/
destroy: DELETE /api/v1/products/{id}/
"""
queryset = Product.objects.select_related('category').all()
permission_classes = [IsAuthenticatedOrReadOnly]
pagination_class = CustomPageNumberPagination
filter_backends = [DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter]
filterset_class = ProductFilter
search_fields = ['name', 'description', 'sku']
ordering_fields = ['price', 'created_at', 'name', 'stock_quantity']
ordering = ['-created_at']
def get_serializer_class(self):
"""Use different serializers for different actions"""
if self.action == 'list':
return ProductListSerializer
elif self.action == 'create':
return ProductCreateSerializer
return ProductDetailSerializer
def get_permissions(self):
"""Set permissions based on action"""
if self.action in ['create', 'update', 'partial_update', 'destroy']:
return [IsAuthenticated()]
return [IsAuthenticatedOrReadOnly()]
def create(self, request, *args, **kwargs):
"""Create a new product"""
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
self.perform_create(serializer)
# Return detailed response
product = Product.objects.get(pk=serializer.instance.pk)
response_serializer = ProductDetailSerializer(product)
return Response(
{
'success': True,
'message': 'Product created successfully',
'data': response_serializer.data
},
status=status.HTTP_201_CREATED
)
def list(self, request, *args, **kwargs):
"""List products with custom response structure"""
queryset = self.filter_queryset(self.get_queryset())
page = self.paginate_queryset(queryset)
if page is not None:
serializer = self.get_serializer(page, many=True)
return self.get_paginated_response({
'success': True,
'data': serializer.data
})
serializer = self.get_serializer(queryset, many=True)
return Response({
'success': True,
'data': serializer.data
})
def retrieve(self, request, *args, **kwargs):
"""Retrieve single product"""
instance = self.get_object()
serializer = self.get_serializer(instance)
return Response({
'success': True,
'data': serializer.data
})
@action(detail=False, methods=['get'])
def low_stock(self, request):
"""Get products with low stock (< 10 items)"""
products = self.get_queryset().filter(stock_quantity__lt=10, stock_quantity__gt=0)
serializer = ProductListSerializer(products, many=True)
return Response({
'success': True,
'data': serializer.data,
'count': products.count()
})
@action(detail=False, methods=['get'])
def out_of_stock(self, request):
"""Get out of stock products"""
products = self.get_queryset().filter(stock_quantity=0)
serializer = ProductListSerializer(products, many=True)
return Response({
'success': True,
'data': serializer.data,
'count': products.count()
})
@action(detail=True, methods=['post'])
def update_stock(self, request, pk=None):
"""Update product stock quantity"""
product = self.get_object()
quantity = request.data.get('quantity')
if quantity is None:
return Response(
{'success': False, 'error': 'Quantity is required'},
status=status.HTTP_400_BAD_REQUEST
)
try:
quantity = int(quantity)
if quantity < 0:
raise ValueError("Quantity cannot be negative")
except ValueError as e:
return Response(
{'success': False, 'error': str(e)},
status=status.HTTP_400_BAD_REQUEST
)
product.stock_quantity = quantity
product.save()
serializer = ProductDetailSerializer(product)
return Response({
'success': True,
'message': 'Stock updated successfully',
'data': serializer.data
})
class CategoryViewSet(viewsets.ModelViewSet):
"""ViewSet for Category management"""
queryset = Category.objects.all()
serializer_class = CategorySerializer
permission_classes = [IsAuthenticatedOrReadOnly]
filter_backends = [filters.SearchFilter, filters.OrderingFilter]
search_fields = ['name', 'description']
ordering = ['name']
@action(detail=True, methods=['get'])
def products(self, request, pk=None):
"""Get all products in a category"""
category = self.get_object()
products = category.products.all()
# Apply pagination
page = self.paginate_queryset(products)
if page is not None:
serializer = ProductListSerializer(page, many=True)
return self.get_paginated_response({
'success': True,
'data': serializer.data
})
serializer = ProductListSerializer(products, many=True)
return Response({
'success': True,
'data': serializer.data
})PythonFile: apps/products/filters.py
import django_filters
from .models import Product
class ProductFilter(django_filters.FilterSet):
"""Custom filters for Product model"""
name = django_filters.CharFilter(lookup_expr='icontains')
min_price = django_filters.NumberFilter(field_name='price', lookup_expr='gte')
max_price = django_filters.NumberFilter(field_name='price', lookup_expr='lte')
min_stock = django_filters.NumberFilter(field_name='stock_quantity', lookup_expr='gte')
max_stock = django_filters.NumberFilter(field_name='stock_quantity', lookup_expr='lte')
class Meta:
model = Product
fields = {
'category': ['exact'],
'in_stock': ['exact'],
'sku': ['exact', 'icontains'],
}PythonFile: apps/products/pagination.py
from rest_framework.pagination import PageNumberPagination
from rest_framework.response import Response
class CustomPageNumberPagination(PageNumberPagination):
"""Custom pagination with enhanced response"""
page_size = 20
page_size_query_param = 'limit'
max_page_size = 100
def get_paginated_response(self, data):
return Response({
'success': True,
'data': data.get('data', data) if isinstance(data, dict) else data,
'pagination': {
'page': self.page.number,
'limit': self.page.paginator.per_page,
'total': self.page.paginator.count,
'totalPages': self.page.paginator.num_pages,
'hasNext': self.page.has_next(),
'hasPrev': self.page.has_previous(),
'links': {
'next': self.get_next_link(),
'previous': self.get_previous_link(),
}
}
})PythonFile: apps/products/tests/test_views.py
import pytest
from django.urls import reverse
from rest_framework import status
from rest_framework.test import APIClient
from apps.products.models import Product, Category
from django.contrib.auth import get_user_model
User = get_user_model()
@pytest.fixture
def api_client():
return APIClient()
@pytest.fixture
def user():
return User.objects.create_user(
username='testuser',
email='test@example.com',
password='testpass123'
)
@pytest.fixture
def authenticated_client(api_client, user):
api_client.force_authenticate(user=user)
return api_client
@pytest.fixture
def category():
return Category.objects.create(
name='Electronics',
description='Electronic products'
)
@pytest.fixture
def product(category):
return Product.objects.create(
name='Test Product',
description='Test description',
price=99.99,
category=category,
sku='TEST-001',
stock_quantity=10
)
@pytest.mark.django_db
class TestProductViewSet:
"""Tests for Product ViewSet"""
def test_list_products(self, api_client, product):
"""Test listing products"""
url = reverse('product-list')
response = api_client.get(url)
assert response.status_code == status.HTTP_200_OK
assert response.data['success'] is True
assert len(response.data['data']) == 1
def test_create_product_authenticated(self, authenticated_client, category):
"""Test creating product with authentication"""
url = reverse('product-list')
data = {
'name': 'New Product',
'description': 'New description',
'price': 49.99,
'category': category.id,
'sku': 'NEW-001',
'stock_quantity': 5
}
response = authenticated_client.post(url, data)
assert response.status_code == status.HTTP_201_CREATED
assert response.data['success'] is True
assert Product.objects.count() == 1
def test_create_product_unauthenticated(self, api_client, category):
"""Test creating product without authentication fails"""
url = reverse('product-list')
data = {
'name': 'New Product',
'price': 49.99,
'category': category.id,
'sku': 'NEW-001'
}
response = api_client.post(url, data)
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_retrieve_product(self, api_client, product):
"""Test retrieving single product"""
url = reverse('product-detail', kwargs={'pk': product.pk})
response = api_client.get(url)
assert response.status_code == status.HTTP_200_OK
assert response.data['success'] is True
assert response.data['data']['name'] == 'Test Product'
def test_update_product(self, authenticated_client, product):
"""Test updating product"""
url = reverse('product-detail', kwargs={'pk': product.pk})
data = {'name': 'Updated Product'}
response = authenticated_client.patch(url, data)
assert response.status_code == status.HTTP_200_OK
product.refresh_from_db()
assert product.name == 'Updated Product'
def test_filter_by_category(self, api_client, product, category):
"""Test filtering products by category"""
url = reverse('product-list')
response = api_client.get(url, {'category': category.id})
assert response.status_code == status.HTTP_200_OK
assert len(response.data['data']) == 1
def test_search_products(self, api_client, product):
"""Test searching products"""
url = reverse('product-list')
response = api_client.get(url, {'search': 'Test'})
assert response.status_code == status.HTTP_200_OK
assert len(response.data['data']) == 1PythonFile: config/urls.py
from django.contrib import admin
from django.urls import path, include
from rest_framework.routers import DefaultRouter
from rest_framework_simplejwt.views import (
TokenObtainPairView,
TokenRefreshView,
TokenVerifyView,
)
from drf_yasg.views import get_schema_view
from drf_yasg import openapi
from rest_framework import permissions
from apps.products.views import ProductViewSet, CategoryViewSet
# API Router
router = DefaultRouter()
router.register(r'products', ProductViewSet, basename='product')
router.register(r'categories', CategoryViewSet, basename='category')
# Swagger/OpenAPI Schema
schema_view = get_schema_view(
openapi.Info(
title="Product API",
default_version='v1',
description="Comprehensive Product Management API",
terms_of_service="https://www.example.com/terms/",
contact=openapi.Contact(email="api@example.com"),
license=openapi.License(name="MIT License"),
),
public=True,
permission_classes=[permissions.AllowAny],
)
urlpatterns = [
# Admin
path('admin/', admin.site.urls),
# API v1
path('api/v1/', include(router.urls)),
# Authentication
path('api/v1/auth/token/', TokenObtainPairView.as_view(), name='token_obtain_pair'),
path('api/v1/auth/token/refresh/', TokenRefreshView.as_view(), name='token_refresh'),
path('api/v1/auth/token/verify/', TokenVerifyView.as_view(), name='token_verify'),
# API Documentation
path('swagger/', schema_view.with_ui('swagger', cache_timeout=0), name='swagger-ui'),
path('redoc/', schema_view.with_ui('redoc', cache_timeout=0), name='redoc'),
path('swagger.json', schema_view.without_ui(cache_timeout=0), name='schema-json'),
path('swagger.yaml', schema_view.without_ui(cache_timeout=0), name='schema-yaml'),
]PythonFastAPI – Complete Production Template
Complete File Structure
fastapi_api_template/
├── .env.example
├── .gitignore
├── docker-compose.yml
├── Dockerfile
├── requirements.txt
├── README.md
├── pytest.ini
├── alembic.ini
├── app/
│ ├── __init__.py
│ ├── main.py
│ ├── api/
│ │ ├── __init__.py
│ │ └── v1/
│ │ ├── __init__.py
│ │ ├── router.py
│ │ └── endpoints/
│ │ ├── __init__.py
│ │ ├── products.py
│ │ ├── categories.py
│ │ └── auth.py
│ ├── core/
│ │ ├── __init__.py
│ │ ├── config.py
│ │ ├── security.py
│ │ ├── database.py
│ │ └── dependencies.py
│ ├── models/
│ │ ├── __init__.py
│ │ ├── product.py
│ │ ├── category.py
│ │ └── user.py
│ ├── schemas/
│ │ ├── __init__.py
│ │ ├── product.py
│ │ ├── category.py
│ │ ├── user.py
│ │ └── response.py
│ ├── crud/
│ │ ├── __init__.py
│ │ ├── base.py
│ │ ├── product.py
│ │ └── category.py
│ └── tests/
│ ├── __init__.py
│ ├── conftest.py
│ ├── test_products.py
│ └── test_auth.py
└── alembic/
├── versions/
└── env.pyBashFile: requirements.txt
# FastAPI Core
fastapi==0.109.0
uvicorn[standard]==0.27.0
pydantic==2.5.3
pydantic-settings==2.1.0
# Database
sqlalchemy==2.0.25
alembic==1.13.1
psycopg2-binary==2.9.9
# Authentication & Security
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4
python-multipart==0.0.6
# Utilities
python-dotenv==1.0.0
email-validator==2.1.0
# Development & Testing
pytest==7.4.3
pytest-asyncio==0.23.3
httpx==0.26.0
faker==22.0.0
# Production
gunicorn==21.2.0
redis==5.0.1
celery==5.3.4BashFile: .env.example
# Application
APP_NAME=FastAPI Product API
VERSION=1.0.0
DEBUG=True
ENVIRONMENT=development
# Server
HOST=0.0.0.0
PORT=8000
# Database
DATABASE_URL=postgresql://postgres:postgres@localhost:5432/fastapi_db
DATABASE_ECHO=False
# Security
SECRET_KEY=your-super-secret-key-change-in-production
ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=30
REFRESH_TOKEN_EXPIRE_DAYS=7
# CORS
ALLOWED_ORIGINS=http://localhost:3000,http://localhost:8080
# Redis
REDIS_URL=redis://localhost:6379/0BashFile: docker-compose.yml
version: '3.8'
services:
db:
image: postgres:16-alpine
environment:
POSTGRES_DB: fastapi_db
POSTGRES_USER: postgres
POSTGRES_PASSWORD: postgres
ports:
- "5432:5432"
volumes:
- postgres_data:/var/lib/postgresql/data
healthcheck:
test: ["CMD-SHELL", "pg_isready -U postgres"]
interval: 10s
timeout: 5s
retries: 5
redis:
image: redis:7-alpine
ports:
- "6379:6379"
volumes:
- redis_data:/data
api:
build: .
command: uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload
volumes:
- .:/app
ports:
- "8000:8000"
environment:
- DATABASE_URL=postgresql://postgres:postgres@db:5432/fastapi_db
- REDIS_URL=redis://redis:6379/0
depends_on:
db:
condition: service_healthy
redis:
condition: service_started
env_file:
- .env
volumes:
postgres_data:
redis_data:YAMLFile: Dockerfile
FROM python:3.11-slim
ENV PYTHONUNBUFFERED=1
ENV PYTHONDONTWRITEBYTECODE=1
WORKDIR /app
# Install system dependencies
RUN apt-get update && apt-get install -y \
postgresql-client \
build-essential \
&& rm -rf /var/lib/apt/lists/*
# Install Python dependencies
COPY requirements.txt .
RUN pip install --upgrade pip && pip install -r requirements.txt
# Copy application
COPY . .
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]DockerfileFile: app/core/config.py
from pydantic_settings import BaseSettings
from typing import List
from functools import lru_cache
class Settings(BaseSettings):
"""Application settings"""
# Application
APP_NAME: str = "FastAPI Product API"
VERSION: str = "1.0.0"
DEBUG: bool = False
ENVIRONMENT: str = "production"
# Server
HOST: str = "0.0.0.0"
PORT: int = 8000
# Database
DATABASE_URL: str
DATABASE_ECHO: bool = False
# Security
SECRET_KEY: str
ALGORITHM: str = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
REFRESH_TOKEN_EXPIRE_DAYS: int = 7
# CORS
ALLOWED_ORIGINS: List[str] = []
# Redis
REDIS_URL: str = "redis://localhost:6379/0"
class Config:
env_file = ".env"
case_sensitive = True
@lru_cache()
def get_settings() -> Settings:
"""Get cached settings instance"""
return Settings()
settings = get_settings()PythonFile: app/core/database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from .config import settings
# Create engine
engine = create_engine(
settings.DATABASE_URL,
echo=settings.DATABASE_ECHO,
pool_pre_ping=True,
pool_size=10,
max_overflow=20
)
# Create session
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# Base class for models
Base = declarative_base()
def get_db():
"""Dependency for getting database session"""
db = SessionLocal()
try:
yield db
finally:
db.close()PythonFile: app/core/security.py
from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from sqlalchemy.orm import Session
from .config import settings
from .database import get_db
from ..models.user import User
# Password hashing
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
# HTTP Bearer token
security = HTTPBearer()
def verify_password(plain_password: str, hashed_password: str) -> bool:
"""Verify password against hash"""
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password: str) -> str:
"""Hash password"""
return pwd_context.hash(password)
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
"""Create JWT access token"""
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
return encoded_jwt
def decode_token(token: str) -> dict:
"""Decode JWT token"""
try:
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
return payload
except JWTError:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
async def get_current_user(
credentials: HTTPAuthorizationCredentials = Depends(security),
db: Session = Depends(get_db)
) -> User:
"""Get current authenticated user"""
token = credentials.credentials
payload = decode_token(token)
username: str = payload.get("sub")
if username is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
)
user = db.query(User).filter(User.username == username).first()
if user is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="User not found",
)
return userPythonFile: app/models/category.py
from sqlalchemy import Column, Integer, String, Text, DateTime
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from ..core.database import Base
class Category(Base):
"""Category model"""
__tablename__ = "categories"
id = Column(Integer, primary_key=True, index=True)
name = Column(String(100), unique=True, nullable=False, index=True)
description = Column(Text, nullable=True)
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), onupdate=func.now())
# Relationships
products = relationship("Product", back_populates="category")
def __repr__(self):
return f"<Category {self.name}>"PythonFile: app/models/product.py
from sqlalchemy import Column, Integer, String, Text, Numeric, Boolean, DateTime, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from ..core.database import Base
class Product(Base):
"""Product model"""
__tablename__ = "products"
id = Column(Integer, primary_key=True, index=True)
name = Column(String(255), nullable=False, index=True)
description = Column(Text, nullable=True)
price = Column(Numeric(10, 2), nullable=False)
sku = Column(String(50), unique=True, nullable=False, index=True)
in_stock = Column(Boolean, default=True)
stock_quantity = Column(Integer, default=0)
category_id = Column(Integer, ForeignKey("categories.id"), nullable=True)
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), onupdate=func.now())
# Relationships
category = relationship("Category", back_populates="products")
def __repr__(self):
return f"<Product {self.name}>"PythonFile: app/models/user.py
from sqlalchemy import Column, Integer, String, Boolean, DateTime
from sqlalchemy.sql import func
from ..core.database import Base
class User(Base):
"""User model"""
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
username = Column(String(50), unique=True, nullable=False, index=True)
email = Column(String(255), unique=True, nullable=False, index=True)
hashed_password = Column(String(255), nullable=False)
is_active = Column(Boolean, default=True)
is_superuser = Column(Boolean, default=False)
created_at = Column(DateTime(timezone=True), server_default=func.now())
def __repr__(self):
return f"<User {self.username}>"PythonFile: app/schemas/response.py
from pydantic import BaseModel
from typing import Generic, TypeVar, Optional, Any
T = TypeVar('T')
class ResponseModel(BaseModel, Generic[T]):
"""Generic response model"""
success: bool = True
message: Optional[str] = None
data: Optional[T] = None
class PaginationMeta(BaseModel):
"""Pagination metadata"""
page: int
limit: int
total: int
totalPages: int
hasNext: bool
hasPrev: bool
class PaginatedResponse(BaseModel, Generic[T]):
"""Paginated response model"""
success: bool = True
data: list[T]
pagination: PaginationMeta
class ErrorDetail(BaseModel):
"""Error detail model"""
field: Optional[str] = None
message: str
class ErrorResponse(BaseModel):
"""Error response model"""
success: bool = False
error: dict[str, Any]PythonFile: app/schemas/product.py
from pydantic import BaseModel, Field, validator
from datetime import datetime
from typing import Optional
from decimal import Decimal
class CategoryBase(BaseModel):
"""Base category schema"""
name: str = Field(..., min_length=1, max_length=100)
description: Optional[str] = None
class CategoryCreate(CategoryBase):
"""Schema for creating categories"""
pass
class CategoryResponse(CategoryBase):
"""Schema for category responses"""
id: int
created_at: datetime
product_count: int = 0
class Config:
from_attributes = True
class ProductBase(BaseModel):
"""Base product schema"""
name: str = Field(..., min_length=2, max_length=255)
description: Optional[str] = Field(None, max_length=1000)
price: Decimal = Field(..., gt=0, decimal_places=2)
sku: str = Field(..., min_length=1, max_length=50)
stock_quantity: int = Field(default=0, ge=0)
@validator('name')
def name_must_not_be_empty(cls, v):
if not v.strip():
raise ValueError('Name cannot be empty')
return v.strip()
@validator('sku')
def sku_must_be_uppercase(cls, v):
return v.upper().strip()
class ProductCreate(ProductBase):
"""Schema for creating products"""
category_id: Optional[int] = None
class ProductUpdate(BaseModel):
"""Schema for updating products"""
name: Optional[str] = Field(None, min_length=2, max_length=255)
description: Optional[str] = Field(None, max_length=1000)
price: Optional[Decimal] = Field(None, gt=0, decimal_places=2)
sku: Optional[str] = Field(None, min_length=1, max_length=50)
stock_quantity: Optional[int] = Field(None, ge=0)
category_id: Optional[int] = None
class ProductResponse(ProductBase):
"""Schema for product responses"""
id: int
in_stock: bool
category_id: Optional[int]
created_at: datetime
updated_at: Optional[datetime]
class Config:
from_attributes = True
class ProductDetailResponse(ProductResponse):
"""Schema for detailed product responses"""
category: Optional[CategoryResponse] = None
class Config:
from_attributes = TruePythonFile: app/crud/product.py
from sqlalchemy.orm import Session
from sqlalchemy import or_
from typing import Optional, Tuple, List
from ..models.product import Product
from ..schemas.product import ProductCreate, ProductUpdate
def get_product(db: Session, product_id: int) -> Optional[Product]:
"""Get product by ID"""
return db.query(Product).filter(Product.id == product_id).first()
def get_product_by_sku(db: Session, sku: str) -> Optional[Product]:
"""Get product by SKU"""
return db.query(Product).filter(Product.sku == sku).first()
def get_products(
db: Session,
skip: int = 0,
limit: int = 20,
category_id: Optional[int] = None,
search: Optional[str] = None,
in_stock: Optional[bool] = None,
sort_by: str = "created_at",
order: str = "desc"
) -> Tuple[List[Product], int]:
"""Get products with filtering and pagination"""
query = db.query(Product)
# Apply filters
if category_id:
query = query.filter(Product.category_id == category_id)
if in_stock is not None:
query = query.filter(Product.in_stock == in_stock)
if search:
query = query.filter(
or_(
Product.name.ilike(f"%{search}%"),
Product.description.ilike(f"%{search}%"),
Product.sku.ilike(f"%{search}%")
)
)
# Get total count
total = query.count()
# Apply sorting
sort_column = getattr(Product, sort_by, Product.created_at)
if order == "desc":
query = query.order_by(sort_column.desc())
else:
query = query.order_by(sort_column.asc())
# Apply pagination
products = query.offset(skip).limit(limit).all()
return products, total
def create_product(db: Session, product: ProductCreate) -> Product:
"""Create new product"""
db_product = Product(
**product.model_dump(),
in_stock=product.stock_quantity > 0
)
db.add(db_product)
db.commit()
db.refresh(db_product)
return db_product
def update_product(
db: Session,
product_id: int,
product_update: ProductUpdate
) -> Optional[Product]:
"""Update product"""
db_product = get_product(db, product_id)
if not db_product:
return None
update_data = product_update.model_dump(exclude_unset=True)
# Update in_stock based on stock_quantity if being updated
if "stock_quantity" in update_data:
update_data["in_stock"] = update_data["stock_quantity"] > 0
for field, value in update_data.items():
setattr(db_product, field, value)
db.commit()
db.refresh(db_product)
return db_product
def delete_product(db: Session, product_id: int) -> bool:
"""Delete product"""
db_product = get_product(db, product_id)
if not db_product:
return False
db.delete(db_product)
db.commit()
return TruePythonFile: app/api/v1/endpoints/products.py
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy.orm import Session
from typing import Optional
from ....core.database import get_db
from ....core.security import get_current_user
from ....models.user import User
from ....schemas.product import (
ProductCreate,
ProductUpdate,
ProductResponse,
ProductDetailResponse
)
from ....schemas.response import ResponseModel, PaginatedResponse, PaginationMeta
from ....crud import product as crud_product
router = APIRouter()
@router.get("/", response_model=PaginatedResponse[ProductResponse])
async def list_products(
db: Session = Depends(get_db),
page: int = Query(1, ge=1, description="Page number"),
limit: int = Query(20, ge=1, le=100, description="Items per page"),
category_id: Optional[int] = Query(None, description="Filter by category ID"),
search: Optional[str] = Query(None, description="Search in name, description, SKU"),
in_stock: Optional[bool] = Query(None, description="Filter by stock status"),
sort_by: str = Query("created_at", regex="^(name|price|created_at|stock_quantity)$"),
order: str = Query("desc", regex="^(asc|desc)$")
):
"""
Retrieve products with pagination and filtering.
- **page**: Page number (default: 1)
- **limit**: Items per page (default: 20, max: 100)
- **category_id**: Filter by category
- **search**: Search in name, description, and SKU
- **in_stock**: Filter by stock availability
- **sort_by**: Sort field (name, price, created_at, stock_quantity)
- **order**: Sort order (asc, desc)
"""
skip = (page - 1) * limit
products, total = crud_product.get_products(
db=db,
skip=skip,
limit=limit,
category_id=category_id,
search=search,
in_stock=in_stock,
sort_by=sort_by,
order=order
)
total_pages = (total + limit - 1) // limit
return {
"success": True,
"data": products,
"pagination": PaginationMeta(
page=page,
limit=limit,
total=total,
totalPages=total_pages,
hasNext=page < total_pages,
hasPrev=page > 1
)
}
@router.post(
"/",
response_model=ResponseModel[ProductDetailResponse],
status_code=status.HTTP_201_CREATED
)
async def create_product(
product: ProductCreate,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""
Create new product.
Requires authentication.
"""
# Check if SKU already exists
existing_product = crud_product.get_product_by_sku(db, product.sku)
if existing_product:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Product with this SKU already exists"
)
db_product = crud_product.create_product(db=db, product=product)
return {
"success": True,
"message": "Product created successfully",
"data": db_product
}
@router.get("/{product_id}", response_model=ResponseModel[ProductDetailResponse])
async def get_product(
product_id: int,
db: Session = Depends(get_db)
):
"""Get product by ID"""
product = crud_product.get_product(db=db, product_id=product_id)
if not product:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Product not found"
)
return {
"success": True,
"data": product
}
@router.put("/{product_id}", response_model=ResponseModel[ProductDetailResponse])
async def update_product(
product_id: int,
product_update: ProductUpdate,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""
Update product.
Requires authentication.
"""
# Check if SKU is being updated and already exists
if product_update.sku:
existing_product = crud_product.get_product_by_sku(db, product_update.sku)
if existing_product and existing_product.id != product_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Product with this SKU already exists"
)
product = crud_product.update_product(
db=db,
product_id=product_id,
product_update=product_update
)
if not product:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Product not found"
)
return {
"success": True,
"message": "Product updated successfully",
"data": product
}
@router.delete("/{product_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_product(
product_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""
Delete product.
Requires authentication.
"""
success = crud_product.delete_product(db=db, product_id=product_id)
if not success:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Product not found"
)
return NonePythonFile: app/api/v1/router.py
from fastapi import APIRouter
from .endpoints import products, categories, auth
api_router = APIRouter()
api_router.include_router(
products.router,
prefix="/products",
tags=["Products"]
)
api_router.include_router(
categories.router,
prefix="/categories",
tags=["Categories"]
)
api_router.include_router(
auth.router,
prefix="/auth",
tags=["Authentication"]
)PythonFile: app/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from .core.config import settings
from .core.database import engine, Base
from .api.v1.router import api_router
# Create database tables
Base.metadata.create_all(bind=engine)
# Create FastAPI app
app = FastAPI(
title=settings.APP_NAME,
version=settings.VERSION,
description="Comprehensive Product Management API with authentication and full CRUD operations",
docs_url="/docs",
redoc_url="/redoc",
openapi_url="/openapi.json"
)
# CORS
if settings.ALLOWED_ORIGINS:
app.add_middleware(
CORSMiddleware,
allow_origins=settings.ALLOWED_ORIGINS,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Include API router
app.include_router(api_router, prefix="/api/v1")
@app.get("/", tags=["Root"])
async def root():
"""Root endpoint"""
return {
"message": f"Welcome to {settings.APP_NAME}",
"version": settings.VERSION,
"docs": "/docs",
"redoc": "/redoc"
}
@app.get("/health", tags=["Health"])
async def health_check():
"""Health check endpoint"""
return {
"status": "healthy",
"version": settings.VERSION,
"environment": settings.ENVIRONMENT
}
@app.exception_handler(Exception)
async def global_exception_handler(request, exc):
"""Global exception handler"""
return JSONResponse(
status_code=500,
content={
"success": False,
"error": {
"message": "Internal server error",
"detail": str(exc) if settings.DEBUG else "An error occurred"
}
}
)PythonFile: app/tests/test_products.py
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from ..main import app
from ..core.database import Base, get_db
from ..models.product import Product
from ..models.category import Category
# Test database
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base.metadata.create_all(bind=engine)
def override_get_db():
try:
db = TestingSessionLocal()
yield db
finally:
db.close()
app.dependency_overrides[get_db] = override_get_db
client = TestClient(app)
@pytest.fixture
def test_category():
db = TestingSessionLocal()
category = Category(name="Test Category", description="Test description")
db.add(category)
db.commit()
db.refresh(category)
yield category
db.query(Category).delete()
db.commit()
db.close()
@pytest.fixture
def test_product(test_category):
db = TestingSessionLocal()
product = Product(
name="Test Product",
description="Test description",
price=99.99,
sku="TEST-001",
stock_quantity=10,
category_id=test_category.id
)
db.add(product)
db.commit()
db.refresh(product)
yield product
db.query(Product).delete()
db.commit()
db.close()
def test_read_root():
"""Test root endpoint"""
response = client.get("/")
assert response.status_code == 200
assert "message" in response.json()
def test_health_check():
"""Test health check endpoint"""
response = client.get("/health")
assert response.status_code == 200
assert response.json()["status"] == "healthy"
def test_list_products_empty():
"""Test listing products when empty"""
response = client.get("/api/v1/products/")
assert response.status_code == 200
assert response.json()["success"] is True
def test_list_products_with_data(test_product):
"""Test listing products with data"""
response = client.get("/api/v1/products/")
assert response.status_code == 200
assert len(response.json()["data"]) > 0
def test_get_product(test_product):
"""Test getting single product"""
response = client.get(f"/api/v1/products/{test_product.id}")
assert response.status_code == 200
assert response.json()["data"]["name"] == "Test Product"
def test_get_product_not_found():
"""Test getting non-existent product"""
response = client.get("/api/v1/products/99999")
assert response.status_code == 404PythonConclusion
Building professional API endpoints requires:
- Structured Design: OpenAPI-first approach with clear resource modeling
- Best Practices: RESTful principles, proper status codes, input validation
- Comprehensive Testing: Unit, integration, and edge case coverage
- Professional Documentation: Interactive docs, README, examples
- Production Readiness: Security, monitoring, scalability
This guide provides the templates and progression path to create production-quality APIs that showcase your development expertise. Start with a simple PoC, progressively add features, and always maintain professional standards throughout the development lifecycle.
Quick Start Commands
Django Template:
# Clone and setup
git clone <your-repo> && cd django_api_template
python -m venv venv && source venv/bin/activate
pip install -r requirements.txt
# Setup database
cp .env.example .env
python manage.py migrate
python manage.py createsuperuser
# Run server
python manage.py runserver
# Run tests
pytest
# Access docs
open http://localhost:8000/swagger/BashFastAPI Template:
# Clone and setup
git clone <your-repo> && cd fastapi_api_template
python -m venv venv && source venv/bin/activate
pip install -r requirements.txt
# Setup database
cp .env.example .env
alembic upgrade head
# Run server
uvicorn app.main:app --reload
# Run tests
pytest
# Access docs
open http://localhost:8000/docsBashDiscover more from Altgr Blog
Subscribe to get the latest posts sent to your email.
