Forem: Shahar Polak

FastAPI, Furious Tests: The Need for Speed

Shahar Polak — Sat, 13 Sep 2025 16:11:27 +0000

A no-nonsense guide to making your test suite so fast, it'll make The Flash jealous

by Shahar Polak

TL;DR - The "I Just Want Results" Section

Your CI is slow. Your developers are sad. Your coffee budget is through the roof because people are waiting 28 minutes for tests to run. Here's how we turned a sluggish test suite into a speed demon that finishes in under 3 minutes.

The Magic Formula:

SQLite in-memory for 95% of tests
MySQL for the 5% that actually need it
pytest-xdist with work stealing
Proper fixtures (finally!)
Split jobs by runtime, not count

Skip to any section, but if you implement this wrong, don't blame us when your tests become flakier than a croissant factory.

Who Are You? (And Why Should You Care?)

The Frustrated Developer: You write a test, run the suite, grab coffee, check Twitter, contemplate life choices, and maybe your tests are done. Maybe.

The DevOps Engineer: Your CI bill makes your CFO weep, and developers complain about feedback loops longer than a soap opera plotline.

The Tech Lead: You know there's a better way, but every "quick fix" turns into a three-week rabbit hole that breaks everything.

The Project Manager: You just want features shipped without the engineering team disappearing into "test optimization" black holes.

If any of these sound familiar, buckle up. We're about to make your tests faster than your last relationship ended.

The Problem: When Tests Become Slower Than Dial-Up Internet

Picture this: You have 2,900 tests. Your CI takes 28 minutes to run them. That's roughly:

1,680 seconds of pure waiting
27.5 minutes longer than your attention span
Infinite sadness for developers trying to iterate quickly

The Real Cost

It's not just time. It's:

Developers skipping test runs locally (risky business)
Context switching while waiting for CI (productivity killer)
Delayed releases because nobody wants to wait for slow feedback
Your AWS bill looking like a phone number

What We're Optimizing For

Speed without stupidity: Fast tests that still catch bugs
Developer happiness: Quick feedback loops
Maintainability: No Rube Goldberg test machines
Cost efficiency: Your CFO might even smile

The Hall of Failed Attempts (Learn From Our Pain)

Attempt #1: Dockerized MySQL for Everything

"Let's just use production-like everything!"

What we thought: Production parity! MySQL everywhere! What could go wrong?

What actually happened:

Container startup: 10-15 seconds per test run
Migration overhead: Another 5-10 seconds
Parallel runs: Database collisions everywhere
Result: Slower, flakier, sadder

Lesson learned: Production parity is great, but not for every single test.

Attempt #2: Shared Test Database

"One database to rule them all!"

What we thought: Share a test database across all workers. Efficiency!

What actually happened:

Data collisions like bumper cars
Order-dependent tests (the worst kind)
Cleanup between tests took forever
Flakiness increased by approximately 1000%

Lesson learned: Shared state is the root of all evil in testing.

Attempt #3: Self-Hosted AWS Runners

"We'll just throw more hardware at it!"

What we thought: Bigger machines, faster tests, problem solved!

What actually happened:

Setup complexity through the roof
Security headaches
Actually slower than managed runners
Higher costs (ouch)

Lesson learned: Sometimes the boring solution is the right solution.

The Solution: SQLite in-Memory (The Chosen One)

Here's the controversial take: You don't need MySQL for most of your tests.

The SQLite Advantage

Speed: In-memory databases are lightning fast
Isolation: Fresh database per test (or per worker)
Simplicity: No containers, no networking, no tears

But Wait, There's a Catch

SQLite isn't MySQL. Shocking, we know. Here's what you need to handle:

The 10 Commandments of SQLite Testing

Foreign Keys: SQLite doesn't enforce them by default. Add ?foreign_keys=1 to your connection string or face the consequences.
Type Affinity: SQLite is more forgiving with data types. Your integer column will happily store "banana" unless you use strict tables.
Auto Increment: Different behavior from MySQL's AUTO_INCREMENT. Test carefully if you rely on specific ID sequences.
Concurrency: SQLite serializes writes. MySQL doesn't. Keep some MySQL tests for concurrency edge cases.
Transaction Isolation: Different default isolation levels. If your code depends on specific isolation behavior, test on MySQL.
DDL Migrations: Limited ALTER TABLE support in SQLite. Test your migrations on the real database.
Date/Time Handling: Different timezone and format handling. Verify timestamp-sensitive logic on MySQL.
JSON Functions: SQLite's JSON support differs from MySQL's native JSON type and functions.
Collations: Case sensitivity and sorting behavior can differ. Test locale-specific queries on MySQL.
Database Engine Features: If you use stored procedures, triggers, or MySQL-specific functions, you need the real deal.

The Implementation: Code That Actually Works

Project Structure (The Foundation)

your-awesome-project/
├── app/
│   ├── __init__.py
│   ├── main.py              # FastAPI app factory
│   ├── db.py                # Database configuration
│   ├── models/              # Your beautiful models
│   └── routers/             # API routes
├── tests/
│   ├── unit/                # Pure logic, no I/O
│   ├── component/           # FastAPI + SQLite tests
│   ├── e2e/                 # The "real deal" MySQL tests
│   └── conftest.py          # Fixture magic happens here
├── pytest.ini              # Configuration central
└── requirements-dev.txt     # All the testing goodies

The Database Setup (app/db.py)

# app/db.py
from contextlib import asynccontextmanager
from typing import Optional
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

# Global sessionmaker (we'll configure this in tests)
SessionLocal: Optional[async_sessionmaker[AsyncSession]] = None

def configure_sessionmaker(sessionmaker: async_sessionmaker[AsyncSession]) -> None:
    """Configure the global sessionmaker. Call this in your app startup and test fixtures."""
    global SessionLocal
    SessionLocal = sessionmaker

@asynccontextmanager
async def get_db() -> AsyncSession:
    """FastAPI dependency for database sessions."""
    assert SessionLocal is not None, "SessionLocal not configured. Did you forget to call configure_sessionmaker?"
    async with SessionLocal() as session:
        yield session

The FastAPI App (app/main.py)

# app/main.py
from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from .db import get_db
from .models import User  # Your SQLModel models

def create_app() -> FastAPI:
    """App factory pattern. Makes testing easier and dependency injection cleaner."""
    app = FastAPI(title="Lightning Fast API")

    @app.get("/healthz")
    async def health_check():
        """The most important endpoint. If this is broken, everything is broken."""
        return {"status": "ok", "message": "Still alive!"}

    @app.post("/users", response_model=User)
    async def create_user(
        email: str, 
        name: str, 
        db: AsyncSession = Depends(get_db)
    ):
        """Create a user. Revolutionary stuff."""
        user = User(email=email, name=name)
        db.add(user)
        await db.commit()
        await db.refresh(user)
        return user

    @app.get("/users/{user_id}", response_model=User)
    async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
        """Get a user by ID. Also revolutionary."""
        user = await db.get(User, user_id)
        if not user:
            raise HTTPException(status_code=404, detail="User not found")
        return user

    return app

The Test Configuration (pytest.ini)

