Python Zero to Advanced #10: Type Hints, asyncio, Concurrency & Testing with pytest

You made it to the final part of the series. The topics here are what turn a working script into production-grade software: type hints catch bugs before you run the code, asyncio and other concurrency tools make I/O-heavy programs fast, and pytest gives you the confidence to refactor without breaking things.

Type hints

Type hints annotate what types a variable, parameter, or return value is expected to have. Python itself does not enforce them at runtime; they are read by your editor (autocomplete, warnings) and by static checkers such as mypy or pyright.

from typing import TypedDict, Callable, Literal

def greet(name: str, times: int = 1) -> str:
    return " ".join([f"Hi {name}!"] * times)

def find_user(user_id: int) -> dict[str, str] | None:
    users = {1: {"name": "Budi"}, 2: {"name": "Siti"}}
    return users.get(user_id)

class User(TypedDict):
    name: str
    email: str
    role: Literal["admin", "member"]

def apply_twice(func: Callable[[int], int], value: int) -> int:
    return func(func(value))

u: User = {"name": "Andi", "email": "andi@example.com", "role": "member"}
scores: list[int] = [80, 92]

print(greet("Budi", 2))
print(find_user(2), find_user(9))
print(u["role"], apply_twice(lambda x: x * 3, 2))
print(greet.__annotations__)

Output:

Hi Budi! Hi Budi!
{'name': 'Siti'} None
member 18
{'name': <class 'str'>, 'times': <class 'int'>, 'return': <class 'str'>}

Modern syntax (Python 3.10+) lets you write list[int], dict[str, int], and X | None without importing anything. Now let mypy catch a mistake for you. Given this file:

# app.py
def greet(name: str) -> str:
    return "Hi " + name

greet(42)

pip install mypy
mypy app.py

Output:

app.py:5: error: Argument 1 to "greet" has incompatible type "int"; expected "str"  [arg-type]
Found 1 error in 1 file (checked 1 source file)

Start by annotating function signatures in new code; you do not need to annotate every local variable, since checkers infer most of them.

Concurrency basics: threads, processes, async

Concurrency means making progress on several tasks at once. Which tool to use depends on what your program spends its time on. In CPython, the GIL (Global Interpreter Lock) allows only one thread to execute Python bytecode at a time, so threads speed up waiting, not computing.

Workload	Tool	Example
I/O-bound, many connections	asyncio	Calling 100 APIs, websockets
I/O-bound, blocking libraries	ThreadPoolExecutor	Downloading files with requests
CPU-bound	ProcessPoolExecutor	Image processing, heavy math

import time
from concurrent.futures import ThreadPoolExecutor

def download(name):
    time.sleep(0.5)              # simulate a slow network call
    return f"{name} done"

files = ["a.csv", "b.csv", "c.csv", "d.csv"]

start = time.perf_counter()
with ThreadPoolExecutor(max_workers=4) as pool:
    results = list(pool.map(download, files))
elapsed = time.perf_counter() - start

print(results)
print("faster than sequential:", elapsed < 2.0)

Output:

['a.csv done', 'b.csv done', 'c.csv done', 'd.csv done']
faster than sequential: True

Four half-second "downloads" finish in about 0.5 seconds instead of 2. pool.map returns results in the same order as the inputs. For CPU-heavy work, swap in ProcessPoolExecutor; it must be started from a script guarded by if __name__ == "__main__":.

from concurrent.futures import ProcessPoolExecutor

def count_primes(limit):
    count = 0
    for n in range(2, limit):
        if all(n % d for d in range(2, int(n ** 0.5) + 1)):
            count += 1
    return count

if __name__ == "__main__":          # required for multiprocessing
    limits = [10_000, 20_000, 30_000]
    with ProcessPoolExecutor() as pool:
        results = list(pool.map(count_primes, limits))
    print(dict(zip(limits, results)))

Output:

{10000: 1229, 20000: 2262, 30000: 3245}

asyncio

asyncio runs many tasks on a single thread using cooperative multitasking. A function defined with async def is a coroutine. Inside it, await pauses the coroutine while it waits for I/O, letting the event loop run other tasks. asyncio.gather() runs coroutines concurrently and returns their results in order.

import asyncio
import time

async def fetch_user(user_id: int, delay: float) -> dict:
    print(f"start {user_id}")
    await asyncio.sleep(delay)          # stands in for an HTTP call
    print(f"end {user_id}")
    return {"id": user_id, "name": f"user-{user_id}"}

