Comprehensive Guide to API Endpoint Creation

    This guide provides structured answers to the fundamental questions of API endpoint development, from initial design through production deployment.


    Table of Contents

    1. Creating API Endpoints: Tasks, Patterns, Architecture, and Progression
    2. JSON Formatting with Database Tables and Progression
    3. Rites of Passage for Showcasing API Development
    4. 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:

    OperationHTTP MethodEndpoint ExampleDescription
    CreatePOST/api/v1/productsCreate new resource
    Read AllGET/api/v1/productsRetrieve collection
    Read OneGET/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
    DeleteDELETE/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
    • Filtering: Allow query parameters for searching
      • Example: GET /api/v1/products?category=electronics&price_max=500
    • Sorting: Enable ordering by fields
      • Example: GET /api/v1/products?sort=price&order=desc

    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 product
    Bash

    Status Code Usage:

    • 200 OK – Successful GET, PUT, PATCH
    • 201 Created – Successful POST
    • 204 No Content – Successful DELETE
    • 400 Bad Request – Invalid input
    • 401 Unauthorized – Missing/invalid auth
    • 403 Forbidden – Insufficient permissions
    • 404 Not Found – Resource doesn’t exist
    • 429 Too Many Requests – Rate limit exceeded
    • 500 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
        }
      }
    }
    GraphQL

    When 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
      }
    }
    JSON

    When 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), 201
    Python

    Deliverable: 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"
      }
    }
    JSON

    Error Response:

    {
      "success": false,
      "error": {
        "code": 400,
        "message": "Validation failed",
        "details": [
          {
            "field": "price",
            "issue": "must be a positive number"
          }
        ]
      }
    }
    JSON

    Phase 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
        })
    Python

    Phase 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
      }
    }
    Bash

    Rate Limiting Headers:

    X-RateLimit-Limit: 1000
    X-RateLimit-Remaining: 987
    X-RateLimit-Reset: 1643630400
    Bash

    Phase 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"
    }
    Bash

    2. 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:00
    Bash

    JSON 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"
      }
    ]
    JSON

    2.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 TypeJSON TypeNotes
    INTEGERnumberNo quotes
    DECIMAL/FLOATnumberPreserve precision
    VARCHAR/TEXTstringUse quotes
    BOOLEANbooleantrue/false, not 1/0
    DATE/TIMESTAMPstringISO 8601 format
    NULLnullDecide if to include or omit
    JSON/JSONBobjectDirect 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;
    SQL

    JSON Output:

    [
      {
        "id": 1,
        "name": "John Doe",
        "email": "john@example.com"
      },
      {
        "id": 2,
        "name": "Jane Smith",
        "email": "jane@example.com"
      }
    ]
    JSON

    Implementation (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])
    Python

    Progression 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;
    SQL

    JSON Output:

    {
      "id": 1,
      "name": "John Doe",
      "email": "john@example.com"
    }
    JSON

    Error Handling:

    // If ID doesn't exist
    {
      "success": false,
      "error": {
        "code": 404,
        "message": "User not found"
      }
    }
    JSON

    Progression 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
    }
    JSON

    Validation Rules:

    • name: Required, min 2 characters, max 100
    • email: Required, valid email format
    • age: Optional, must be positive integer

    SQL Insert:

    INSERT INTO users (name, email, age) 
    VALUES ('Alice Johnson', 'alice@example.com', 28);
    SQL

    Success Response (201 Created):

    {
      "success": true,
      "data": {
        "id": 3,
        "name": "Alice Johnson",
        "email": "alice@example.com",
        "age": 28,
        "createdAt": "2026-01-31T10:30:00Z"
      }
    }
    JSON

    Validation Error (400 Bad Request):

    {
      "success": false,
      "error": {
        "code": 400,
        "message": "Validation failed",
        "details": [
          {
            "field": "email",
            "issue": "Invalid email format"
          }
        ]
      }
    }
    JSON

    Progression 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 Ave
    Bash

    SQL 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;
    SQL

    Nested 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"
        }
      ]
    }
    JSON

    Alternative: Separate Endpoints

    GET /api/v1/users/1            User data
    GET /api/v1/users/1/addresses  User's addresses
    Bash

    Progression 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"
    }
    JSON

    SQL Update:

    UPDATE users 
    SET email = 'newemail@example.com', 
        updated_at = NOW()
    WHERE id = 1;
    SQL

    Response (200 OK):

    {
      "success": true,
      "data": {
        "id": 1,
        "name": "John Doe",
        "email": "newemail@example.com",
        "updatedAt": "2026-01-31T11:00:00Z"
      }
    }
    JSON

    Progression 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
    );
    SQL

    Insert with JSON:

    INSERT INTO products (name, metadata) 
    VALUES (
        'Laptop',
        '{"brand": "Dell", "specs": {"ram": "16GB", "storage": "512GB SSD"}}'
    );
    SQL

    Query 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';
    SQL

    Add GIN Index for Performance:

    CREATE INDEX idx_products_metadata 
    ON products USING GIN (metadata);
    SQL

    API Response:

    {
      "id": 1,
      "name": "Laptop",
      "metadata": {
        "brand": "Dell",
        "specs": {
          "ram": "16GB",
          "storage": "512GB SSD"
        }
      }
    }
    JSON

    Database-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;
    SQL

    SQL 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;
    SQL

    Best Practices for JSON Formatting

    1. Consistency: Always use the same naming convention (camelCase recommended)
    2. Omit Null vs Include Null: Decide project-wide whether to include null values
    3. Date Formatting: Use ISO 8601 format (2026-01-31T10:30:00Z)
    4. Avoid Deep Nesting: Limit to 2-3 levels; consider separate endpoints instead
    5. Use DTOs: Don’t expose raw database schema; use Data Transfer Objects
    6. 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: true
    JSON

    1.2 Mock Validation

    Tools: Postman Mock Server, Prism, Swagger Codegen

    Process:

    1. Generate mock server from OpenAPI spec
    2. Create test requests in Postman
    3. Validate responses match specification
    4. 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.yml
    Bash

    2.2 Versioning Strategy

    Why: Maintain backward compatibility

    Approaches:

    1. URL Versioning: /api/v1/products, /api/v2/products
    2. Header Versioning: Accept: application/vnd.api+json; version=1
    3. 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"
        }
      }
    }
    JSON

    2.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 data
    Python

    2.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 errors
    Python

    Rite 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 == 404
    Python

    3.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);",
                "});"
              ]
            }
          ]
        }
      ]
    }
    JSON

    Rite 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
    )
    Python

    Access: 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 head
    Markdown

    Authentication

    Obtain JWT token:

    POST /api/v1/auth/login
    {
      "username": "demo@example.com",
      "password": "demo123"
    }
    Bash

    Use token in requests:

    Authorization: Bearer <your_token_here>
    Bash

    API Endpoints

    Products

    • GET /api/v1/products – List all products
    • POST /api/v1/products – Create product
    • GET /api/v1/products/{id} – Get product by ID
    • PUT /api/v1/products/{id} – Update product
    • DELETE /api/v1/products/{id} – Delete product

    Testing

    # Run tests
    pytest
    
    # With coverage
    pytest --cov=app --cov-report=html
    Bash

    Documentation

    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 head
    Markdown
    1. AWS (Scalable):
    • EC2 with Nginx reverse proxy
    • RDS for PostgreSQL
    • ElastiCache for Redis
    • Route 53 for DNS
    • Certificate Manager for SSL
    1. 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 logic
    Python

    5.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 503
    Python

    4. API Endpoint Building Templates for Reference

    Template A: URL Structure & Resource Naming

    Base URL Pattern

    https://{domain}/api/{version}/{resource}/{identifier}
    Bash

    Examples:

    • https://api.example.com/api/v1/products
    • https://api.example.com/api/v1/products/123
    • https://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=789
    Bash

    Template B: HTTP Methods Standard

    MethodActionRequest BodyResponse BodyIdempotentSafe
    GETRetrieveNoYesYesYes
    POSTCreateYesYesNoNo
    PUTReplaceYesYesYesNo
    PATCHPartial UpdateYesYesNoNo
    DELETERemoveNoOptionalYesNo
    OPTIONSGet allowed methodsNoYesYesYes

    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"
      }
    }
    JSON

    Standard 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"
      }
    }
    JSON

    Pagination 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"
        }
      }
    }
    JSON

    Template 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'
    JSON

    Template 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.py
    Bash

    Models (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.name
    Python

    Serializers (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']
    Python

    Views (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)
            })
    Python

    URLs (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'),
    ]
    Python

    Template 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.yml
    Bash

    Main 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
        }
    Python

    Schemas (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: dict
    Python

    Endpoints (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 None
    Python

    Summary & 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

    MetricTargetDescription
    Response Time< 200msAverage API response time
    Availability> 99.9%Uptime percentage
    Error Rate< 0.1%Percentage of 5xx errors
    Test Coverage> 80%Code covered by tests
    Documentation Coverage100%All endpoints documented
    Security ScoreA+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__.py
    Bash

    File: 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.1
    Bash

    File: .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 API
    Bash

    File: 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:
    YAML

    File: 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"]
    Dockerfile

    File: 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 = True
    Python

    File: 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',
        },
    }
    Python

    File: 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)
    Python

    File: 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 data
    Python

    File: 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
            })
    Python

    File: 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'],
            }
    Python

    File: 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(),
                    }
                }
            })
    Python

    File: 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']) == 1
    Python

    File: 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'),
    ]
    Python

    FastAPI – 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.py
    Bash

    File: 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.4
    Bash

    File: .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/0
    Bash

    File: 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:
    YAML

    File: 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"]
    Dockerfile

    File: 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()
    Python

    File: 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()
    Python

    File: 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 user
    Python

    File: 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}>"
    Python

    File: 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}>"
    Python

    File: 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}>"
    Python

    File: 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]
    Python

    File: 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 = True
    Python

    File: 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 True
    Python

    File: 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 None
    Python

    File: 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"]
    )
    Python

    File: 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"
                }
            }
        )
    Python

    File: 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 == 404
    Python

    Conclusion

    Building professional API endpoints requires:

    1. Structured Design: OpenAPI-first approach with clear resource modeling
    2. Best Practices: RESTful principles, proper status codes, input validation
    3. Comprehensive Testing: Unit, integration, and edge case coverage
    4. Professional Documentation: Interactive docs, README, examples
    5. 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/
    Bash

    FastAPI 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/docs
    Bash

    Discover more from Altgr Blog

    Subscribe to get the latest posts sent to your email.

    Leave a Reply

    Your email address will not be published. Required fields are marked *