[pytest]
addopts = -ra -q -m "not e2e"
testpaths = tests
markers =
    unit: Pure logic tests, no I/O, fast as lightning
    component: FastAPI + SQLite tests, reasonably fast
    e2e: Full stack with MySQL, use sparingly
    slow: Tests that take >1 second (we're watching you)
asyncio_mode = auto

The Fixture Magic (tests/conftest.py)

# tests/conftest.py
import os
import pytest
from httpx import AsyncClient
from sqlmodel import SQLModel
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from app.main import create_app
from app.db import configure_sessionmaker, get_db

@pytest.fixture(scope="session")
def worker_id():
    """Get the pytest-xdist worker ID. Each worker gets its own database file."""
    return os.getenv("PYTEST_XDIST_WORKER", "gw0")

@pytest.fixture(scope="session")
async def engine(worker_id):
    """
    Create a SQLite engine per worker. 

    Per-worker file databases are more reliable than :memory: for parallel testing.
    Each worker gets its own file, avoiding connection sharing issues.

    CRITICAL: foreign_keys=1 in connection string enables FK for ALL connections.
    """
    db_file = f"./test_db_{worker_id}.sqlite3"
    # FIXED: foreign_keys=1 in connection string, not per-connection pragma
    engine = create_async_engine(f"sqlite+aiosqlite:///{db_file}?foreign_keys=1", echo=False)

    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

    yield engine

    # Cleanup
    await engine.dispose()
    try:
        os.remove(db_file)
    except FileNotFoundError:
        pass  # Already deleted or never created

@pytest.fixture(scope="session")
def sessionmaker(engine):
    """Create a sessionmaker factory."""
    return async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

@pytest.fixture(autouse=True, scope="session")
def _configure_sessionmaker(sessionmaker):
    """Auto-configure the global sessionmaker for all tests."""
    configure_sessionmaker(sessionmaker)

@pytest.fixture
async def app(sessionmaker):
    """Create a FastAPI app with test database dependency override."""
    app = create_app()

    # Override the database dependency
    async def get_test_db():
        async with sessionmaker() as session:
            yield session

    app.dependency_overrides[get_db] = get_test_db
    return app

@pytest.fixture
async def client(app):
    """HTTP client for testing FastAPI endpoints."""
    async with AsyncClient(app=app, base_url="http://testserver") as client:
        yield client

# E2E fixtures for MySQL testing
@pytest.fixture(scope="session") 
def mysql_url():
    """MySQL connection URL for e2e tests."""
    return os.getenv("TEST_MYSQL_URL", "mysql+asyncmy://root:test@localhost:3306/testdb")

@pytest.fixture(scope="session")
def run_e2e(pytestconfig):
    """Check if e2e tests are selected."""
    selected_markers = pytestconfig.getoption("-m") or ""
    return "e2e" in selected_markers

@pytest.fixture(scope="session")
async def mysql_engine(run_e2e, mysql_url):
    """MySQL engine for e2e tests only."""
    if not run_e2e:
        pytest.skip("e2e tests not selected")

    engine = create_async_engine(mysql_url, echo=False)
    async with engine.begin() as conn:
        # Run migrations or create tables here
        await conn.run_sync(SQLModel.metadata.create_all)

    yield engine
    await engine.dispose()

@pytest.fixture
async def e2e_app(mysql_engine):
    """FastAPI app configured with MySQL for e2e tests."""
    # FIXED: Use existing configure_sessionmaker function consistently
    mysql_sessionmaker = async_sessionmaker(mysql_engine, class_=AsyncSession, expire_on_commit=False)
    configure_sessionmaker(mysql_sessionmaker)
    app = create_app()
    return app

Test Data Factories (tests/factories.py)

# tests/factories.py
import factory
from faker import Faker
from app.models import User

# Deterministic fake data for consistent tests
fake = Faker()
Faker.seed(1337)

class UserFactory(factory.Factory):
    class Meta:
        model = User

    email = factory.LazyAttribute(lambda _: fake.unique.email())
    name = factory.LazyAttribute(lambda _: fake.name())

# Helper for creating database entities
from contextlib import asynccontextmanager
from sqlalchemy.ext.asyncio import AsyncSession

@asynccontextmanager
async def create_entity(session: AsyncSession, entity):
    """Create and persist an entity, yielding it for use in tests."""
    session.add(entity)
    await session.flush()  # Get the ID without committing
    yield entity
    # Cleanup happens automatically when session ends

Sample Tests That Actually Work

# tests/unit/test_math.py
import pytest

@pytest.mark.unit
def test_addition_still_works():
    """The most important test. If this fails, we have bigger problems."""
    assert 1 + 1 == 2

@pytest.mark.unit  
def test_our_business_logic():
    """Test your actual business logic here."""
    from app.utils import calculate_something_important
    result = calculate_something_important(42)
    assert result == "The answer to everything"

# tests/component/test_users.py
import pytest
from tests.factories import UserFactory, create_entity

@pytest.mark.component
async def test_create_user_endpoint(client):
    """Test user creation through the API."""
    response = await client.post("/users", json={
        "email": "test@example.com",
        "name": "Test User"
    })

    assert response.status_code == 200
    data = response.json()
    assert data["email"] == "test@example.com"
    assert data["name"] == "Test User"
    assert data["id"] is not None

@pytest.mark.component
async def test_get_user_endpoint(client, sessionmaker):
    """Test getting a user by ID."""
    # Create a user in the database
    async with sessionmaker() as session:
        async with create_entity(session, UserFactory.build()) as user:
            await session.commit()

            # Test the endpoint
            response = await client.get(f"/users/{user.id}")
            assert response.status_code == 200

            data = response.json()
            assert data["id"] == user.id
            assert data["email"] == user.email

@pytest.mark.component
async def test_get_nonexistent_user(client):
    """Test 404 behavior for missing users."""
    response = await client.get("/users/99999")
    assert response.status_code == 404
    assert "not found" in response.json()["detail"].lower()

# tests/e2e/test_mysql_specifics.py
import pytest

@pytest.mark.e2e
async def test_foreign_key_constraints(e2e_client, mysql_sessionmaker):
    """Test that foreign key constraints work properly on MySQL."""
    # This test would fail on SQLite without proper FK setup
    # but should work correctly on MySQL
    pass

@pytest.mark.e2e  
async def test_transaction_isolation(e2e_client):
    """Test MySQL-specific transaction behavior."""
    # Test concurrent access patterns that behave differently on MySQL
    pass

@pytest.mark.e2e
async def test_json_queries(e2e_client):
    """Test MySQL JSON functions that SQLite doesn't support."""
    pass

The Parallelization: pytest-xdist Wizardry

Basic Parallelization

# Start with this
pytest -n auto

# Graduate to this
pytest -n logical --dist worksteal

Why `--dist worksteal`?

When some tests are slower than others (looking at you, that one test that takes 30 seconds), work stealing lets idle workers grab tasks from busy workers. It's like having helpful coworkers instead of the ones who disappear when there's work to do.

Avoiding Parallel Test Hell

The Golden Rules:

No shared state between tests: Each test should be an island
Use unique file paths: tmp_path fixture or worker-specific names
Clean up globals: Reset singletons or make them worker-aware
Deterministic test data: Seed your faker, control your randomness

Ensuring Test Independence

Instead of relying on randomization to catch order dependencies:

Use proper fixtures for test isolation (per-worker databases, clean state)
pytest-xdist parallel execution will reveal most order issues naturally
Careful test design prevents dependencies from forming in the first place

If tests pass individually but fail in parallel, check for:

Global variables being modified
File system conflicts (use tmp_path fixture)
Database record pollution between tests

Debugging Parallel Failures

When tests work individually but fail in parallel:

# Reproduce the failure
pytest -n 0 -k "failing_test_name" -vv -s

# Check for shared state
# Look for global variables, file system conflicts, or database sharing

The CI Configuration: GitHub Actions That Don't Suck

The Fast Lane (SQLite Tests)

# .github/workflows/tests.yml
name: Tests That Actually Finish
on: [push, pull_request]

concurrency:
  group: ci-${{ github.ref }}
  cancel-in-progress: true

jobs:
  fast-lane:
    name: Fast Tests (SQLite)
    strategy:
      fail-fast: false
      matrix: 
        group: [1, 2, 3, 4]  # Split into 4 parallel jobs
    runs-on: ubuntu-latest
    timeout-minutes: 15  # If it takes longer, something's wrong

    steps:
      - name: Checkout code
        uses: actions/checkout@v4

      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.12'
          cache: 'pip'

      - name: Install dependencies
        run: |
          pip install -r requirements-dev.txt

      - name: Run fast tests
        env:
          PYTEST_SPLIT_FILE: .pytest-split-durations.json
        run: |
          pytest -m "not e2e" \
                 --splits 4 --group ${{ matrix.group }} \
                 -n logical --dist worksteal \
                 --alluredir=allure-results/${{ matrix.group }} \
                 --durations=25

      - name: Upload test results
        uses: actions/upload-artifact@v4
        if: always()
        with:
          name: allure-fast-${{ matrix.group }}
          path: allure-results/${{ matrix.group }}

The E2E Lane (MySQL Tests)

  e2e-tests:
    name: E2E Tests (MySQL)
    needs: [fast-lane]  # Only run if fast tests pass
    runs-on: ubuntu-latest
    timeout-minutes: 20

    services:
      mysql:
        image: mysql:8.0
        env:
          MYSQL_ROOT_PASSWORD: test
          MYSQL_DATABASE: testdb
        ports:
          - 3306:3306
        options: >-
          --health-cmd="mysqladmin ping -h 127.0.0.1 -ptest"
          --health-interval=5s
          --health-timeout=2s
          --health-retries=20

    steps:
      - name: Checkout code
        uses: actions/checkout@v4

      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.12'
          cache: 'pip'

      - name: Install dependencies
        run: pip install -r requirements-dev.txt

      - name: Wait for MySQL
        run: |
          for i in {1..30}; do
            mysqladmin ping -h 127.0.0.1 -ptest && break
            sleep 2
          done

      - name: Run E2E tests
        env:
          TEST_MYSQL_URL: "mysql+asyncmy://root:test@127.0.0.1:3306/testdb"
        run: |
          pytest -m e2e \
                 -n 0 \
                 --alluredir=allure-results/e2e \
                 --durations=25

      - name: Upload E2E results
        uses: actions/upload-artifact@v4
        if: always()
        with:
          name: allure-e2e
          path: allure-results/e2e

### Taking CI Speed Even Further: Blacksmith.sh
> No they do not pay me to say that, I just truly love them 🙏

![ ](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/l3p03np6zz2xqvd8ibbd.jpg)
After optimizing our tests, we discovered our CI runners were the next bottleneck. GitHub Actions runners, while convenient, aren't optimized for compute-heavy workloads like parallel test execution.

We migrated to [Blacksmith.sh](https://blacksmith.sh) for our CI infrastructure and saw additional improvements:

**Performance gains:**
- **2-3x faster** build times due to dedicated bare-metal runners
- **More predictable** performance (no noisy neighbor problems)
- **Better parallelization** support with higher core counts

**Cost savings:**
- **40-60% reduction** in CI costs compared to GitHub Actions
- Pay only for actual usage, not idle time
- Better price/performance ratio for compute-intensive workflows

**Migration effort:**
- **1-3 lines of code** changes per repository
- Minimal disruption to existing workflows
- Same GitHub Actions syntax, just different runners

yaml

Before (GitHub Actions)

runs-on: ubuntu-latest

After (Blacksmith)

runs-on: blacksmith-4vcpu-ubuntu-2204


The combination of optimized tests + faster runners reduced our total CI time from 28 minutes to under 2 minutes, while cutting costs by over 70%.

**When to consider dedicated CI infrastructure:**
- High-frequency builds (multiple times per hour)
- Compute-intensive test suites (parallel execution, compilation)
- Cost-sensitive environments where CI bills are significant
- Teams that value predictable, fast feedback loops

Smart Test Splitting with pytest-split

Most people split tests by count: "Run tests 1-100 on worker 1, 101-200 on worker 2." This is like dividing a pizza by number of slices instead of size - you might get one tiny slice and one massive one.

The Problem with Count-Based Splitting

When you have tests that take vastly different amounts of time (some finish in 0.1s, others take 30s), splitting by count leads to:

Job 1: 100 fast tests (finishes in 2 minutes)
Job 2: 100 tests including 3 slow ones (finishes in 15 minutes)
Your CI is only as fast as the slowest job

The pytest-split Solution

pytest-split distributes tests based on actual runtime data, ensuring each CI job takes roughly the same total time.

Step 1: Generate Timing Data

First, collect timing data from your test suite:

# Run this once to generate timing data
pytest --store-durations --durations-path=.pytest-durations.json -m "not e2e"

This creates a .pytest-durations.json file with historical timing data for each test.

Step 2: Use Runtime-Based Splitting in CI

# Split tests into 4 groups based on runtime, not count
export PYTEST_SPLIT_FILE=.pytest-durations.json

# Each group gets ~25% of total runtime
pytest --splits 4 --group 1 -m "not e2e"  # Group 1
pytest --splits 4 --group 2 -m "not e2e"  # Group 2
pytest --splits 4 --group 3 -m "not e2e"  # Group 3
pytest --splits 4 --group 4 -m "not e2e"  # Group 4

When to Use pytest-split

Use pytest-split when:

Your tests have highly variable runtimes (some 0.1s, others 10s+)
You're running multiple CI jobs and jobs finish at very different times
You have a large test suite (1000+ tests) where timing distribution matters
You want predictable CI job completion times

Skip pytest-split when:

Your tests have relatively uniform runtime (all finish within 1-2 seconds)
You're only running one CI job
Your test suite is small (<200 tests)
--dist worksteal already balances your workload well

Setting Up pytest-split for Your Project

Install pytest-split:

pip install pytest-split

Generate initial timing data:

# Run your full test suite once to collect timings
pytest --store-durations --durations-path=.pytest-durations.json

Update your CI configuration:

- name: Run fast tests
  env:
    PYTEST_SPLIT_FILE: .pytest-durations.json
  run: |
    pytest -m "not e2e" \
           --splits 4 --group ${{ matrix.group }} \
           -n logical --dist worksteal \
           --durations=25

Keep timing data updated:

# Regenerate timing data weekly or when adding many new tests
pytest --store-durations --durations-path=.pytest-durations.json -m "not e2e"

pytest-split Best Practices

Timing Data Maintenance:

Regenerate timing data when you add/remove many tests
Update weekly for active projects, monthly for stable ones
Store the .pytest-durations.json file in your repo

Debugging Imbalanced Jobs:

Check timing data: cat .pytest-durations.json | head -20
Look for tests without timing data (new tests default to 0 seconds)
Verify your job matrix splits match your --splits parameter

Combining with worksteal:

# Best of both worlds: balanced groups + work stealing within groups
pytest --splits 4 --group ${{ matrix.group }} -n logical --dist worksteal

This ensures each CI job gets a balanced workload, and within each job, workers can steal work from each other when they finish early.

Memory Management and Cleanup

Memory Guidelines

With 8-core runners and 32GB RAM, your test suite should rarely exceed:

4-6 GB RSS during parallel execution
Fail the build if you hit 8GB+ (something's leaking)

Monitoring Memory Usage

# Add this to your measurement script
for i in 1 2 3 4 5; do
  /usr/bin/time -f "cpu=%P rss=%Mkb elapsed=%E" \
  pytest -m "not e2e" -n logical --dist worksteal \
         --durations=25 -q | tee reports/run_$i.log
done

SQLite File Cleanup

The per-worker SQLite files need cleanup:

# In your engine fixture
yield engine
await engine.dispose()

# Clean up the database file
try:
    os.remove(f"./test_db_{worker_id}.sqlite3")
except FileNotFoundError:
    pass  # Already gone

Measurement Protocol (No More Hand-Wavy Claims)

Hardware Baseline

Record this info and include it in your repo:

## Test Environment Specifications

**Local Development:**
- MacBook Pro M2 Max, 12-core CPU, 32GB RAM
- macOS 14.2, Python 3.12

**CI Environment:**  
- GitHub Actions ubuntu-latest
- 4-core x86_64, 16GB RAM, SSD storage
- Python 3.12, pytest 8.x

The 5-Run Protocol

#!/bin/bash
# scripts/measure_performance.sh

echo "=== Test Performance Measurement ==="
echo "Hardware: $(uname -m) $(uname -s)"  
echo "Python: $(python --version)"
echo "Cores: $(nproc)"
echo "RAM: $(free -h | grep '^Mem:' | awk '{print $2}')"
echo ""

# Warm up (don't count this run)
echo "Warming up..."
pytest -m "not e2e" -n logical --dist worksteal -q > /dev/null

# Measure 5 runs
echo "Measuring performance across 5 runs..."
for i in {1..5}; do
    echo "Run $i/5"
    /usr/bin/time -f "run_${i}: cpu=%P rss=%Mkb elapsed=%E" \
    pytest -m "not e2e" \
           -n logical --dist worksteal \
           --durations=25 -q \
           --alluredir=allure-results/perf_run_${i} \
           2>&1 | tee reports/perf_run_${i}.log
done

echo ""
echo "=== Results Summary ==="
grep "elapsed=" reports/perf_run_*.log | sort

Results Table Format

Always include these details with any performance claims:

## Performance Results

**Test Environment:**
- Hardware: [Your specific hardware]
- OS: [Specific version]
- Python: [Version]
- Test count: [Total number] ([breakdown by type])

**Before:** [Describe the baseline setup]
**After:** [Describe the optimized setup]

| Metric                    | Before    | After   | Notes |
|---------------------------|-----------|---------|--------|
| Median wall time (5 runs) | 28m 34s  | 3m 47s  | Measured with `/usr/bin/time` |
| P95 wall time             | 31m 12s  | 4m 15s  | Worst case scenario |
| Memory peak (RSS)         | 8.2 GB   | 4.1 GB  | Peak during parallel execution |
| Test composition          | 100% MySQL | 93% SQLite, 7% MySQL | [breakdown] |

**Methodology:** 5 consecutive runs, median reported, hardware specs documented above.

Red Flags to Avoid:

Claims without specific hardware mentioned
No test composition breakdown
Hand-wavy estimates ("~9x faster")
Missing measurement methodology
Comparing different hardware/environments

Common Pitfalls (Learn From Others' Mistakes)

The "It Works on My Machine" Special

Problem: Tests pass locally but fail in CI.

Usually caused by:

Different database versions
Timezone differences
File path separators (Windows vs. Unix)
Environment variables not set in CI

Solution: Make your test environment configuration explicit and reproducible.

The "Shared State Surprise"

Problem: Tests pass individually but fail when run together.

Usually caused by:

Global variables being modified
Database records not cleaned up
File system state persisting
Module-level imports with side effects

Solution: Embrace test isolation like it's your religion.

The "Foreign Key Forgotten"

Problem: Tests pass but production breaks due to referential integrity violations.

Usually caused by: SQLite's PRAGMA foreign_keys=OFF default behavior, or incorrectly setting FK pragma per connection instead of in the connection string.

Solution: Use ?foreign_keys=1 in your SQLite connection string, not per-connection pragma statements.

The "Performance Regression Creep"

Problem: Tests gradually get slower over time.

Usually caused by:

New tests being added without performance consideration
Test data growing without bounds
Dependencies being added carelessly

Solution: Monitor your test durations and fail builds when they exceed thresholds.

The "False Confidence Trap"

Problem: Fast tests give you confidence, but they don't catch real bugs.

Usually caused by: Mocking too much or using overly simplified test scenarios.

Solution: Balance fast feedback with meaningful coverage. Keep that MySQL test suite!

The Incremental Migration Plan

Don't try to do everything at once. You'll break things, blame this guide, and we'll both be sad.

Phase 1: Measure Everything

Implement the 5-run measurement protocol
Document current performance
Identify the 10 slowest tests

Success criteria: You have baseline numbers you can defend.

Phase 2: Fix the Fixtures

Implement proper test database isolation
Add the session-scoped engine fixture
Keep using MySQL initially

Success criteria: Tests are isolated but not necessarily faster yet.

Phase 3: Switch to SQLite

Convert component tests to use SQLite fixtures
Keep a small MySQL suite for edge cases
Add ?foreign_keys=1 to connection string

Success criteria: Most tests use SQLite, MySQL suite is <5% of total.

Phase 4: Enable Parallelization

Start with pytest -n auto
Fix any parallel test failures
Verify test isolation with proper fixtures

Success criteria: All tests pass consistently with -n auto.

Phase 5: Consider Runtime-Based Splitting (Optional)

First try: --dist worksteal to see if it resolves job imbalance
If needed: Implement manual category-based job splitting
Advanced: Add pytest-split only if you have stable timing patterns and maintenance capacity

Success criteria: CI jobs finish within 2-3 minutes of each other, no job consistently much slower.

Phase 6: Polish and Monitor

Add memory monitoring
Set up performance regression detection
Document what belongs in MySQL vs. SQLite suites

Success criteria: Performance is stable and regressions are caught early.

Troubleshooting Guide

"My tests are using :memory: but still failing in parallel"

Likely cause: Multiple connections to the same in-memory database.

Fix: Switch to per-worker file databases as shown in the fixtures.

"SQLite tests pass but MySQL tests fail"

Likely cause: You found a real difference between the databases.

Fix: Either adjust your code to be database-agnostic, or move that test case to the MySQL suite permanently.

"One worker is much slower than others"

Likely cause: Imbalanced test distribution.

Fix: Use pytest-split with actual timing data, not just test counts.

"Tests are fast but flaky"

Likely causes:

Shared global state
Race conditions in parallel execution
Non-deterministic test data
Insufficient test isolation

Fix: Add more debugging output, check for global variables, ensure proper fixture isolation.

"Memory usage is through the roof"

Likely causes:

Database connections not being closed
Large test data sets being held in memory
Circular references preventing garbage collection

Fix: Add explicit connection cleanup, use smaller test datasets, profile with memory_profiler.

The Results (Finally!)

After implementing this approach across multiple projects, here's what we typically see:

Performance Improvements

Metric	Typical Before	Typical After	Improvement
Total wall time (CI)	25-35 minutes	3-6 minutes	6-10x faster
Local test feedback	5-10 minutes	30-60 seconds	5-10x faster
Memory usage (peak)	6-12 GB	2-4 GB	50-70% less
CI cost (monthly)	$400-800	$150-300	60-70% savings
Developer happiness	😢	😊	Priceless

What You Get

Faster feedback loops: Developers actually run tests locally again.

Reliable CI: No more "let's just restart the build and hope" debugging sessions.

Lower costs: Your AWS bill will thank you.

Better test coverage: When tests are fast, people write more of them.

Fewer production bugs: Fast tests get run more often, catching issues earlier.

What You Don't Get

Perfect MySQL compatibility: You'll still need that small MySQL test suite.

Zero maintenance: Test infrastructure still needs care and feeding.

Magic bullet: Some tests will still be inherently slow - that's okay.

The Complete Checklist

Print this out and check items off as you implement them:

Setup Phase

[ ] Document current test performance (5-run protocol)
[ ] Set up project structure with proper test directories
[ ] Configure pytest.ini with markers and asyncio mode
[ ] Create the database abstraction layer (app/db.py)

Fixtures Phase

[ ] Implement session-scoped SQLite engine fixture
[ ] Add per-worker database file creation
[ ] Set up ?foreign_keys=1 in connection string
[ ] Create FastAPI app fixture with dependency overrides
[ ] Add HTTP client fixture

Testing Phase

[ ] Convert component tests to use SQLite fixtures
[ ] Set up test data factories with deterministic seeds
[ ] Identify and preserve critical MySQL test cases
[ ] Add E2E fixtures for MySQL testing

Parallelization Phase

[ ] Enable pytest -n auto and fix any failures
[ ] Upgrade to pytest -n logical --dist worksteal
[ ] Ensure test isolation with proper fixtures
[ ] Verify no shared state between tests

CI Phase

[ ] Split fast tests into multiple parallel jobs
[ ] Set up MySQL service for E2E tests
[ ] Consider pytest-split for runtime-based job splitting
[ ] Add Allure or similar reporting
[ ] Configure artifact upload for test results

Monitoring Phase

[ ] Add memory usage monitoring
[ ] Set up performance regression detection
[ ] Document what belongs in MySQL vs SQLite suites
[ ] Create runbook for troubleshooting parallel test failures

Resources and Further Reading

Essential Tools

pytest-xdist: Parallel test execution - docs
pytest-split: Runtime-based test splitting - GitHub
factory_boy: Test data factories - docs
Faker: Generate fake data - docs

FastAPI Testing

Official FastAPI Testing Guide: fastapi.tiangolo.com/tutorial/testing
AsyncClient documentation: For testing async endpoints

Database Testing

SQLAlchemy Testing: docs.sqlalchemy.org/en/20/orm/session_transaction.html
SQLite vs MySQL differences: Know what you're getting into

CI/CD Optimization

GitHub Actions best practices: Caching, artifacts, and parallelization
pytest-split documentation: Runtime-based test distribution

FAQ (The Questions You're Going to Ask)

Q: "Should I use SQLite for ALL tests?"

A: No. Use SQLite for the majority (90-95%) but keep MySQL tests for database-specific behavior, migrations, and edge cases that depend on MySQL semantics.

Q: "What about other databases like PostgreSQL?"

A: The same principles apply. SQLite for speed, PostgreSQL for the cases that need PostgreSQL-specific features. Adjust the edge cases accordingly.

Q: "How do I know what tests need the real database?"

A: Start by converting everything to SQLite. The tests that break will tell you what needs the real database. Common candidates: migrations, JSON queries, stored procedures, specific isolation levels.

Q: "Is this approach suitable for microservices?"

A: Absolutely. Each service can have its own fast test suite with selective real database testing for integration points.

Q: "What about integration tests with external services?"

A: Mock them in your component tests, test them for real in your E2E suite. Same principle as SQLite vs MySQL.

Q: "How often should I regenerate test timing data?"

A: Weekly for active projects, monthly for stable ones. Whenever you add a lot of new tests or notice job imbalance.

Q: "Can I use this with other web frameworks?"

A: Yes! The core principles (fast database, fixtures, parallelization) apply to Django, Flask, or any other framework. Adjust the specifics accordingly.

Q: "What if my tests are CPU-bound instead of I/O-bound?"

A: Parallelization will still help, but you might need to optimize algorithms rather than just database access. Profile to find the bottlenecks.

Q: "How do I convince my team to adopt this?"

A: Start with measuring current performance. Numbers don't lie. Implement incrementally and show results at each phase.

Conclusion: Go Forth and Test Quickly

You now have a complete, battle-tested approach to making your FastAPI tests faster without sacrificing reliability. This isn't theoretical - it's been used to optimize test suites across multiple production systems.

The Key Insights

Speed and quality aren't mutually exclusive. You can have fast tests that catch real bugs by being strategic about what you test where.

Infrastructure matters. Proper fixtures, isolation, and parallelization are force multipliers for test performance.

Measure everything. Without numbers, you're just guessing. With numbers, you can make informed decisions and track progress.

Incremental improvement beats big rewrites. You can adopt this approach piece by piece without breaking your existing workflow.

What's Next?

Implement incrementally: Don't try to do everything at once
Measure before and after: Document your improvements
Share your results: Help others learn from your experience
Iterate and improve: This is a living system, not a one-time fix

A Final Warning

Fast tests are addictive. Once your team experiences 3-minute CI runs instead of 30-minute ones, there's no going back. You'll find yourself optimizing other slow things in your development process.

Don't say we didn't warn you.

Now stop reading and start implementing. Your future self (and your teammates) will thank you.

About the Author

Shahar Polak has spent way too much time waiting for slow tests and decided to do something about it. When not optimizing test suites, he can be found building things with FastAPI, SQLAlchemy, and probably complaining about some CI pipeline somewhere.

Got questions? Found a bug in this guide? Implemented this successfully (or unsuccessfully)? Share your experience - the testing community learns from real-world results, not just theory.

Great article, I love every single tool here

Shahar Polak — Wed, 23 Apr 2025 22:07:39 +0000

Jonas Scholz

Apr 21 '25

How I save $$$ by self-hosting these 5 open source tools

#opensource #cloud #devops #webdev

164

Comments 16

4 min read

Stopping Errors Before They Stop You

Shahar Polak — Thu, 19 Sep 2024 15:01:08 +0000

Stopping Errors Before They Stop You: The Safe Assignment Operator (`?=`) and Handling Promises Gracefully

As JavaScript evolves, new features and proposals keep rolling in, aiming to make coding more efficient and error-proof. One such feature is the Safe Assignment Operator (?=), a proposed addition to the language. While we're still waiting for its official release, we can implement similar functionality today to safeguard our code from common issues like null or undefined values.

In this article, we’ll explore the ?= operator, build our own version using existing JavaScript, and introduce practical ways to handle promises more gracefully in asynchronous operations.

Understanding the Safe Assignment Operator (`?=`)

What is the `?=` Operator?

The Safe Assignment Operator (?=) allows developers to assign a value to a variable only if the target is null or undefined. It’s a more concise way of saying, "Assign this value if the variable is empty."

Here's how it works:

let username = null;
username ?= "Shahar"; 
console.log(username); // Output: "Shahar"

In this case, the variable username gets assigned "Shahar" because its value was null. If username had an existing value, the operator would simply pass over the assignment.

Why It's Useful

The ?= operator simplifies code by reducing the need for explicit if checks or ternary operations to ensure safe assignment. However, this operator is still in the proposal stage within ECMAScript, meaning it could change before becoming part of the JavaScript language. You can track its development here.

Crafting a Safe Assignment Function

Rolling Out `safeAssign`

While we're waiting for ?= to become official, we can mimic its behavior today using a custom utility function called safeAssign. This function uses the nullish coalescing operator (??), which is already widely supported in modern environments.

Here’s our safeAssign function:

function safeAssign(target, value) {
  return target ?? value;
}

Example in Action

Let’s see how it works:

let username = undefined;
username = safeAssign(username, "Shahar");
console.log(username); // Output: "Shahar"

This is effectively what the ?= operator would do. If the variable is null or undefined, we assign it a value; otherwise, we leave it untouched.

Limitations of `safeAssign`

While safeAssign provides similar functionality to ?=, it has limitations:

Simplicity: safeAssign is a utility function and cannot provide the same level of syntactic elegance as the native ?= operator. Overusing custom functions can lead to more verbose code.
Performance: Although the performance impact of safeAssign is negligible in small-scale applications, native operators like ?= will likely be faster in larger-scale systems due to engine optimizations.
Browser Support: The nullish coalescing operator (??) used in safeAssign is supported in most modern browsers and environments, but older environments may not support it without polyfills.

A Quick Comparison with Other Languages

Many other languages offer similar features to the proposed ?= operator:

C# has the null-coalescing assignment operator (??=), which behaves similarly to JavaScript’s ?= proposal.
Python uses the or keyword for safe assignments, where a = a or value is a common pattern to assign a value only if a is falsy.

These operators make handling potentially empty values more straightforward, reducing boilerplate code.

Handling Asynchronous Operations with `safeAwait`

Introducing `safeAwait`

When working with asynchronous operations in JavaScript, it’s easy to run into rejected promises or unexpected results. Instead of manually handling every rejection with .catch(), we can streamline the process using a custom function called safeAwait, which wraps promises in a cleaner, safer structure.

Here’s the safeAwait function:

async function safeAwait(promise, errorHandler) {
  try {
    const data = await promise;
    return [null, data]; // Success: No error, return the data
  } catch (error) {
    if (errorHandler) errorHandler(error); // Optional error handler
    return [error, null]; // Error occurred, return error with null data
  }
}

Example: Fetching Data with Error Handling

Let’s use safeAwait to fetch data from an API and handle potential errors:

async function getData() {
  const [error, response] = await safeAwait(
    fetch("https://api.example.com"),
    (err) => console.error("Request failed:", err)
  );

  if (error) return; // Exit if there's an error
  return response; // Return response if successful
}

In this example, safeAwait handles both the success and error cases, allowing the calling function to handle the result in a more predictable way.

Variations of `safeAwait`

We can also extend safeAwait for different use cases. For instance, here’s a version that retries the promise once before failing:

async function safeAwaitWithRetry(promise, errorHandler, retries = 1) {
  let attempt = 0;
  while (attempt <= retries) {
    const [error, data] = await safeAwait(promise, errorHandler);
    if (!error) return [null, data];
    attempt++;
  }
  return [new Error("Max retries reached"), null];
}

This variation retries the promise up to a specified number of times before throwing in the towel.

Best Practices for Error Handling in JavaScript

When working with asynchronous code, proper error handling is crucial. Here are some best practices:

Always handle rejected promises: Unhandled promise rejections can lead to crashes or undefined behavior. Use try/catch or .catch() to ensure promises are properly handled.
Centralize error handling: Utility functions like safeAwait allow you to centralize error handling, making it easier to manage and debug your code.
Graceful degradation: Ensure that your application can recover from errors gracefully without crashing or leaving the user in an undefined state.
Use custom error messages: When throwing errors, provide meaningful error messages to help with debugging.

Before and After: Clean Code with `safeAssign` and `safeAwait`

Here’s a quick comparison of how these utilities can clean up your code.

Without `safeAssign`:

if (user === null || user === undefined) {
  user = "Shahar";
}

With `safeAssign`:

user = safeAssign(user, "Shahar");

Without `safeAwait`:

try {
  const response = await fetch("https://api.example.com");
} catch (error) {
  console.error("Request failed:", error);
}

With `safeAwait`:

const [error, response] = await safeAwait(fetch("https://api.example.com"), (err) => console.error("Request failed:", err));

Conclusion

In summary, while the Safe Assignment Operator (?=) is still a proposal, we can replicate its behavior today using the safeAssign function for nullish values and safeAwait for more complex asynchronous operations. Both utilities simplify your code, making it more readable and maintainable.

Key Takeaways:

The ?= operator simplifies safe assignments but is still in the proposal stage.
You can replicate ?= functionality with safeAssign using the nullish coalescing operator (??), which is widely supported.
For asynchronous operations, safeAwait provides a cleaner way to handle promise rejections and errors.
Keep an eye on ECMAScript proposals for future updates.

By leveraging these patterns, you can handle errors like a pro and keep your code clean, readable, and safe.

Blast Off with Python: Mastering Concurrent Processing in Space and Beyond

Shahar Polak — Mon, 27 Nov 2023 07:56:17 +0000

🚀 Welcome, Python space explorers and data scientists! Are you ready to supercharge your Python scripts and bid farewell to the sluggishness of sequential execution? Embrace the power of concurrent.futures!

This module isn't just a tool; it's your mission control for parallelizing tasks, launching your code from the mundane to the realms of high-efficiency computing.

A big shout-out to my dear friend Adar Cohen for his invaluable assistance in writing this article. Your insights and support were crucial in navigating the vast universe of Python's concurrent processing. Thank you, Adar!

What will we find in this article?

Python concurrency
Parallel processing
Efficient Python scripting
Data science optimization
Python multiprocessing

The Slow March Across the Data Universe

Visualize your Python code as a lone rover on the vast plains of Mars, tasked with a mission to analyze a colossal dataset:

all_data = load_massive_dataset()

for item in all_data:
   result = super_cpu_intensive_work(item)
   save_result(result)

This scenario screams inefficiency. super_cpu_intensive_work consumes all CPU resources, while save_result waits idly. This is a classic example of sequential processing at its most tedious.

And the memory usage? It's as barren and uneventful as the Martian landscape:

It's time for a revolutionary leap in data processing!

Mission Control: Concurrent Futures

Now, let's inject some excitement with concurrent.futures:

from concurrent.futures import ProcessPoolExecutor  

with ProcessPoolExecutor() as executor:
   for item in all_data: 
      result = super_cpu_intensive_work(item)
      executor.submit(save_result, result)

This is where the magic happens! super_cpu_intensive_work and save_result now work in tandem, efficiently processing data like a team of synchronized satellites.

Behold the transformation in memory usage – it's as dynamic and bustling as a space station at peak hour:

Steering Through ProcessPoolExecutor

In the vast expanse of CPU-bound tasks, ProcessPoolExecutor is your trusted spacecraft:

Specialized in managing multiple processes, perfect for parallelizing CPU-intensive tasks.
Operates like separate spacecraft, each with its own Python command center.
Isolated processes ensure mission integrity, even if one encounters an issue.
Ideal for heavy-duty tasks that require significant computational power.

For tasks that are more about communication than brute force, ThreadPoolExecutor is your go-to option, leveraging threads for I/O-bound operations.

CPU vs GPU: Choosing Your Spacecraft

When to Harness GPU Power:

Parallel Numerical Computations: Ideal for tasks like matrix multiplications or operations on large data arrays, which are common in data science and machine learning.
Image and Video Processing: Essential for quick processing of visual data, a must in fields like computer vision and digital media.
Deep Learning: GPUs are indispensable for training neural networks and AI models, offering unparalleled processing speeds.
Graphical Simulations: In simulations and gaming, GPUs provide the necessary horsepower for rendering complex graphics.

When to Rely on CPU:

General Computing: From running operating systems to executing diverse software applications, CPUs are the backbone of general computing.
Sequential Processing: For tasks where operations must occur in a specific order, CPUs are more efficient.
Complex Logic: When your code involves intricate decision-making, CPUs offer the required sophistication.
Server-side Operations: Hosting web servers or applications where parallel processing isn't critical, CPUs are more reliable.

To Infinity and Beyond!

With concurrent.futures, we're no longer confined to the tedious world of sequential data processing. We've stepped into the era of high-speed, efficient Python computing, perfect for the demanding needs of modern data science and machine learning. Python programmers, it's time to strap on your rocket boosters and explore the limitless possibilities of multiprocessing and parallel computing! 🌌

Converting MongoDB ObjectId to Timestamp in Snowflake: A Friendly Guide

Shahar Polak — Fri, 24 Nov 2023 14:02:27 +0000

Introduction

Hey there, fellow developers! Today, let's unravel a nifty trick that often stumps many of us when working with MongoDB and Snowflake.

Picture this: You've got data in MongoDB, each record stamped with that unique ObjectId MongoDB is famous for. Now, you need to migrate this data into Snowflake, but wait – how do you make sense of those timestamps hidden inside those ObjectIds? Fear not!

In this post, I'll walk you through a simple yet effective solution to extract those timestamps and bring them into the light in Snowflake.

The Challenge Explained

MongoDB's ObjectId isn't just a random jumble of characters; it's a treasure trove of information. The first 8 characters are actually a timestamp, representing when the data was created.

The catch? It's in hexadecimal format. And here's where Snowflake throws us a curveball – it doesn't have a built-in function to convert hexadecimal to decimal. So, how do we crack this code? Let's find out!

Step-by-Step Solution

Step 1: The Goal

Objective: Extract the timestamp from MongoDB's ObjectId in Snowflake.
The Hurdle: Snowflake's SQL doesn't directly convert hex to decimal.

Step 2: Our Secret Weapon – A Custom JavaScript UDF

UDFs to the Rescue: In Snowflake, we can create User-Defined Functions (UDFs) to perform custom operations. We'll use JavaScript for our UDF.

UDF Creation Script: Here's how we set up our UDF.

CREATE OR REPLACE FUNCTION hex_to_dec(hex_str STRING)
RETURNS FLOAT
LANGUAGE JAVASCRIPT
AS $$
    return parseInt(HEX_STR, 16);
$$;

What's Happening Here?: This snippet tells Snowflake, "Hey, let's turn those hex values into something we can work with in decimal!"

Step 3: Putting Our UDF to Work

The Magic Query: Now, we use our UDF in a SQL query.

CREATE OR REPLACE TABLE MONGO_IMPORT_WITH_TIMESTAMP AS
SELECT
    *,
    TO_TIMESTAMP_NTZ(CAST(hex_to_dec(SUBSTR(_id, 1, 8)) AS INTEGER)) AS created_at
FROM
    MONGO_IMPORT;

Breaking It Down: This query is our secret sauce. It says, "Let's create a new table, MONGO_IMPORT_WITH_TIMESTAMP, where we'll turn those hex timestamps into human-friendly dates."

Step 4: Test and Celebrate

Always Test: Before we break out the confetti, let's test this with some sample data.
What to Expect: If all goes well, you'll have a shiny new table in Snowflake, with all your MongoDB timestamps now readable and ready to use.

Wrapping Up

In wrapping up, let's recall the words of W. Edwards Deming:

'Without data, you're just another person with an opinion.'

Our journey through converting MongoDB ObjectIds to readable timestamps in Snowflake is a testament to this truth. By unlocking the data hidden within these identifiers, we turn mere numbers into meaningful insights.

Helpful Resources

Check out Snowflake's UDF documentation for more cool tricks.
Stuck or want to share your success? Join the conversation in Snowflake forums.

Going async in python

Shahar Polak — Wed, 07 Dec 2022 09:04:16 +0000

Going async in python - the missing guide

TLDR;
Make your code run faster and scalable using async API calls in python. with a quick explanation and an example anyone can follow.

Many times we have to make API requests, and one of the most familiar repositories to do so is requests.

Unfortunately, the requests library is not equipped for performing asynchronous requests.
We can wrap async/await syntax around requests, but that will make the underlying requests no less synchronous.

Why should we make our code async?

There are several reasons why we might want our code to be async.
Async code is a programming paradigm that allows multiple tasks to be performed concurrently without blocking.
This can improve the performance, responsiveness, and scalability of an application.

Async code can be run on a single thread, allowing it to make better use of available resources and handle more concurrent requests without running out of resources.

Async code can also make the code easier to read and maintain by allowing complex operations to be broken down into smaller tasks.

So how do we turn our code to async?

In order to get async requests, we must address other tools that can provide it.

One of my go-to tools is asyncio and using aiohttp. (they have amazing docs )

Can we see some examples of sync turn into async code?

In the following section we will take a requests code and convert it to async code using aiohttp.

import requests

IP_GEOLOCATION_KEY="YOUR API KEY"

# requests example 
class GeolocationClient:
    def __init__(self, key=IP_GEOLOCATION_KEY):
        self.key = key

    def get_geolocation(self, ip: str):
        base = "https://api.ipgeolocation.io/ipgeo"
        params = dict(apiKey=self.key, ip=ip)
        data = requests.get(base, params=params)
        return data.json()

Now lets see how we can convert to async method.

To use an async library to fetch data in the GeolocationClient class, you would need to first import the async library and then use its methods to make the HTTP request and retrieve the response.

import aiohttp

IP_GEOLOCATION_KEY="YOUR API KEY"


class GeolocationClient:
    def __init__(self, key=IP_GEOLOCATION_KEY):
        self.key = key

    async def get_geolocation(self, ip: str):
        base = "https://api.ipgeolocation.io/ipgeo"
        params = dict(apiKey=self.key, ip=ip)
        async with aiohttp.ClientSession() as session:
            async with session.get(base, params=params) as response:
                data = await response.json()
                return data

Here, the aiohttp.ClientSession is used to create a session for making HTTP requests. Then, the session.get() method is used to make a GET request to the specified URL with the given query parameters.

The response.json() method is used to parse the response as JSON and return it.

Note that this code uses async/await syntax, which is only supported in Python 3.7 and above. If you are using an older version of Python, you would need to use a different syntax for async operations, so upgrade your version! :)

Here is a full example running the code


import aiohttp
import asyncio

IP_GEOLOCATION_KEY="YOUR API KEY"


class GeolocationClient:
    def __init__(self, key=IP_GEOLOCATION_KEY):
        self.key = key

    async def get_geolocation(self, ip: str):
        base = "https://api.ipgeolocation.io/ipgeo"
        params = dict(apiKey=self.key, ip=ip)
        async with aiohttp.ClientSession() as session:
            async with session.get(base, params=params) as response:
                data = await response.json()
                return data


async def main():
    client = GeolocationClient()
    ips = ["1.1.1.1", "8.8.8.8", "9.9.9.9"]

    # Start all the tasks concurrently
    tasks = [client.get_geolocation(ip) for ip in ips]
    results = await asyncio.gather(*tasks)

    # Print the results as they come in
    for result in results:
        print(result)


# Run the main function
asyncio.run(main())

In this code, the main function is an async function that creates a GeolocationClient and a list of IP addresses.
It then uses a list comprehension to create a list of tasks that call the get_geolocation method on each IP address.

The asyncio.gather function is used to start all of the tasks concurrently and wait for them to complete.

It returns a list of results, which are the JSON responses from the get_geolocation method. The results are then printed as they come in.

Finally, the asyncio.run function is used to run the main function.
This will start the asynchronous tasks and wait for them to complete before exiting the program.

Hope that helps, and as usual, I'd love to know what you think.

Keep calm and code on

Aim for inspiration; motivation will follow.

Shahar Polak — Mon, 05 Apr 2021 17:26:20 +0000

Every organization's role has its challenges, and as much as I wish to tackle them all, I'll go with the bite-sized chunks and share my two cents about motivation.

Motivating your team and keeping high morale is one of the fundamental challenges every manager needs to master.
Is it an easy task? No-no-no.

There are so many things that can go wrong in the process without considering the manager's own motivation.
I can not talk through the eyes of every manager in the world, but I can share some of the experience I had with managers over the years and my managerial style and its progress during the past two decades.

So before I begin, I wish to share a tiny piece of one of my first f%^&kups as a manager. Let's go back almost 20 years ago to 2003, to a time, information flows were sooo much different from today, and I was a young operation sergeant in the Israeli Defense Force.
With little knowledge of how to treat my platoon, I was sure that the sun was revolving around me with a sense of entitlement that was highly misplaced.
One of my female soldiers was underperforming horribly; instead of understanding why I decided that the best approach is to go hard on her without even hearing her side of the story.
Then the crying river burst, bawling her eyes out, and the 19 years old me didn't know what to do or say. Between the tears, she has managed to tell me that she just had an abortion.
My inner world had crumbled.
At that moment, I gave her a big hug and told her I'm here for whatever she needs, I know it can not undo the way I've behaved, but I was determined to ease her pain in whatever way possible.

That day will stay with me for decades to come.

1) Inspire your team

It's effortless to tell people what to do, but I highly doubt it if this is how people wanna live their life, always being told what to do.

Instead of telling your employees what to do, tell them what you believe. Your job is to inspire and guide them, but let them pave the way, and I promise you, you are to be rewarded with happy, loyal and enthusiastic employees.

You make decisions that impact the success or failure of your team. As a leader, you are not meant to be a martyr. With this in mind, think about how you want your team to feel about working with you. Are they thankful and energized? Do they respect you? Do they enjoy the process, and can they see themselves performing well in your environment?

Trust is built through communication, and your members will feel comfortable sharing information with you. Try asking the members of your team, "What's going well?" "What can I do to help?" "How can we improve?" Also, listen to your team.

2) Lead your team

In order to truly empower your employees, you have to be willing to create a common language within your teams, know what the mistakes are that you are ready to let them make, and trust them when they want to try something that challenges the status quo.

It's not all about producing endless worksheets. There are several important aspects to making work fun and inspirational. Make it authentic, offer challenging work, give your employees permission to do their best, and let them know that you are committed to helping them grow and succeed.

Building their confidence and the chance to rise to the challenge. People usually like to do the stuff that they are good at. True, each and every single one of us needs to do things that we unnecessarily love and have to eat some frogs every now and then.
But if you see that an employee talks about a subject passionately, know if you could let them do it a bit more often, you'd be surprised how much they will go above and beyond.

Besides, try to let the team know why you've made the choices you've made. You are not operating in a vacuum, and transparency works both ways. The last word can be yours, but let your team be part of the process of coming to those conclusions.

3) Reward your team

Many see the financial reward as the best reward, I would like to suggest another approach.
I do find acknowledgment as a reward that goes a more extended range than any financial compensation I've ever received. How come?

Whenever any team member does outstanding work, being an exceptional team player, or had a great idea, you should let them know, not only to them but also to the rest of the team/group/company.

At my previous company, we used to have happy-hour every Thursday, and we started this tradition from day one of the company when every week could have been our last.
It started with thanking for everything we have managed to do this week and be thankful for each other. After some time, it has become the most iconic ritual in the company, where we gave authentic phrases to each other and a place where we could share our gratitude.

Be aware that some people do not feel comfortable getting praise in public, so feel free to express this gratitude in private.

You do not have to be a manager to lead.
Many leaders do not have an official title, but people will follow them because they believe. they can communicate their visions clearly.

One of my favorite authors of all times, Simon Sinek, said, "People don't buy WHAT you do: they buy WHY you do it".
I find this sentence to be my own personal guideline.

When I've joined BreezoMeter, a company that helps fight air pollution, I've done it because my wife has asthma. This is my story, and this is where it meets me.
After six years, when I've decided that it is my time to move on, I've joined ActiveFence, a fantastic company that helps fight bad actors online and make the internet a safer place. As a father, this is where it met me - creating a safe environment for my kid and all the other kids out there in the world.

Even before I had a managerial role, I have shared my personal WHY, that what connected me to the people I work with, the belief that we are paving the way together to the exact cause.

4) Be present

