Press ESC to close

Tutorial FastAPI #6: Guide Lengkap Testing dan Deployment ke Production

Tutorial lengkap cara testing aplikasi FastAPI dan deployment ke production. Pelajari unit testing, integration testing, CI/CD pipeline, dan deployment ke berbagai platform cloud dengan best practices.

Setup Testing Environment {#setup-testing}

Install Dependencies

pip install pytest pytest-asyncio httpx pytest-cov pytest-env pytest-mock

Project Structure

myapp/
├── app/
│   └── ... (aplikasi FastAPI)
├── tests/
│   ├── __init__.py
│   ├── conftest.py
│   ├── test_auth.py
│   ├── test_api.py
│   └── test_db.py
├── Dockerfile
├── docker-compose.yml
├── .env.test
└── pytest.ini

Test Configuration (pytest.ini)

[pytest]
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*
addopts = -v --cov=app --cov-report=term-missing
env_files =
    .env.test

Unit Testing {#unit-testing}

Setup Test Database (tests/conftest.py)

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from fastapi.testclient import TestClient
from app.database import Base, get_db
from app.main import app

# Test database URL
SQLALCHEMY_TEST_DATABASE_URL = "sqlite:///./test.db"

@pytest.fixture(scope="session")
def engine():
    engine = create_engine(
        SQLALCHEMY_TEST_DATABASE_URL,
        connect_args={"check_same_thread": False}
    )
    Base.metadata.create_all(bind=engine)
    yield engine
    Base.metadata.drop_all(bind=engine)

@pytest.fixture(scope="function")
def db_session(engine):
    TestingSessionLocal = sessionmaker(
        autocommit=False,
        autoflush=False,
        bind=engine
    )
    db = TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()

@pytest.fixture(scope="function")
def client(db_session):
    def override_get_db():
        try:
            yield db_session
        finally:
            pass

    app.dependency_overrides[get_db] = override_get_db
    with TestClient(app) as client:
        yield client
    app.dependency_overrides.clear()

@pytest.fixture
def test_user():
    return {
        "username": "testuser",
        "email": "test@example.com",
        "password": "testpass123"
    }

Test Authentication (tests/test_auth.py)

import pytest
from fastapi import status

def test_create_user(client, test_user):
    response = client.post("/users/", json=test_user)
    assert response.status_code == status.HTTP_201_CREATED
    data = response.json()
    assert data["email"] == test_user["email"]
    assert "id" in data

def test_login(client, test_user):
    # Create user first
    client.post("/users/", json=test_user)

    # Test login
    login_data = {
        "username": test_user["username"],
        "password": test_user["password"]
    }
    response = client.post("/token", data=login_data)
    assert response.status_code == status.HTTP_200_OK
    assert "access_token" in response.json()

@pytest.mark.parametrize(
    "email,password,status_code",
    [
        ("wrong@example.com", "testpass123", 401),
        ("test@example.com", "wrongpass", 401),
        (None, "testpass123", 422),
        ("test@example.com", None, 422),
    ]
)
def test_login_validation(client, test_user, email, password, status_code):
    # Create user first
    client.post("/users/", json=test_user)

    # Test invalid login
    login_data = {
        "username": email,
        "password": password
    }
    response = client.post("/token", data=login_data)
    assert response.status_code == status_code

Integration Testing {#integration-testing}

Test Complete Flow (tests/test_api.py)

def test_create_and_read_item(client, test_user):
    # 1. Create user and login
    client.post("/users/", json=test_user)
    login_response = client.post(
        "/token",
        data={
            "username": test_user["username"],
            "password": test_user["password"]
        }
    )
    token = login_response.json()["access_token"]
    headers = {"Authorization": f"Bearer {token}"}

    # 2. Create item
    item_data = {
        "title": "Test Item",
        "description": "Test Description",
        "price": 99.99
    }
    create_response = client.post(
        "/items/",
        json=item_data,
        headers=headers
    )
    assert create_response.status_code == status.HTTP_201_CREATED
    item_id = create_response.json()["id"]

    # 3. Read item
    read_response = client.get(f"/items/{item_id}")
    assert read_response.status_code == status.HTTP_200_OK
    assert read_response.json()["title"] == item_data["title"]

Deployment Setup {#deployment-setup}

Dockerfile

# Use official Python image
FROM python:3.9-slim

# Set working directory
WORKDIR /app

# Set environment variables
ENV PYTHONDONTWRITEBYTECODE 1
ENV PYTHONUNBUFFERED 1

# Install system dependencies
RUN apt-get update \
    && apt-get install -y --no-install-recommends \
        postgresql-client \
    && rm -rf /var/lib/apt/lists/*

# Install Python dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy project
COPY . .

# Run migrations
RUN alembic upgrade head

# Run the application
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

Docker Compose

version: '3.8'

services:
  web:
    build: .
    ports:
      - "8000:8000"
    env_file:
      - .env
    depends_on:
      - db
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3

  db:
    image: postgres:13
    volumes:
      - postgres_data:/var/lib/postgresql/data
    env_file:
      - .env
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres"]
      interval: 5s
      timeout: 5s
      retries: 5

volumes:
  postgres_data:

CI/CD Pipeline {#cicd}

GitHub Actions Workflow

name: FastAPI CI/CD

on:
  push:
    branches: [ main ]
  pull_request:
    branches: [ main ]

jobs:
  test:
    runs-on: ubuntu-latest

    services:
      postgres:
        image: postgres:13
        env:
          POSTGRES_USER: postgres
          POSTGRES_PASSWORD: postgres
          POSTGRES_DB: test_db
        ports:
          - 5432:5432
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5

    steps:
    - uses: actions/checkout@v2

    - name: Set up Python
      uses: actions/setup-python@v2
      with:
        python-version: '3.9'

    - name: Install dependencies
      run: |
        python -m pip install --upgrade pip
        pip install -r requirements.txt
        pip install pytest-cov

    - name: Run tests
      env:
        DATABASE_URL: postgresql://postgres:postgres@localhost:5432/test_db
      run: |
        pytest --cov=app --cov-report=xml

    - name: Upload coverage
      uses: codecov/codecov-action@v2

  deploy:
    needs: test
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'

    steps:
    - name: Deploy to Heroku
      uses: akhileshns/heroku-deploy@v3.12.12
      with:
        heroku_api_key: ${{secrets.HEROKU_API_KEY}}
        heroku_app_name: ${{secrets.HEROKU_APP_NAME}}
        heroku_email: ${{secrets.HEROKU_EMAIL}}
        dockerfile_path: "Dockerfile"

Production Best Practices {#production-practices}

1. Environment Configuration

from pydantic import BaseSettings, PostgresDsn

class Settings(BaseSettings):
    PROJECT_NAME: str
    ENVIRONMENT: str
    DEBUG: bool = False

    # Database
    POSTGRES_SERVER: str
    POSTGRES_USER: str
    POSTGRES_PASSWORD: str
    POSTGRES_DB: str

    # Security
    SECRET_KEY: str
    ALLOWED_HOSTS: list[str]

    # Performance
    WORKERS_COUNT: int = 4
    RELOAD: bool = False

    class Config:
        env_file = f".env.{ENVIRONMENT}"

settings = Settings()

2. Logging Configuration

import logging
from datetime import datetime

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('app.log'),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)

@app.middleware("http")
async def log_requests(request: Request, call_next):
    start_time = datetime.now()
    response = await call_next(request)
    duration = datetime.now() - start_time

    logger.info(
        f"{request.method} {request.url.path} "
        f"- Status: {response.status_code} "
        f"- Duration: {duration.total_seconds():.2f}s"
    )

    return response

3. Health Check Endpoint

from fastapi import status
from datetime import datetime

@app.get("/health", tags=["Health Check"])
async def health_check():
    return {
        "status": "healthy",
        "timestamp": datetime.now().isoformat(),
        "environment": settings.ENVIRONMENT,
        "version": "1.0.0"
    }

Leave a Reply

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

logo

Selamat datang di blog latihan coding, Kita akan belajar bareng teknologi terkini.

Explore Topics