async def main() -> None:
    start = time.perf_counter()
    users = await asyncio.gather(
        fetch_user(1, 0.3),
        fetch_user(2, 0.1),
        fetch_user(3, 0.2),
    )
    elapsed = time.perf_counter() - start
    print([u["name"] for u in users])
    print("took about 0.3s:", 0.25 < elapsed < 0.5)

asyncio.run(main())

Output:

start 1
start 2
start 3
end 2
end 3
end 1
['user-1', 'user-2', 'user-3']
took about 0.3s: True

All three tasks start immediately, finish in order of their delay, and the total time equals the slowest task, not the sum. Yet gather still returns the results in the order you passed them in.

Add timeouts so one slow call cannot hang everything:

import asyncio

async def slow_report() -> str:
    await asyncio.sleep(2)
    return "report"

async def main() -> None:
    try:
        result = await asyncio.wait_for(slow_report(), timeout=0.2)
        print(result)
    except asyncio.TimeoutError:
        print("report timed out, showing cached version")

asyncio.run(main())

Output:

report timed out, showing cached version

Golden rule: never call blocking functions such as time.sleep() or requests.get() inside a coroutine; they freeze the whole event loop. Use async libraries like httpx or aiohttp, or offload blocking work with await asyncio.to_thread(func).

Testing with pytest

pytest is the de-facto testing framework for Python. Tests are plain functions whose names start with test_, in files named test_*.py, using the ordinary assert statement. Here is a small module to test:

# cart.py
def calculate_total(items: list[tuple[str, int, int]], discount: float = 0.0) -> int:
    if not 0 <= discount <= 1:
        raise ValueError("discount must be between 0 and 1")
    subtotal = sum(price * qty for _, price, qty in items)
    return round(subtotal * (1 - discount))

And the tests, including parametrize to run one test with many inputs and pytest.raises to check errors:

# test_cart.py
import pytest
from cart import calculate_total

def test_empty_cart_is_zero():
    assert calculate_total([]) == 0

def test_total_without_discount():
    items = [("Coffee", 18000, 2), ("Cake", 25000, 1)]
    assert calculate_total(items) == 61000

@pytest.mark.parametrize("discount, expected", [
    (0.0, 100000),
    (0.1, 90000),
    (0.5, 50000),
])
def test_discounts(discount, expected):
    assert calculate_total([("Item", 50000, 2)], discount) == expected

def test_invalid_discount_raises():
    with pytest.raises(ValueError, match="between 0 and 1"):
        calculate_total([("Item", 1000, 1)], discount=1.5)

pip install pytest
pytest -q

Output:

......                                                                   [100%]
6 passed in 0.02s

Each dot is a passing test (the parametrized test counts as three). When an assertion fails, pytest shows the expected and actual values in detail. A few more features worth knowing: fixtures (functions decorated with @pytest.fixture that provide reusable setup, such as a temporary directory via the built-in tmp_path), pytest -k name to run selected tests, and pytest --cov (with the pytest-cov plugin) for coverage.

Tips and common mistakes

• Expecting type hints to validate data at runtime. They do not; use a checker in CI, or a library like Pydantic for runtime validation.
• Calling a coroutine without await returns a coroutine object and triggers a "was never awaited" warning.
• Using threads for CPU-bound work gives no speed-up because of the GIL. Use processes.
• Tests that depend on each other or on real external services are flaky. Keep tests isolated and mock the network.

Exercise

Write a typed async function check_all(urls) that "checks" each URL concurrently (simulate with asyncio.sleep) and returns a dict of URL to status. Then write a pytest test for a synchronous helper of your choice.

import asyncio

async def check(url: str) -> tuple[str, int]:
    await asyncio.sleep(0.1)
    status = 404 if url.endswith("/missing") else 200
    return url, status

async def check_all(urls: list[str]) -> dict[str, int]:
    results = await asyncio.gather(*(check(u) for u in urls))
    return dict(results)

urls = [
    "https://example.com",
    "https://example.com/blog",
    "https://example.com/missing",
]
for url, status in asyncio.run(check_all(urls)).items():
    print(status, url)

Output:

200 https://example.com
200 https://example.com/blog
404 https://example.com/missing

Summary and what's next

• Type hints plus mypy/pyright catch bugs early and document your intent.
• Use asyncio for many concurrent I/O tasks, threads for blocking I/O libraries, and processes for CPU-bound work.
• pytest makes tests simple: plain functions, plain assert, plus parametrize and fixtures.

Congratulations on finishing Python Zero to Advanced! From here, pick a direction and build a real project: a REST API with FastAPI, a data analysis notebook with pandas, or an automation script for your daily work. The best way to lock in everything from Parts 1 to 10 is to ship something. You can always return to Part 1 to review the fundamentals.