This last piece of advice is for everyone, not only leaders and managers.

When my son was born almost 6 years ago, I found myself sitting with him, but I wasn't really there. While I was sitting with my son, I was on my phone, scrolling Facebook, looking at how other people spend time with their kids. I know what you are thinking, what a jerk, and yes, you are right.
After doing some introspective, I've realized how much injustice I've created. I have deprived my son and me of having the time of our lives and making memories together.

It seems that all we need to do to write everything off in our to-do lists is multitasking. Well... don't. In any conversation, just be there. It is not enough to place your phone in your pocket on silent; you need to turn off all those mental notifications.

People can sense when you are not present, so please give them your undivided attention.

Project Loo: the best time-wasting, amazing, over-engineered project of my life

Shahar Polak — Sun, 18 Oct 2020 17:51:05 +0000

At my previous company, BreezoMeter, we had a problem, but it wasn’t a software architecture one, a code plight, or a cultural issue.

The problem was... [cue Scooby-Doo sound effect - DAM DAM DAM] - the bathroom.

THE PROBLEM

Our offices were great back when we were 8 people, but over time we grew to over 40 employees in the same offices having to share only TWO BATHROOM STALLS.

Now, I know what you're thinking, those spoiled engineers probably expect a personal stall each; but I assure you, that was not the case, and to make matters worse, there was no direct line of sight from our desks to the bathroom doors, so we couldn’t tell when they were vacant.

We decided that we have to find a remedy to this situation, so that people may relieve themselves when they please or at least be notified whenever the bathroom is vacant.

Not every invention needs to change the world; sometimes, it’s enough to change one person’s world.
-- Shahar Polak

THE SOLUTION

Every couple of months we used to have "Lab Days" - mini hackathons in which we could try new technologies, fiddle with ideas, and do so in teams that we did not have too much interaction with in our day-to-day.

My fellow teammate, Simon, and I played with the idea of connecting those stalls to the internal network, as every stall should be.

We knew that time was of the essence; we had two days to come up with a plan, buy all the equipment we needed, install the hardware, write the software, and make sure everything was working as expected.

THE ARCHITECTURE

After thinking it over, we decided to go with Google Firestore’s real-time database for the bathroom doors’ state.

We created a Vue.js web application to present users with the doors' status and connected to Firestore.

We added Cloud Functions so we could integrate it with Slack as well.

And lastly, we created workers so users would be able to subscribe and get a notification whenever the status changed.

Did I mention that we had two days to pull this off? Ambitious, I know.

THE PROCESS

On the day of the Lab Day, we both arrived at the office at 7:00 after a weekend of excitement and anticipation.

We decided to go with Raspberry Pi 3 with a microswitch as we thought it would be our best bet.

The microswitch is an on/off clip that connects to the Raspberry Pi 3 and can send a boolean signal.

Simon took the drill and started making holes so we could run the wires to the restroom. Not the most elegant way, but hey, it worked.

We installed Raspbian OS and Python 3 on the Raspberry Pi and then the coding began.

We had to build a small program that would sample the microswitch every X seconds and, once the status changed, update the database.

The first problem we encountered was false positives. The microswitch that we installed on the door was extremely sensitive, so every couple of seconds, even if someone only touched the door, it would send through a signal. We decided to change strategy and update our server only after 5 times that the microswitch sent the same signal.

The Python program would check every second what the microswitch's status is; once it detects the same signal 5 times and the status is different from the current one, it would update the database's state.

A small code sample, just to explain the Raspberry Pi part without too many tedious details:

class Queue:
    def __init__(self, limit):
        self._limit = limit
        self._items = []

    def is_empty(self):
        return self._items == []

    def append(self, value):
        should_remove_first = (self.size() >= self._limit) and not self.is_empty()
        if should_remove_first:
            self._items.pop(0)
        self._items.append(value)

    def size(self):
        return len(self._items)

    def all_same(self):
        if self.is_empty():
            return False
        return all(elem == self._items[0] for elem in self._items)

    @property
    def items(self):
        return self._items

def get_door_status():
    # reads the data from the Raspberry pi micro-switch


def update_door_server_status(door_id, status):
    # update firebase with the new status

def main():
    status = False
    q = Queue(5)
    door_id = 1 
    while True:
        time.sleep(1)
        door_status = get_door_status()
        q.append(door_status)
        all_same = q.all_same()
        if not all_same:
            continue
        if status == door_status:
            continue
        update_door_server_status(door_id, door_status)
        status = door_status

Now that we had code that updates the database, we had to create a web client, so users could access it.

We decided to go with a lean application using Vue.js hosted on Firebase Hosting - why you ask? Well, it's free.

We hustled our way into our designer’s heart and convinced him that designing the restroom's web pages is probably the best use of his time. The designs came out great!

THE FINAL RESULTS

This is the final result:

We even had a favicon that changed according to the status, so people could collapse the browser tab and see by the icon if the restrooms are available or not.

We added a button to subscribe and get a notification the second that the restrooms are vacant.

CONCLUSION

Not every application is meant to change the world. This is one of those applications that did one thing and did it well, and it was a heck of a lot of fun building it.

So after building loo.cloud (yes, we actually bought this domain, but we do not use it anymore), we pondered these questions.
Was it perfect? Far from it!
Was it over-engineered? Like 10 levels too much for the task at hand.
Was it awesome? Sure as hell was.

Building fun stuff is part of what makes being a software engineer so much fun.

Keep Calm and Cache On

Shahar Polak — Sat, 10 Oct 2020 13:05:28 +0000

I had a great conversation with a friend about premature optimizations.
One thing led to another, and we have started talking about caching and Memoization.

Each of us had a very different perspective on the matter, but the one thing we both agreed upon is the importance of performance.

He asked me if I could explain my thoughts in a layman's terms, and as Barney Stinson used to say, challenge accepted!

So before we begin, let's talk about what Memoization is and why we even need it.

What is Memoization?

Memoization is an optimization technique used primarily to prevent the re-calculation of the save results for the same output.
Basically, it means that our software will run faster.

Why should we use Memoization?

We should use Memoization for better performance and faster results.
For example, if we use any client-side JavaScript code, we are less likely to choke the main thread and have laggy UI, and nobody likes that ¯\(ツ)/¯.

ENOUGH TALKING! LET ME SEE THE CODE!

You are right; I know that I would like to see some action before I keep reading.

Let's say that we have a simple function "add"; add takes two numbers and return the value of the bough of them;

const add = (a, b) => {
  return a + b;
};

In this function, we re-evaluate a+b every single time it is called.
This is not an "expensive" calculation. Therefore, we would unlikely to use Memoization for something like that, but we could do something like that if we would.

const cachedAdd = memoizer(add);

cachedAdd(2,3); // 5 Not Cached
cachedAdd(2,3); // 5 Cached
cachedAdd(2,3); // 5 Cached

That's all nice and well, but how the heck does "memoizer" works?

Let's see if we can create a simple generic "memoizer" high-order function that we can reuse.

/**
 * Cache function results for given params
 *
 * @param {function} func
 * @returns {function(): (*)}
 */
function memoizer(func) {
  const cache = {};
  return function() {
    const key = JSON.stringify(arguments);
    if (cache[key] !== undefined) {
      return cache[key];
    }
    const result = func(...arguments);
    cache[key] = result;
    return result;
  };
}

There are many ways to go about writing this function, but let's go over this implementation step by step.
The "memoizer" takes a function, it uses the arguments object, and stringify it to create the key.
Once it has the key, the function checks to see if the key is available in the cache object; if it does, it returns the cached result, and we are done.
In case it does not, it will calculate the value, save it in the cache, and then return it.

Memoization exchanges time for space.

I know what you think, "I am not convinced that it worth the hassle."

Show me the money

Let's see some runtime results.
To see the following, I'll use the notorious Fibonacci Sequence function.

The Fibonacci Sequence is the series of numbers:
0, 1, 1, 2, 3, 5, 8, 13, 21, 34, ...
The next number is found by adding up the two numbers before it;

We could implement such a function like so:

const fibonacci = n => {
  if (n <= 1) return 1;
  return fibonacci(n - 1) + fibonacci(n - 2);
};

const getFibonacci = (limit = 1) => {
   const arr = [];
   for (let i = 0; i <= limit; i++) {
      arr.push(fibonacci(i));
   }
   return arr;
};

We can call the function like so:

getFibonacci(30); // will result [ 1, 1, 2, 3, 5, 8, 13, 21, 34, 55, 89 ...]

Lets will run a benchmark test when the limit is 30:

console.time("fibonacci");
for (let i = 0; i <= 100; i++) {
   getCachedFibonacci(30);
}
console.timeEnd("fibonacci");

The first time we run it, it will result in 193.097ms;
The problem is that in case we will run this code 100 times, it will not get any better and might just get worst.
For example, this code ran 100 times in a total of 18357.116ms, which is shit tones.

Let's see if we could do better?
We will use the Memoization function that we wrote earlier to create a new cached Fibonacci function:

const cachedFibonacci = memoizer(fibonacci);

const getCachedFibonacci = (limit = 1) => {
  const arr = [];
  for (let i = 0; i <= limit; i++) {
    arr.push(cachedFibonacci(i));
  }
  return arr;
};
console.time("cachedFibonacci");
for (let i = 0; i <= 100; i++) {
   getCachedFibonacci(30);
}
console.timeEnd("cachedFibonacci");

This time around, we will get other results.
The first time we run it, it will result like before, and take around 193.509ms to resolve, but from the second time and beyond, the function resulted in an average of 0.027ms;
To a total of 199.988ms for the 100 iterations.

👑 That results is 7,000~ times faster for each iteration.

Now, I know what you are thinking; not every problem is a Fibonacci one;
I can not stress it enough, Memoization is not a silver bullet, and it is not suitable for every scenario.
On the other hand, it is another powerful tool that can help your application performance when using correctly.

Should I create my own Memoization function?

Of course, you can do it, but in case you wish to use one of the open-source, well tested, well-documented Memoization function, here is a short list:

If you have any questions or thoughts on the matter, I would love to hear them, and in the meantime, Keep Calm 👑 Cache On.

Full Stack - Where to Start

Shahar Polak — Mon, 14 Sep 2020 16:49:57 +0000

Last week I got a phone call from a friend of a friend. They told me that they’re in their final semester of CS, and want to become a full-stack engineer.
I thought to myself, well, all you need now is years of experience in backend development, front-end, DBA, DevOps, logs, and so on (you get the point).

I didn’t want to discourage him, and if there’s one thing that I value, it’s enthusiasm and willingness to learn. So I told him to start where I tell all my students to start - at the beginning.
Fundamentals are the most valuable part at this stage - acquiring good habits, understanding principles, and using them. I also told him that some of my bad habits took me months and even years to unlearn (while some are still with me to this day).

So, he asked me, "Where do I start?"

You start with the fundamentals. If you want to become a web developer, you should learn HTML, CSS, and Javascript before diving deep into any framework.

This list was meant for him, but it will be a shame not to share it here as well.

So, where to start?

Wes Bos, Javascript 30 (free)

An incredible 30 exercise challenge, with explanations of how Wes Bos goes on about each of them.
This one is also highly recommended for veteran developers who wish to sharpen their Vanilla JavaScript skills.

Team Treehouse online school ($14.99/m)

One of the best online video schools. Funny, with great teachers.

What I love the most about Team Treehouse is the "tracks." They really help you hit the ground running by creating a track that lets you know what to study next.

Personally, when I have too many options to choose from, I sometimes get lost trying to study everything.

Frontend Masters online school ($39/m)

An outstanding online school that, like Team Treehouse’s "tracks," has "learning paths."

They have managed to assemble some of the most well-known names in the front-end industry to deliver their courses, such as:

Kyle Simpson, the author of “You Don't Know JavaScript.”
Steve Kinney from Twillo.
Sarah Drasner from Netlify/Microsoft, and many more.

Pluralsight online school ($29)

One school that had to make it to this list.
Pluralsight is an online video school with thousands of courses in many languages and technologies.
With that said, I do not know if this will be the first place to learn the basics; I think that Pluralsight is better a little bit down the road.

📚 Books

You don't know JavaScript by Kyle Simpson - online books(free)

An online series that deep dives into JavaScript, and is highly recommended for learning concepts and truly understanding JavaScript fundamentals.

The coding carrier handbook - ebook(39$)

AKA The Missing Manual for Early Career Devs: Guides, Principles, Strategies, and Tactics.
This book is a must to any developer, and I often heard developers saying it is a book that they wish they read years ago.

The Pragmatic Programmer (2020 Edition) By David Thomas & Andrew Hunt - audible($15), paperback($44)

This book has literally changed my life as a programmer.
It gave me insights and tough me not to work on autopilot.

Summary

Steve Jobs put it best - "Stay hungry. Stay foolish."

Hope this helps, and please let me know what you think or what should be added to this list.

Full disclosure - I am not associated or affiliated with any of the schools above and do not get anything other than the satisfaction of knowing I’ve managed to help even a single person.

Forem: Shahar Polak

FastAPI, Furious Tests: The Need for Speed

TL;DR - The "I Just Want Results" Section

Who Are You? (And Why Should You Care?)

The Problem: When Tests Become Slower Than Dial-Up Internet

The Real Cost

What We're Optimizing For

The Hall of Failed Attempts (Learn From Our Pain)

Attempt #1: Dockerized MySQL for Everything

Attempt #2: Shared Test Database

Attempt #3: Self-Hosted AWS Runners

The Solution: SQLite in-Memory (The Chosen One)

The SQLite Advantage

But Wait, There's a Catch

The 10 Commandments of SQLite Testing

The Implementation: Code That Actually Works

Project Structure (The Foundation)

The Database Setup (app/db.py)

The FastAPI App (app/main.py)

The Test Configuration (pytest.ini)

The Fixture Magic (tests/conftest.py)

Test Data Factories (tests/factories.py)

Sample Tests That Actually Work

The Parallelization: pytest-xdist Wizardry

Basic Parallelization

Why --dist worksteal?

Avoiding Parallel Test Hell

Ensuring Test Independence

Debugging Parallel Failures

The CI Configuration: GitHub Actions That Don't Suck

The Fast Lane (SQLite Tests)

The E2E Lane (MySQL Tests)

Before (GitHub Actions)

After (Blacksmith)

Smart Test Splitting with pytest-split

The Problem with Count-Based Splitting

The pytest-split Solution

Step 1: Generate Timing Data

Step 2: Use Runtime-Based Splitting in CI

When to Use pytest-split

Setting Up pytest-split for Your Project

pytest-split Best Practices

Memory Management and Cleanup

Memory Guidelines

Monitoring Memory Usage

SQLite File Cleanup

Measurement Protocol (No More Hand-Wavy Claims)

Hardware Baseline

The 5-Run Protocol

Results Table Format

Common Pitfalls (Learn From Others' Mistakes)

The "It Works on My Machine" Special

The "Shared State Surprise"

The "Foreign Key Forgotten"

The "Performance Regression Creep"

The "False Confidence Trap"

The Incremental Migration Plan

Phase 1: Measure Everything

Phase 2: Fix the Fixtures

Phase 3: Switch to SQLite

Phase 4: Enable Parallelization

Phase 5: Consider Runtime-Based Splitting (Optional)

Phase 6: Polish and Monitor

Troubleshooting Guide

"My tests are using :memory: but still failing in parallel"

"SQLite tests pass but MySQL tests fail"

"One worker is much slower than others"

"Tests are fast but flaky"

"Memory usage is through the roof"

The Results (Finally!)

Performance Improvements

What You Get

What You Don't Get

The Complete Checklist

Setup Phase

Fixtures Phase

Testing Phase

Parallelization Phase

CI Phase

Monitoring Phase

Why `--dist worksteal`?

Stopping Errors Before They Stop You: The Safe Assignment Operator (`?=`) and Handling Promises Gracefully

Understanding the Safe Assignment Operator (`?=`)

What is the `?=` Operator?

Rolling Out `safeAssign`

Limitations of `safeAssign`

Handling Asynchronous Operations with `safeAwait`

Introducing `safeAwait`

Variations of `safeAwait`

Before and After: Clean Code with `safeAssign` and `safeAwait`

Without `safeAssign`:

With `safeAssign`:

Without `safeAwait`:

With `safeAwait`: