Forem: MEROLINE LIZLENT

Python asyncio

MEROLINE LIZLENT — Wed, 13 May 2026 12:55:08 +0000

One of the most potent standard library modules in Python is called "asyncio," but it's also one of the most misunderstood. When asyncio might work better, developers reach for threads or processes, or they use async/await everywhere without knowing what it truly does.

This article discusses the useful patterns you'll utilize on a daily basis after building a real mental model from the ground up.

The Core Problem: I/O Is Slow

Your CPU can execute billions of operations per second. A network request takes 50–500 milliseconds. Reading from disk takes milliseconds. If your code does this:

result1 = fetch_from_database(query1)  # wait 100ms
result2 = call_external_api(url)       # wait 200ms
result3 = read_large_file(path)        # wait 50ms
# Total: ~350ms

Your computers brain, the CPU is not doing anything for most of the time. It is just waiting around.

Threading is a way to make use of this time. It does this by doing things at the same time. But Threading has some problems. It needs memory it has to switch between tasks and it can have bugs. There is also something called the GIL.

asyncio does things differently. Of doing many things at the same time it does one thing at a time.. When the thing it is doing is waiting, asyncio switches, to something else. This way, asyncio only uses one thread. It works together with the tasks so it does not use up too many resources.

The Event Loop

The thing about asyncio is that it is about the event loop. This event loop is like a scheduler. Here is what it does:

It runs a coroutine until it gets to an await.
Then it sets up the I/O operation.
After that the event loop moves on, to another coroutine.
The event loop comes back when the I/O operation is finished.

import asyncio

async def main():
    print("Start")
    await asyncio.sleep(1)  # "I'm waiting — go run something else"
    print("Done")

asyncio.run(main())  # Create an event loop and run main()

asyncio.run() is the modern entry point (Python 3.7+). It creates an event loop, runs your coroutine, and cleans up.

Coroutines: `async def` and `await`

A coroutine is a kind of function that you define with async def.
When you call it it does not start running away. Instead it gives you a coroutine object. You have to use await or schedule it to make it run. The coroutine only starts when you do one of these things. You have to tell it to go it won't go on its own. It needs await to begin.

import asyncio

async def greet(name: str):
    print(f"Hello, {name}")
    await asyncio.sleep(0.5)
    print(f"Goodbye, {name}")

async def main():
    # This runs sequentially — each greet waits for the previous to finish
    await greet("Alice")
    await greet("Bob")

asyncio.run(main())

Hello, Alice
Goodbye, Alice
Hello, Bob
Goodbye, Bob

Sequential, not concurrent. To get concurrency, you need asyncio.gather() or tasks.

`asyncio.gather()`: Run Coroutines Concurrently

import asyncio

async def fetch_user(user_id: int):
    print(f"Fetching user {user_id}...")
    await asyncio.sleep(1)  # Simulate network call
    return {"id": user_id, "name": f"User {user_id}"}

async def main():
    # Run all three concurrently — total time ~1s, not ~3s
    results = await asyncio.gather(
        fetch_user(1),
        fetch_user(2),
        fetch_user(3),
    )
    print(results)

asyncio.run(main())

Fetching user 1...
Fetching user 2...
Fetching user 3...
[{'id': 1, ...}, {'id': 2, ...}, {'id': 3, ...}]

All three started immediately; all three finished in ~1 second total.

Handling Errors in gather()

results = await asyncio.gather(
    fetch_user(1),
    fetch_user(2),
    fetch_user(999),
    return_exceptions=True,  # Don't raise — return exceptions as values
)

for result in results:
    if isinstance(result, Exception):
        print(f"Error: {result}")
    else:
        print(result)

Tasks: Fire and Don't Wait

asyncio.create_task() schedules a coroutine to run concurrently without immediately awaiting it.

import asyncio

async def background_job(name: str, delay: float):
    await asyncio.sleep(delay)
    print(f"{name} completed after {delay}s")

async def main():
    # Schedule tasks without waiting for them
    task1 = asyncio.create_task(background_job("Job A", 2))
    task2 = asyncio.create_task(background_job("Job B", 1))

    print("Tasks started, doing other work...")
    await asyncio.sleep(0.5)
    print("Still doing other work...")

    # Wait for both tasks to finish
    await task1
    await task2
    print("All done")

asyncio.run(main())

Tasks started, doing other work...
Still doing other work...
Job B completed after 1s
Job A completed after 2s
All done

`asyncio.wait()` and `as_completed()`

More control over how you wait for multiple tasks:

import asyncio

async def slow_operation(n: int):
    await asyncio.sleep(n * 0.5)
    return n * 10

async def main():
    tasks = [asyncio.create_task(slow_operation(i)) for i in range(1, 5)]

    # Process results as they complete (not in order)
    for coro in asyncio.as_completed(tasks):
        result = await coro
        print(f"Got result: {result}")

asyncio.run(main())

Got result: 10   ← task 1 finished first
Got result: 20
Got result: 30
Got result: 40

Timeouts

import asyncio

async def slow_api_call():
    await asyncio.sleep(10)  # Takes forever
    return {"data": "..."}

async def main():
    try:
        # Raise TimeoutError if it takes more than 2 seconds
        result = await asyncio.wait_for(slow_api_call(), timeout=2.0)
    except asyncio.TimeoutError:
        print("Request timed out")

asyncio.run(main())

Python 3.11+ adds asyncio.timeout() as a context manager:

async def main():
    try:
        async with asyncio.timeout(2.0):
            result = await slow_api_call()
            other_result = await another_call()
    except asyncio.TimeoutError:
        print("Timed out")

Queues: Producer-Consumer Pattern

asyncio.Queue is perfect for worker pool patterns:

import asyncio
import random

async def producer(queue: asyncio.Queue, num_items: int):
    for i in range(num_items):
        await asyncio.sleep(random.uniform(0.1, 0.5))
        item = f"task_{i}"
        await queue.put(item)
        print(f"Produced: {item}")
    # Signal workers to stop
    for _ in range(3):  # one sentinel per worker
        await queue.put(None)

async def worker(name: str, queue: asyncio.Queue):
    while True:
        item = await queue.get()
        if item is None:
            break
        print(f"  {name} processing {item}")
        await asyncio.sleep(0.3)  # simulate work
        queue.task_done()
    print(f"  {name} shutting down")

async def main():
    queue = asyncio.Queue(maxsize=5)  # Buffer of 5 items

    # Start 3 workers
    workers = [
        asyncio.create_task(worker(f"Worker-{i}", queue))
        for i in range(3)
    ]

    await producer(queue, 10)
    await asyncio.gather(*workers)

asyncio.run(main())

Semaphores: Limit Concurrency

When you have 1000 URLs to fetch but don't want to hammer a server with 1000 simultaneous connections:

import asyncio
import httpx

async def fetch(client: httpx.AsyncClient, url: str, semaphore: asyncio.Semaphore):
    async with semaphore:  # Only N coroutines can enter at once
        response = await client.get(url)
        return response.status_code

async def main():
    urls = [f"https://httpbin.org/delay/1?n={i}" for i in range(20)]
    semaphore = asyncio.Semaphore(5)  # Max 5 concurrent requests

    async with httpx.AsyncClient() as client:
        tasks = [fetch(client, url, semaphore) for url in urls]
        results = await asyncio.gather(*tasks)

    print(results)

asyncio.run(main())

Locks and Synchronization

When multiple coroutines access shared state:

import asyncio

class Counter:
    def __init__(self):
        self.value = 0
        self._lock = asyncio.Lock()

    async def increment(self):
        async with self._lock:
            current = self.value
            await asyncio.sleep(0)  # Yield to event loop (simulate real work)
            self.value = current + 1

async def main():
    counter = Counter()
    tasks = [asyncio.create_task(counter.increment()) for _ in range(100)]
    await asyncio.gather(*tasks)
    print(f"Final count: {counter.value}")  # 100, not some race-condition number

asyncio.run(main())

Other synchronization primitives:

asyncio.Event — signal between coroutines
asyncio.Condition — wait for a condition to be true
asyncio.Semaphore — limit concurrent access

Running Blocking Code in asyncio

Sometimes you need to call a blocking function (CPU-intensive or a non-async library) without blocking the event loop:

import asyncio
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

# For blocking I/O (database drivers without async support, etc.)
async def read_file_async(path: str) -> str:
    loop = asyncio.get_event_loop()
    with ThreadPoolExecutor() as pool:
        content = await loop.run_in_executor(pool, open(path).read)
    return content

# For CPU-heavy work (data processing, image resizing, etc.)
def cpu_intensive_task(data: list) -> int:
    return sum(x ** 2 for x in data)

async def main():
    loop = asyncio.get_event_loop()
    data = list(range(1_000_000))

    with ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, cpu_intensive_task, data)
    print(result)

asyncio.run(main())

Async Context Managers and Iterators

import asyncio

# Async context manager
class AsyncDatabase:
    async def __aenter__(self):
        print("Connecting to DB...")
        await asyncio.sleep(0.1)
        return self

    async def __aexit__(self, *args):
        print("Closing DB connection")
        await asyncio.sleep(0.05)

    async def fetch(self, query: str):
        await asyncio.sleep(0.1)
        return [{"id": 1}, {"id": 2}]

# Async generator (async iterator)
async def paginate(db, query: str, page_size: int = 10):
    page = 0
    while True:
        results = await db.fetch(f"{query} LIMIT {page_size} OFFSET {page * page_size}")
        if not results:
            break
        for row in results:
            yield row
        page += 1

async def main():
    async with AsyncDatabase() as db:
        async for row in paginate(db, "SELECT * FROM users"):
            print(row)

asyncio.run(main())

Common Mistakes

Calling a coroutine without awaiting it

async def main():
    fetch_data()  # Does nothing! Returns a coroutine object silently.
    await fetch_data()  # Correct

Using `time.sleep()` instead of `asyncio.sleep()`

import time

async def bad():
    time.sleep(2)  # Blocks the ENTIRE event loop for 2 seconds

async def good():
    await asyncio.sleep(2)  # Yields control — other coroutines run

Forgetting `asyncio.run()` or running in wrong context

# This does nothing at the top level
async def main():
    ...

main()  # Just creates a coroutine object

# Correct
asyncio.run(main())

CPU-bound work on the event loop

async def bad():
    result = sum(x**2 for x in range(10_000_000))  # Blocks event loop
    return result

async def good():
    loop = asyncio.get_event_loop()
    with ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, lambda: sum(x**2 for x in range(10_000_000)))
    return result

When to Use asyncio (and When Not To)

Use asyncio when:

Making many concurrent network requests
Building web servers or APIs (FastAPI, Starlette, aiohttp)
Working with WebSockets or streaming data
I/O-heavy scripts (database queries, file operations, API calls)

Don't use asyncio when:

Your code is CPU-bound — use multiprocessing instead
Your entire stack uses synchronous libraries that don't support async
The script is simple and sequential — it just adds boilerplate

Wrapping Up

asyncio is not about running things faster on a single CPU — it's about keeping your program productive while waiting for I/O. The mental model is simple: one thread, cooperative switching at every await.

Master gather(), create_task(), Semaphore, and Queue, and you'll have the tools to handle virtually any concurrent I/O workload Python throws at you.

Python's `ast`

MEROLINE LIZLENT — Wed, 13 May 2026 12:34:19 +0000

The ability for Python to parse itself is a capability that most developers never utilize. You can read Python source code as a structured tree of nodes rather than as text by using the ast (Abstract Syntax Tree) module. You can then examine, evaluate, or modify the code programmatically.

This is how code analysis tools, formatters, linters, and type checkers are constructed. Let's see how it functions.

What Is an Abstract Syntax Tree?

When Python runs your code, the first thing it does is parse it into an AST, which is a tree structure that represents each statement, expression, operator and value. The representation that lies in between raw source text and bytecode is called the AST.

import ast

source = "x = 1 + 2"
tree = ast.parse(source)
print(ast.dump(tree, indent=2))

Module(
  body=[
    Assign(
      targets=[
        Name(id='x', ctx=Store())],
      value=BinOp(
        left=Constant(value=1),
        op=Add(),
        right=Constant(value=2)))],
  type_ignores=[])

Every component of that expression, including the variable name, assignment, operator, and constants, is a separate node in the tree. With 'ast', you may stroll around the tree and do whatever you want with it.

Parsing Code

import ast

# Parse a string of source code
tree = ast.parse("x = 42")

# Parse an expression (not a full statement)
expr_tree = ast.parse("1 + 2 * 3", mode="eval")

# Parse from a file
with open("mymodule.py") as f:
    tree = ast.parse(f.read(), filename="mymodule.py")

mode options:

"exec" (default) — full module / multiple statements
"eval" — single expression
"single" — single interactive statement

Inspecting Nodes

Every node has a _fields attribute listing its children:

import ast

tree = ast.parse("result = foo(x, y=10)")

# Walk every node
for node in ast.walk(tree):
    print(type(node).__name__, end=" ")

Module Assign Name Call Name Name keyword Constant

Getting Line Numbers

import ast

source = """
def greet(name):
    return f"Hello, {name}"

greet("Alice")
"""

tree = ast.parse(source)
for node in ast.walk(tree):
    if hasattr(node, "lineno"):
        print(f"Line {node.lineno}: {type(node).__name__}")

NodeVisitor

ast.NodeVisitor gives you a clean visitor pattern for traversing the AST. You define visit_<NodeType> methods and they're called automatically.

Example: Find All Function Definitions

import ast

class FunctionFinder(ast.NodeVisitor):
    def __init__(self):
        self.functions = []

    def visit_FunctionDef(self, node):
        self.functions.append({
            "name": node.name,
            "line": node.lineno,
            "args": [arg.arg for arg in node.args.args],
            "is_async": False,
        })
        self.generic_visit(node)  # Continue walking into nested functions

    def visit_AsyncFunctionDef(self, node):
        self.functions.append({
            "name": node.name,
            "line": node.lineno,
            "args": [arg.arg for arg in node.args.args],
            "is_async": True,
        })
        self.generic_visit(node)

source = """
def login(username, password):
    pass

async def fetch_data(url, timeout=30):
    pass

class UserService:
    def create_user(self, name, email):
        pass
"""

tree = ast.parse(source)
finder = FunctionFinder()
finder.visit(tree)

for fn in finder.functions:
    prefix = "async " if fn["is_async"] else ""
    print(f"Line {fn['line']}: {prefix}def {fn['name']}({', '.join(fn['args'])})")

Line 2: def login(username, password)
Line 5: async def fetch_data(url, timeout)
Line 9: def create_user(self, name, email)

Real-World Analysis Examples

Find All Imports

import ast

class ImportFinder(ast.NodeVisitor):
    def __init__(self):
        self.imports = []

    def visit_Import(self, node):
        for alias in node.names:
            self.imports.append(alias.name)

    def visit_ImportFrom(self, node):
        module = node.module or ""
        for alias in node.names:
            self.imports.append(f"{module}.{alias.name}")

with open("mymodule.py") as f:
    tree = ast.parse(f.read())

finder = ImportFinder()
finder.visit(tree)
print(sorted(set(finder.imports)))

Find All `print()` Calls

Useful for catching debug prints before committing:

import ast

class PrintFinder(ast.NodeVisitor):
    def visit_Call(self, node):
        if isinstance(node.func, ast.Name) and node.func.id == "print":
            print(f"Line {node.lineno}: print() call found")
        self.generic_visit(node)

with open("main.py") as f:
    tree = ast.parse(f.read())

PrintFinder().visit(tree)

Detect Mutable Default Arguments

A classic Python bug: def foo(items=[]). Let's detect it automatically:

import ast

class MutableDefaultFinder(ast.NodeVisitor):
    MUTABLE_TYPES = (ast.List, ast.Dict, ast.Set)

    def visit_FunctionDef(self, node):
        for default in node.args.defaults:
            if isinstance(default, self.MUTABLE_TYPES):
                print(
                    f"Line {node.lineno}: '{node.name}' has a mutable default argument: "
                    f"{ast.unparse(default)}"
                )
        self.generic_visit(node)

source = """
def append_item(item, items=[]):
    items.append(item)
    return items

def update_config(key, config={}):
    config[key] = True
    return config
"""

tree = ast.parse(source)
MutableDefaultFinder().visit(tree)

Line 2: 'append_item' has a mutable default argument: []
Line 6: 'update_config' has a mutable default argument: {}

Count Complexity

import ast

class ComplexityAnalyzer(ast.NodeVisitor):
    """Approximates cyclomatic complexity per function."""

    BRANCH_NODES = (ast.If, ast.For, ast.While, ast.ExceptHandler,
                    ast.With, ast.Assert, ast.comprehension)

    def visit_FunctionDef(self, node):
        complexity = 1 + sum(
            1 for child in ast.walk(node)
            if isinstance(child, self.BRANCH_NODES)
        )
        status = "✅" if complexity <= 5 else "⚠️" if complexity <= 10 else "❌"
        print(f"{status} Line {node.lineno}: {node.name}() — complexity: {complexity}")
        self.generic_visit(node)

with open("mymodule.py") as f:
    tree = ast.parse(f.read())

ComplexityAnalyzer().visit(tree)

`ast.unparse()`

ast.unparse(), which transforms an AST node back into source code, is included in Python 3.9+. When showing what a node represents, this is quite helpful.

import ast

source = "result = [x**2 for x in range(10) if x % 2 == 0]"
tree = ast.parse(source, mode="eval")

# Round-trip: source → AST → source
print(ast.unparse(tree))
# result = [x ** 2 for x in range(10) if x % 2 == 0]

NodeTransformer: Modifying the AST

ast.NodeTransformer is like NodeVisitor but lets you return modified nodes — effectively rewriting code.

Replace All `print()` with `logger.info()`

import ast

class PrintToLogger(ast.NodeTransformer):
    def visit_Call(self, node):
        self.generic_visit(node)  # Transform children first

        if isinstance(node.func, ast.Name) and node.func.id == "print":
            # Replace print(x) with logger.info(x)
            node.func = ast.Attribute(
                value=ast.Name(id="logger", ctx=ast.Load()),
                attr="info",
                ctx=ast.Load(),
            )
        return node

source = """
print("Starting process")
result = compute()
print(f"Done: {result}")
"""

tree = ast.parse(source)
new_tree = PrintToLogger().visit(tree)
ast.fix_missing_locations(new_tree)  # Fill in missing line numbers

print(ast.unparse(new_tree))

logger.info('Starting process')
result = compute()
logger.info(f'Done: {result}')

Safe Expression Evaluation: `ast.literal_eval()`

ast.literal_eval() safely evaluates Python literals without executing arbitrary code. It's the safe alternative to eval():

import ast

# Safely parse Python literals from strings
print(ast.literal_eval("42"))           # 42
print(ast.literal_eval("[1, 2, 3]"))    # [1, 2, 3]
print(ast.literal_eval("{'a': 1}"))    # {'a': 1}
print(ast.literal_eval("(True, None)")) # (True, None)

# Raises ValueError for anything that isn't a literal
ast.literal_eval("os.system('rm -rf /')")  # ❌ ValueError — safe!
ast.literal_eval("1 + 2")                  # ❌ ValueError — expressions not allowed

# Common use case: parse config values from env vars or config files
import os
raw = os.getenv("ALLOWED_HOSTS", '["localhost", "127.0.0.1"]')
allowed_hosts = ast.literal_eval(raw)
print(allowed_hosts)  # ['localhost', '127.0.0.1']

Building a Simple Linter

Let's put it all together — a basic linter that checks for common issues:

import ast
import sys
from pathlib import Path

class SimpleLinter(ast.NodeVisitor):
    def __init__(self, filename: str):
        self.filename = filename
        self.issues = []

    def warn(self, node, message: str):
        self.issues.append(f"{self.filename}:{node.lineno}: {message}")

    def visit_FunctionDef(self, node):
        # Check for mutable defaults
        for default in node.args.defaults:
            if isinstance(default, (ast.List, ast.Dict, ast.Set)):
                self.warn(node, f"Mutable default argument in '{node.name}': {ast.unparse(default)}")

        # Check for functions with too many arguments
        num_args = len(node.args.args)
        if num_args > 7:
            self.warn(node, f"'{node.name}' has {num_args} arguments (max recommended: 7)")

        # Check for missing docstring
        if not (node.body and isinstance(node.body[0], ast.Expr) and
                isinstance(node.body[0].value, ast.Constant)):
            self.warn(node, f"'{node.name}' is missing a docstring")

        self.generic_visit(node)

    def visit_Call(self, node):
        # Flag print() calls
        if isinstance(node.func, ast.Name) and node.func.id == "print":
            self.warn(node, "print() call found — use logging instead")
        self.generic_visit(node)

    def visit_Import(self, node):
        # Flag wildcard imports
        for alias in node.names:
            if alias.name == "*":
                self.warn(node, "Wildcard import found — be explicit")
        self.generic_visit(node)

    def visit_ImportFrom(self, node):
        for alias in node.names:
            if alias.name == "*":
                self.warn(node, f"Wildcard import from '{node.module}' — be explicit")
        self.generic_visit(node)


def lint_file(filepath: str):
    source = Path(filepath).read_text()
    tree = ast.parse(source, filename=filepath)
    linter = SimpleLinter(filepath)
    linter.visit(tree)
    return linter.issues


if __name__ == "__main__":
    for path in sys.argv[1:]:
        issues = lint_file(path)
        for issue in issues:
            print(issue)
    if not issues:
        print("No issues found ")

Run it:

python linter.py mymodule.py

mymodule.py:3: 'process' is missing a docstring
mymodule.py:12: Mutable default argument in 'append_items': []
mymodule.py:20: print() call found — use logging instead

Summary

A class of tools that most developers believe require in-depth compiler knowledge are unlocked by the ast module. In actuality, Python's AST is very strong, user-friendly, and well-documented for:

Static analysis is used to quantify complexity, enforce style, and detect problems.
Python code can be generated programmatically.
Rework code at the AST level.
Parsing literals without using eval() is known as "safe evaluation." ast is your starting point if you're creating developer tools, a code review bot, a custom linter, or anything else that requires a structural understanding of Python code.

Additional reading:Docs for the ast module:
docs.python.org/3/library/ast.html
Pretty-print AST nodes with astpretty
The concrete syntax tree library for lossless code transformation is called libcst.

Python's `shutil`

MEROLINE LIZLENT — Wed, 13 May 2026 12:08:52 +0000

When you want to manipulate files and directories in Python you typically grab os. However, if your goal isn't manipulating file paths, then it's time to use shutil (shell utilities). Copying, moving, archiving and disk usage are all done through a very high level API.

Let's go through everything it can do.

Why `shutil` Over `os`?

os provides you primitives : os.rename(), os.remove() os.mkdir(). These work, but they don't work with trees, they don't copy file meta data, they don't compress anything.

shutil works at a higher level - directories, archives, full file trees - and it comes in the standard library. No need to install.

import shutil

Copying Files

`shutil.copy()` — Content + Permissions

import shutil

# Copies file content and permissions (not metadata like timestamps)
shutil.copy("source.txt", "destination.txt")

# Can also copy into a directory
shutil.copy("report.pdf", "/home/user/documents/")

`shutil.copy2()` — Content + Full Metadata

# Copies content + permissions + timestamps + other metadata
# Use this when you want a true duplicate
shutil.copy2("source.txt", "backup.txt")

`shutil.copyfile()` — Content Only

# Copies only file content — no permissions, no metadata
shutil.copyfile("source.txt", "destination.txt")

`shutil.copyfileobj()` — File Object to File Object

# Useful when you already have open file handles
with open("source.txt", "rb") as src, open("dest.txt", "wb") as dst:
    shutil.copyfileobj(src, dst)

# Also works with HTTP responses or any file-like object
import urllib.request
with urllib.request.urlopen("https://example.com/file.csv") as response:
    with open("file.csv", "wb") as out:
        shutil.copyfileobj(response, out)

Copying Entire Directory Trees

# Copy an entire directory tree
shutil.copytree("project/", "project_backup/")

# The destination must NOT exist (it will be created)

With Filters

import shutil
import os

def ignore_cache(dir, contents):
    """Ignore __pycache__ and .pyc files"""
    return [
        item for item in contents
        if item == "__pycache__" or item.endswith(".pyc")
    ]

shutil.copytree("myproject/", "myproject_clean/", ignore=ignore_cache)

shutil.ignore_patterns() is a shortcut for common patterns:

shutil.copytree(
    "myproject/",
    "myproject_clean/",
    ignore=shutil.ignore_patterns("*.pyc", "__pycache__", "*.log", ".git")
)

Copy Into Existing Directory (Python 3.8+)

# dirs_exist_ok=True allows copying into an existing destination
shutil.copytree("source/", "destination/", dirs_exist_ok=True)

Moving Files and Directories

# Move a file
shutil.move("report.pdf", "/archive/2024/report.pdf")

# Move a directory (rename if on same filesystem, copy+delete otherwise)
shutil.move("old_project/", "archive/old_project/")

shutil.move() is smarter than os.rename() — it works across filesystems and handles directories.

Deleting Directory Trees

# Remove an entire directory tree (like `rm -rf`)
shutil.rmtree("old_project/")

# With error handling
def handle_error(func, path, exc_info):
    print(f"Error deleting {path}: {exc_info[1]}")

shutil.rmtree("old_project/", onerror=handle_error)

shutil.rmtree() is irreversible. There is no trash — it's gone. Always double-check the path.

Archiving: zip, tar, gzip, bz2

Creating Archives

# Create a zip archive
# shutil.make_archive(base_name, format, root_dir, base_dir)
shutil.make_archive(
    "project_backup",   # output filename (without extension)
    "zip",              # format: zip, tar, gztar, bztar, xztar
    ".",                # root directory
    "myproject"         # directory to archive
)
# Creates: project_backup.zip

# Create a gzipped tar
shutil.make_archive("project_backup", "gztar", ".", "myproject")
# Creates: project_backup.tar.gz

Extracting Archives

# Extract any supported archive format
shutil.unpack_archive("project_backup.zip", "extracted/")

# Auto-detects format from extension
shutil.unpack_archive("project_backup.tar.gz", "extracted/")

Supported Formats

# See what formats are available on your system
print(shutil.get_archive_formats())
# [('bztar', "bzip2'ed tar-file"), ('gztar', "gzip'ed tar-file"),
#  ('tar', 'uncompressed tar file'), ('xztar', "xz'ed tar-file"), ('zip', 'ZIP file')]

Disk Usage

# Check disk usage on a path
usage = shutil.disk_usage("/")
print(f"Total:  {usage.total / 1e9:.1f} GB")
print(f"Used:   {usage.used / 1e9:.1f} GB")
print(f"Free:   {usage.free / 1e9:.1f} GB")
print(f"Usage:  {usage.used / usage.total * 100:.1f}%")

Total:  500.1 GB
Used:   127.4 GB
Free:   372.7 GB
Usage:  25.5%

Finding Executables

# Like `which` in bash
python_path = shutil.which("python3")
print(python_path)  # /usr/bin/python3

git_path = shutil.which("git")
print(git_path)  # /usr/bin/git

# Returns None if not found
npm_path = shutil.which("npm")
if npm_path is None:
    print("npm is not installed")

Terminal Size

# Get the current terminal dimensions
size = shutil.get_terminal_size()
print(f"Columns: {size.columns}, Lines: {size.lines}")
# Columns: 220, Lines: 50

# With fallback for non-terminal environments
size = shutil.get_terminal_size(fallback=(80, 24))

Real-World Patterns

Timestamped Backups

import shutil
from datetime import datetime

def backup_directory(source: str, backup_root: str) -> str:
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    backup_name = f"{backup_root}/backup_{timestamp}"
    shutil.copytree(source, backup_name)
    print(f"Backup created: {backup_name}")
    return backup_name

backup_directory("myproject/", "~/backups")

Safe Directory Reset

import shutil
import os

def reset_directory(path: str):
    """Delete and recreate a directory cleanly."""
    if os.path.exists(path):
        shutil.rmtree(path)
    os.makedirs(path)

reset_directory("build/")
reset_directory("dist/")

Deployment Script

import shutil
import os

def deploy(source: str, destination: str):
    """Copy build artifacts to deployment directory, ignoring dev files."""
    if os.path.exists(destination):
        shutil.rmtree(destination)

    shutil.copytree(
        source,
        destination,
        ignore=shutil.ignore_patterns(
            "*.pyc", "__pycache__", "*.test.py",
            ".env", ".git", "tests/", "*.md"
        )
    )
    print(f"Deployed {source} → {destination}")

deploy("myapp/", "/var/www/myapp/")

Archive and Rotate Old Logs

import shutil
import os
from pathlib import Path

def archive_logs(log_dir: str, archive_dir: str, max_archives: int = 5):
    """Archive log directory and keep only the N most recent archives."""
    os.makedirs(archive_dir, exist_ok=True)

    # Create new archive
    archive_path = shutil.make_archive(
        base_name=os.path.join(archive_dir, "logs"),
        format="gztar",
        root_dir=os.path.dirname(log_dir),
        base_dir=os.path.basename(log_dir),
    )
    print(f"Archived: {archive_path}")

    # Rotate — keep only max_archives most recent
    archives = sorted(Path(archive_dir).glob("*.tar.gz"))
    for old_archive in archives[:-max_archives]:
        old_archive.unlink()
        print(f"Removed old archive: {old_archive}")

archive_logs("logs/", "log_archives/")

Quick Reference

Function	What It Does
`shutil.copy(src, dst)`	Copy file with permissions
`shutil.copy2(src, dst)`	Copy file with full metadata
`shutil.copyfile(src, dst)`	Copy file content only
`shutil.copyfileobj(fsrc, fdst)`	Copy between file objects
`shutil.copytree(src, dst)`	Copy entire directory tree
`shutil.move(src, dst)`	Move file or directory
`shutil.rmtree(path)`	Delete directory tree
`shutil.make_archive(name, fmt, ...)`	Create zip/tar archive
`shutil.unpack_archive(file, dst)`	Extract archive
`shutil.disk_usage(path)`	Get disk usage stats
`shutil.which(name)`	Find executable path
`shutil.get_terminal_size()`	Get terminal dimensions

Summary

shutil is one of those modules that solves dozens of problems that you might otherwise implement with subprocesses or third-party libraries or brittle hand-crafted loops. Copying files with metadata, moving files across filesystems, recursing through directories, handling archives - stdlib, all completely bulletproof.

GitPython

MEROLINE LIZLENT — Wed, 13 May 2026 11:27:35 +0000

Have you ever written a deployment script with subprocess.run(["git", "pull"]) and felt like there just must be a better way? There is! GitPython provides you with a real Python API for doing anything Git can do: clone, commit, create branches, diffs, read history, etc.

Installation

pip install gitpython

GitPython assumes you have git installed on your system. It uses shelling out to git behind the scenes, translating into Python objects as cleanly as possible.

Opening a Repository

from git import Repo, InvalidGitRepositoryError

# Open an existing repo
repo = Repo("/path/to/your/project")

# Open the repo at the current working directory
repo = Repo(".")

# Open from any subdirectory (search_parent_directories=True)
repo = Repo(".", search_parent_directories=True)

# Safe opening with error handling
try:
    repo = Repo("/some/path")
except InvalidGitRepositoryError:
    print("Not a git repository")

Repository Basics

from git import Repo

repo = Repo(".")

# Basic info
print(repo.working_dir)       # /home/user/myproject
print(repo.git_dir)           # /home/user/myproject/.git
print(repo.is_dirty())        # True if there are uncommitted changes
print(repo.untracked_files)   # ['new_file.py', 'notes.txt']
print(repo.active_branch.name)  # main

# Check if repo has any commits
print(repo.head.is_valid())   # False if repo is empty

Cloning a Repository

from git import Repo

# Clone a public repo
repo = Repo.clone_from(
    "https://github.com/user/myproject.git",
    "/tmp/myproject"
)

# Clone a specific branch
repo = Repo.clone_from(
    "https://github.com/user/myproject.git",
    "/tmp/myproject",
    branch="develop"
)

# Clone with depth (shallow clone — faster for CI)
repo = Repo.clone_from(
    "https://github.com/user/myproject.git",
    "/tmp/myproject",
    depth=1
)

# Clone with SSH
repo = Repo.clone_from(
    "git@github.com:user/myproject.git",
    "/tmp/myproject"
)

Staging and Committing

from git import Repo
import os

repo = Repo(".")

# Stage specific files
repo.index.add(["README.md", "src/main.py"])

# Stage all changes (like `git add .`)
repo.git.add(A=True)

# Or using the index directly
changed_files = [item.a_path for item in repo.index.diff(None)]
untracked = repo.untracked_files
repo.index.add(changed_files + untracked)

# Commit
commit = repo.index.commit(
    "feat: add user authentication module",
    author_date="2024-01-15T10:00:00",
    commit_date="2024-01-15T10:00:00",
)
print(f"Committed: {commit.hexsha[:7]} — {commit.message.strip()}")

Commit with Custom Author

from git import Repo, Actor

repo = Repo(".")

author = Actor("Alice", "alice@example.com")
committer = Actor("CI Bot", "ci@example.com")

repo.index.add(["deploy.yaml"])
repo.index.commit(
    "chore: update deployment config",
    author=author,
    committer=committer,
)

Branches

from git import Repo

repo = Repo(".")

# List all branches
for branch in repo.branches:
    print(branch.name)

# List remote branches
for ref in repo.remotes.origin.refs:
    print(ref.name)

# Create a new branch
new_branch = repo.create_head("feature/login")

# Switch to a branch (checkout)
new_branch.checkout()

# Or one-liner
repo.git.checkout("-b", "feature/signup")

# Delete a branch
repo.delete_head("feature/old-stuff", force=True)

# Check current branch
print(repo.active_branch.name)  # feature/login

Working with Remotes

from git import Repo

repo = Repo(".")

# List remotes
for remote in repo.remotes:
    print(f"{remote.name}: {remote.url}")

# Fetch
repo.remotes.origin.fetch()

# Pull
repo.remotes.origin.pull()

# Push
repo.remotes.origin.push()

# Push a specific branch
repo.remotes.origin.push(refspec="feature/login:feature/login")

# Add a new remote
repo.create_remote("upstream", "https://github.com/original/repo.git")

# Remove a remote
repo.delete_remote("upstream")

Reading Commit History

from git import Repo

repo = Repo(".")

# Iterate over commit history
for commit in repo.iter_commits("main", max_count=10):
    print(f"{commit.hexsha[:7]}  {commit.authored_datetime}  {commit.author.name}")
    print(f"  {commit.message.strip()}")
    print()

a3f1c22  2024-01-15 10:30:00  Alice
  feat: add payment gateway

7b2d891  2024-01-14 16:45:00  Bob
  fix: correct validation logic

Filter by Author, Path, or Date

from datetime import datetime

# Commits by a specific author
for commit in repo.iter_commits("main", author="Alice"):
    print(commit.message.strip())

# Commits touching a specific file
for commit in repo.iter_commits("main", paths="src/auth.py"):
    print(f"{commit.hexsha[:7]} {commit.message.strip()}")

# Commits since a date
since = datetime(2024, 1, 1)
for commit in repo.iter_commits("main"):
    if commit.authored_datetime.replace(tzinfo=None) < since:
        break
    print(commit.message.strip())

Diffs: What Changed?

from git import Repo

repo = Repo(".")

# Diff between working directory and index (unstaged changes)
for diff in repo.index.diff(None):
    print(f"Modified: {diff.a_path}")

# Diff between index and HEAD (staged changes)
for diff in repo.index.diff("HEAD"):
    print(f"Staged: {diff.a_path}")

# Diff between two commits
commits = list(repo.iter_commits("main", max_count=2))
diffs = commits[1].diff(commits[0])

for diff in diffs:
    print(f"Changed: {diff.a_path}")
    if diff.diff:
        print(diff.diff.decode("utf-8"))

Reading File Contents from Git

You can read file contents from any commit without touching the filesystem:

from git import Repo

repo = Repo(".")

# Read a file at HEAD
blob = repo.head.commit.tree["README.md"]
content = blob.data_stream.read().decode("utf-8")
print(content)

# Read from a specific commit
commit = repo.commit("a3f1c22")
blob = commit.tree["src/main.py"]
print(blob.data_stream.read().decode("utf-8"))

# Navigate into subdirectories
blob = repo.head.commit.tree["src"]["auth"]["jwt.py"]
print(blob.data_stream.read().decode("utf-8"))

Submodules

from git import Repo

repo = Repo(".")

# List submodules
for submodule in repo.submodules:
    print(f"{submodule.name}: {submodule.url}")

# Add a submodule
repo.create_submodule("mylib", "libs/mylib", url="https://github.com/user/mylib.git")

# Update all submodules
for submodule in repo.submodules:
    submodule.update(init=True)

Real-World Patterns

Auto-Commit Changed Files

from git import Repo, Actor
from datetime import datetime

def auto_commit(repo_path: str, message: str = None):
    repo = Repo(repo_path)

    if not repo.is_dirty(untracked_files=True):
        print("Nothing to commit")
        return

    # Stage everything
    repo.git.add(A=True)

    msg = message or f"auto: update {datetime.now().strftime('%Y-%m-%d %H:%M')}"
    commit = repo.index.commit(msg, author=Actor("AutoBot", "bot@example.com"))
    print(f"Committed: {commit.hexsha[:7]}")
    return commit

auto_commit(".", "chore: automated sync")

Generate a Changelog

from git import Repo
from collections import defaultdict

def generate_changelog(repo_path: str, from_tag: str, to_tag: str = "HEAD") -> str:
    repo = Repo(repo_path)
    commits = list(repo.iter_commits(f"{from_tag}..{to_tag}"))

    categories = defaultdict(list)
    for commit in commits:
        msg = commit.message.strip().split("\n")[0]
        if msg.startswith("feat"):
            categories["Features"].append(msg)
        elif msg.startswith("fix"):
            categories["Bug Fixes"].append(msg)
        elif msg.startswith("chore") or msg.startswith("ci"):
            categories["Maintenance"].append(msg)
        else:
            categories["Other"].append(msg)

    lines = [f"# Changelog: {from_tag} → {to_tag}\n"]
    for category, items in categories.items():
        lines.append(f"\n## {category}")
        for item in items:
            lines.append(f"- {item}")

    return "\n".join(lines)

print(generate_changelog(".", "v1.0.0", "v1.1.0"))

Find Who Last Modified a Line (git blame)

from git import Repo

def blame_file(repo_path: str, file_path: str):
    repo = Repo(repo_path)
    blame = repo.blame("HEAD", file_path)

    for commit, lines in blame:
        for line in lines:
            print(f"{commit.hexsha[:7]}  {commit.author.name:<20}  {line.decode('utf-8')}", end="")

blame_file(".", "src/auth.py")

Check if Branch Is Behind Remote

from git import Repo

def check_sync_status(repo_path: str):
    repo = Repo(repo_path)
    origin = repo.remotes.origin
    origin.fetch()

    branch = repo.active_branch
    tracking = branch.tracking_branch()

    if tracking is None:
        print("Branch has no remote tracking")
        return

    ahead = list(repo.iter_commits(f"{tracking}..{branch}"))
    behind = list(repo.iter_commits(f"{branch}..{tracking}"))

    print(f"Branch '{branch.name}':")
    print(f"  Ahead by:  {len(ahead)} commit(s)")
    print(f"  Behind by: {len(behind)} commit(s)")

check_sync_status(".")

Using the Raw Git Interface

For executing Git commands not exposed through GitPython's API, call repo.git with any git command you want to run:

from git import Repo

repo = Repo(".")

# Any git command as a method call
output = repo.git.log("--oneline", "-5")
print(output)

# git stash
repo.git.stash("save", "work in progress")
repo.git.stash("pop")

# git cherry-pick
repo.git.cherry_pick("a3f1c22")

# git rebase
repo.git.rebase("main")

Summarizing

GitPython elevates Git from a shell-scriptable command line tool to a fully fledged Python object. Writing deployment tools, code analysis apps, changelog generators or scripts to massage repository data? GitPython has you covered. No more cutting and parsing subprocess output.

Ideally suited to automation: anything you would normally do by hand, with a series of git commands, can be turned into a neat, testable Python function.

Pydantic v2: The Data Validation Library That Makes Python Feel Type-Safe

MEROLINE LIZLENT — Wed, 13 May 2026 10:44:41 +0000

Pydantic is a full data validation, serialization, and settings management library. And with v2 which has been rewritten in Rust, is blazingly fast. This article covers everything from the basics to the advanced patterns that will change how you write Python.

What Is Pydantic?

Pydantic lets you define data schemas using Python type hints and then validates data against those schemas at runtime. When validation fails, you get clear, structured error messages not cryptic KeyError or AttributeError exceptions buried in your business logic.

pip install pydantic
pip install "pydantic[email]"  # for EmailStr and other extras

The Basics: BaseModel

from pydantic import BaseModel

class User(BaseModel):
    id: int
    name: str
    email: str
    is_active: bool = True

# Valid data
user = User(id=1, name="Alice", email="alice@example.com")
print(user.id)        # 1
print(user.is_active) # True (default)

# Pydantic converts compatible types
user2 = User(id="42", name="Bob", email="bob@example.com")
print(type(user2.id))  # <class 'int'> — "42" was coerced to 42

Type coercion is one of Pydantic's most useful behaviors. It tries to convert compatible values rather than failing immediately.

Validation

from pydantic import BaseModel, ValidationError

class Product(BaseModel):
    name: str
    price: float
    quantity: int

try:
    p = Product(name="Widget", price="not_a_number", quantity=10)
except ValidationError as e:
    print(e.json())

Output:

[
  {
    "type": "float_parsing",
    "loc": ["price"],
    "msg": "Input should be a valid number, unable to parse string as a number",
    "input": "not_a_number",
    "url": "https://errors.pydantic.dev/2.x/v/float_parsing"
  }
]

Structured, machine-readable errors with field location, error type, and the offending input.

Field: Fine-Grained Validation

The Field function lets you add constraints, defaults, aliases, and metadata to individual fields.

from pydantic import BaseModel, Field
from typing import Optional

class Article(BaseModel):
    title: "str = Field(..., min_length=3, max_length=200)"
    slug: str = Field(..., pattern=r"^[a-z0-9-]+$")
    body: str = Field(..., min_length=10)
    views: int = Field(default=0, ge=0)
    rating: float = Field(default=5.0, ge=1.0, le=5.0)
    tags: list[str] = Field(default_factory=list, max_length=10)
    author_id: Optional[int] = Field(None, description="Foreign key to users table")

... (Ellipsis) means the field is required. Common constraints:

min_length / max_length — for strings and lists
ge / le / gt / lt — numeric bounds (≥, ≤, >, <)
pattern — regex validation for strings
default_factory — callable that produces the default (use this instead of default=[])

Custom Validators

Field Validators

from pydantic import BaseModel, field_validator

class UserRegistration(BaseModel):
    username: str
    password: str
    confirm_password: str

    @field_validator("username")
    @classmethod
    def username_must_be_alphanumeric(cls, v: str) -> str:
        if not v.isalnum():
            raise ValueError("Username must contain only letters and numbers")
        return v.lower()  # normalize to lowercase

    @field_validator("password")
    @classmethod
    def password_strength(cls, v: str) -> str:
        if len(v) < 8:
            raise ValueError("Password must be at least 8 characters")
        if not any(c.isupper() for c in v):
            raise ValueError("Password must contain at least one uppercase letter")
        if not any(c.isdigit() for c in v):
            raise ValueError("Password must contain at least one digit")
        return v

Model Validators (Cross-Field Validation)

Sometimes you need to validate multiple fields together. Use @model_validator:

from pydantic import BaseModel, model_validator

class PasswordReset(BaseModel):
    password: str
    confirm_password: str

    @model_validator(mode="after")
    def passwords_must_match(self) -> "PasswordReset":
        if self.password != self.confirm_password:
            raise ValueError("Passwords do not match")
        return self

class DateRange(BaseModel):
    start_date: date
    end_date: date

    @model_validator(mode="after")
    def end_must_be_after_start(self) -> "DateRange":
        if self.end_date <= self.start_date:
            raise ValueError("end_date must be after start_date")
        return self

Types That Do the Heavy Lifting

Pydantic ships with a rich set of built-in types that handle common validation patterns:

from pydantic import BaseModel, EmailStr, HttpUrl, AnyUrl
from pydantic import IPvAnyAddress, SecretStr, PositiveInt, NegativeFloat
from datetime import datetime
from uuid import UUID

class RichModel(BaseModel):
    # Validated email address
    email: EmailStr

    # Validated URLs
    website: HttpUrl
    profile_picture: AnyUrl

    # UUID (accepts strings and UUID objects)
    user_id: UUID

    # Hides value in repr and logs — great for passwords/tokens
    api_key: SecretStr

    # Type shortcuts
    age: PositiveInt          # int > 0
    temperature: NegativeFloat  # float < 0

    # IP address (v4 or v6)
    ip_address: IPvAnyAddress

    # Datetime parsing from strings, timestamps, etc.
    created_at: datetime

data = RichModel(
    email="user@example.com",
    website="https://example.com",
    profile_picture="s3://bucket/image.png",
    user_id="550e8400-e29b-41d4-a716-446655440000",
    api_key="super-secret-key",
    age=25,
    temperature=-3.5,
    ip_address="192.168.1.1",
    created_at="2024-01-15T10:30:00Z",
)

print(data.api_key)          # **********  (hidden!)
print(data.api_key.get_secret_value())  # super-secret-key

Nested Models and Composition

Pydantic models compose naturally:

from pydantic import BaseModel
from typing import Optional

class Address(BaseModel):
    street: str
    city: str
    country: str
    postal_code: Optional[str] = None

class Company(BaseModel):
    name: str
    address: Address

class Employee(BaseModel):
    name: str
    company: Company
    work_address: Optional[Address] = None

# Nested dict input works seamlessly
data = {
    "name": "Alice",
    "company": {
        "name": "Acme Corp",
        "address": {
            "street": "123 Main St",
            "city": "Nairobi",
            "country": "Kenya"
        }
    }
}

employee = Employee(**data)
print(employee.company.address.city)  # Nairobi

Validation cascades through all nested models automatically.

Generics: Reusable Response Wrappers

One of Pydantic's most useful patterns for APIs is a generic response wrapper:

from pydantic import BaseModel
from typing import Generic, TypeVar, Optional, List

T = TypeVar("T")

class PaginatedResponse(BaseModel, Generic[T]):
    items: List[T]
    total: int
    page: int
    per_page: int
    has_next: bool

class APIResponse(BaseModel, Generic[T]):
    success: bool
    data: Optional[T] = None
    error: Optional[str] = None

class UserOut(BaseModel):
    id: int
    name: str
    email: str

# Usage
response = APIResponse[UserOut](
    success=True,
    data=UserOut(id=1, name="Alice", email="alice@example.com")
)

paginated = PaginatedResponse[UserOut](
    items=[UserOut(id=1, name="Alice", email="alice@example.com")],
    total=1,
    page=1,
    per_page=10,
    has_next=False,
)

In FastAPI, these generics show up correctly in the OpenAPI schema too.

Serialization: model_dump and model_dump_json

from pydantic import BaseModel, Field
from datetime import datetime

class Event(BaseModel):
    event_id: int = Field(alias="eventId")
    title: str
    created_at: datetime

event = Event(eventId=1, title="Conference", created_at=datetime.now())

# Convert to dict
print(event.model_dump())
# {'event_id': 1, 'title': 'Conference', 'created_at': datetime(...)}

# Use the alias in output
print(event.model_dump(by_alias=True))
# {'eventId': 1, 'title': 'Conference', 'created_at': datetime(...)}

# Exclude fields
print(event.model_dump(exclude={"created_at"}))

# Include only specific fields
print(event.model_dump(include={"title", "event_id"}))

# Serialize to JSON string (fast Rust-based serializer)
print(event.model_dump_json())

Aliases: Bridging Different Naming Conventions

APIs often use camelCase while Python uses snake_case. Pydantic handles this elegantly:

from pydantic import BaseModel, Field
from pydantic.alias_generators import to_camel

class UserProfile(BaseModel):
    model_config = {"alias_generator": to_camel, "populate_by_name": True}

    first_name: str
    last_name: str
    phone_number: str

# Accept camelCase input (from a JS frontend)
profile = UserProfile(firstName="Alice", lastName="Smith", phoneNumber="+254712345678")
print(profile.first_name)  # Alice

# Output camelCase (for a JS frontend)
print(profile.model_dump(by_alias=True))
# {'firstName': 'Meroline', 'lastName': 'Lizlent', 'phoneNumber': '+254712345678'}

Model Config: Customizing Behavior

model_config controls how your model behaves:

from pydantic import BaseModel, ConfigDict

class StrictUser(BaseModel):
    model_config = ConfigDict(
        # Reject extra fields instead of ignoring them
        extra="forbid",

        # Don't coerce types — "42" won't become 42
        strict=True,

        # Validate default values too
        validate_default=True,

        # Allow population by field name even when alias is set
        populate_by_name=True,

        # Freeze the model (make it immutable/hashable)
        frozen=True,
    )

    id: int
    name: str

Common extra options:

"ignore" (default) — silently discard extra fields
"forbid" — raise a validation error if extra fields are present
"allow" — keep extra fields in the model

Pydantic Settings: Managing Configuration

One of Pydantic's killer features for production apps: pydantic-settings for environment variable management.

pip install pydantic-settings

from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        case_sensitive=False,
    )

    # Required — will raise error if not set
    database_url: str
    secret_key: str

    # Optional with defaults
    debug: bool = False
    allowed_hosts: list[str] = ["localhost"]
    max_connections: int = Field(default=10, ge=1, le=100)
    app_name: str = "My App"

# Loads from environment variables and .env file automatically
settings = Settings()
print(settings.database_url)

Your .env:

DATABASE_URL=postgresql://user:pass@localhost/mydb
SECRET_KEY=your-secret-key-here
DEBUG=true
MAX_CONNECTIONS=20

Use it with a singleton pattern in FastAPI:

from functools import lru_cache

@lru_cache
def get_settings() -> Settings:
    return Settings()

# In your FastAPI routes
from fastapi import Depends

@app.get("/info")
def app_info(settings: Settings = Depends(get_settings)):
    return {"app": settings.app_name, "debug": settings.debug}

Computed Fields

Need a field derived from other fields?

from pydantic import BaseModel, computed_field

class Rectangle(BaseModel):
    width: float
    height: float

    @computed_field
    @property
    def area(self) -> float:
        return self.width * self.height

    @computed_field
    @property
    def perimeter(self) -> float:
        return 2 * (self.width + self.height)

r = Rectangle(width=5.0, height=3.0)
print(r.area)       # 15.0
print(r.model_dump())  # includes 'area' and 'perimeter'

Discriminated Unions: Polymorphic Models

When you have multiple model types that share a field used to distinguish them:

from pydantic import BaseModel
from typing import Literal, Union, Annotated
from pydantic import Field

class CreditCardPayment(BaseModel):
    payment_type: Literal["credit_card"]
    card_number: str
    expiry: str
    cvv: str

class BankTransferPayment(BaseModel):
    payment_type: Literal["bank_transfer"]
    account_number: str
    routing_number: str

class MpesaPayment(BaseModel):
    payment_type: Literal["mpesa"]
    phone_number: str
    transaction_id: str

Payment = Annotated[
    Union[CreditCardPayment, BankTransferPayment, MpesaPayment],
    Field(discriminator="payment_type")
]

class Order(BaseModel):
    order_id: int
    amount: float
    payment: Payment

# Pydantic knows exactly which model to use based on payment_type
order = Order(
    order_id=1,
    amount=99.99,
    payment={"payment_type": "mpesa", "phone_number": "+254712345678", "transaction_id": "MPX123"}
)
print(type(order.payment))  # <class 'MpesaPayment'>

Performance: Pydantic v2 Is Fast

Pydantic v2 was rewritten in Rust and is dramatically faster than v1:

Operation	v1	v2
Model validation	baseline	~5-50x faster
JSON serialization	baseline	~10x faster
Schema generation	baseline	~2-5x faster

For high-throughput APIs handling thousands of requests per second, this matters.

You can also pre-compile validators for repeated use:

from pydantic import TypeAdapter

# Validate a list of items without defining a full model
adapter = TypeAdapter(list[int])
result = adapter.validate_python(["1", "2", "3"])
print(result)  # [1, 2, 3]

# Or validate JSON directly (fast path)
result = adapter.validate_json("[1, 2, 3]")

Common Pitfalls

Mutable defaults

# BAD — shared mutable default (classic Python bug)
class Model(BaseModel):
    tags: list = []  # all instances share the same list!

# GOOD
class Model(BaseModel):
    tags: list[str] = Field(default_factory=list)

Modifying a model directly

user = User(id=1, name="Alice")
user.name = "Bob"  # This works but bypasses validation in some contexts

# For truly immutable models:
class User(BaseModel):
    model_config = ConfigDict(frozen=True)
    id: int
    name: str

The dict returned by model_dump() contains the same objects. If you need a deep copy, use model.model_copy(deep=True).

Summary

Pydantic is one of those libraries that, once you learn it properly, changes how you think about data in Python. It brings structure, validation, and type safety to a language that's historically been loose about all three.

The patterns covered here; Field constraints, custom validators, generics, discriminated unions, Settings management are production patterns used in real systems.

JWT Authentication From Scratch: What It Is, How It Works, and How to Implement It Properly

MEROLINE LIZLENT — Wed, 13 May 2026 08:41:58 +0000

JSON Web Tokens are everywhere. They're the de facto standard for stateless API authentication and they're also one of the most misunderstood and misimplemented security mechanisms in web development.

This article covers JWT from first principles, including common mistakes that create security vulnerabilities, and walks through a complete, production-ready implementation in Python with FastAPI.

What Is a JWT?

A JSON Web Token (JWT) is a compact, URL-safe token that encodes a JSON payload and optionally signs or encrypts it. It's used to transmit verified claims between parties most commonly, "this user is authenticated and has these permissions."

A JWT looks like this:

eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJ1c2VyXzEyMyIsImV4cCI6MTcxNjY1MDAwMH0.3rW7TE8iCCiLBOBs6nLBpWxXuV3zqkL9mYpC2HBkd4s

It's three Base64URL-encoded segments joined by dots:

HEADER.PAYLOAD.SIGNATURE

Header

{
  "alg": "HS256",
  "typ": "JWT"
}

Specifies the algorithm used to sign the token.

Payload (Claims)

{
  "sub": "user_123",
  "name": "Alice",
  "role": "admin",
  "iat": 1716563600,
  "exp": 1716650000
}

Standard claims include:

sub — subject (typically a user ID)
iat — issued at
exp — expiry
nbf — not before
iss — issuer
aud — audience

The payload is NOT encrypted by default. It's only Base64-encoded, which is trivially reversible. Never store sensitive data (passwords, credit card numbers) in a JWT payload.

Signature

HMACSHA256(
  base64url(header) + "." + base64url(payload),
  secret_key
)

The signature is what makes JWTs trustworthy. The server verifies that nobody tampered with the payload by recomputing the signature and comparing it.

How JWT Authentication Works

1. User logs in with credentials (username + password)
2. Server validates credentials, issues a signed JWT
3. Client stores the JWT (usually in memory or localStorage)
4. Client sends JWT in the Authorization header on subsequent requests:
   Authorization: Bearer <token>
5. Server verifies the signature and extracts the claims
6. No database lookup required — the token is self-contained

This is what "stateless" authentication means: the server doesn't need to store session data. All the information is in the token itself.

JWT vs Sessions: When to Use Each

	JWT	Sessions
Storage	Client-side	Server-side
Scalability	Easy (no shared state)	Harder (needs shared session store)
Revocation	Hard (tokens are valid until expiry)	Easy (delete from store)
Size	Larger (header on every request)	Small (just a session ID)
Best for	Stateless APIs, microservices	Traditional web apps with logout

JWTs are great for APIs. For apps where you need instant revocation (logout that actually works), session-based auth or short-lived JWTs + refresh tokens is better.

Implementation: FastAPI + JWT

Dependencies

pip install fastapi uvicorn python-jose[cryptography] passlib[bcrypt] python-multipart

python-jose: JWT encoding/decoding
passlib: Password hashing with bcrypt

Project Structure

app/
├── main.py
├── auth.py        # JWT logic
├── models.py      # Pydantic models
└── database.py    # Fake DB for demo

`auth.py` — Core JWT Logic

from datetime import datetime, timedelta, timezone
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext

# Configuration — store these in environment variables in production!
SECRET_KEY = "your-super-secret-key-change-this-in-production"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
REFRESH_TOKEN_EXPIRE_DAYS = 7

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + (
        expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    to_encode.update({"exp": expire, "type": "access"})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def create_refresh_token(data: dict) -> str:
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
    to_encode.update({"exp": expire, "type": "refresh"})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def decode_token(token: str) -> dict:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError:
        return None

`models.py`

from pydantic import BaseModel
from typing import Optional

class UserLogin(BaseModel):
    username: str
    password: str

class Token(BaseModel):
    access_token: str
    refresh_token: str
    token_type: str = "bearer"

class TokenData(BaseModel):
    user_id: Optional[str] = None
    role: Optional[str] = None

`database.py` (Mock DB)

from auth import hash_password

# In production, replace with real database queries
FAKE_USERS_DB = {
    "alice": {
        "id": "user_001",
        "username": "alice",
        "hashed_password": hash_password("secret123"),
        "role": "admin",
        "is_active": True,
    },
    "bob": {
        "id": "user_002",
        "username": "bob",
        "hashed_password": hash_password("password456"),
        "role": "user",
        "is_active": True,
    },
}

def get_user(username: str):
    return FAKE_USERS_DB.get(username)

`main.py` — The Full Application

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from auth import (
    verify_password, create_access_token, create_refresh_token, decode_token
)
from database import get_user
from models import UserLogin, Token

app = FastAPI(title="JWT Auth Demo")
security = HTTPBearer()

# Authentication 
@app.post("/auth/login", response_model=Token)
def login(credentials: UserLogin):
    user = get_user(credentials.username)

    if not user or not verify_password(credentials.password, user["hashed_password"]):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )

    if not user["is_active"]:
        raise HTTPException(status_code=400, detail="Inactive user")

    token_data = {"sub": user["id"], "username": user["username"], "role": user["role"]}

    return Token(
        access_token=create_access_token(token_data),
        refresh_token=create_refresh_token(token_data),
    )

@app.post("/auth/refresh", response_model=Token)
def refresh_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
    payload = decode_token(credentials.credentials)

    if not payload or payload.get("type") != "refresh":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid or expired refresh token",
        )

    token_data = {
        "sub": payload["sub"],
        "username": payload["username"],
        "role": payload["role"],
    }

    return Token(
        access_token=create_access_token(token_data),
        refresh_token=create_refresh_token(token_data),
    )

# Dependency: Get Current User 

def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
    payload = decode_token(credentials.credentials)

    if not payload or payload.get("type") != "access":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Could not validate credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )

    user = get_user(payload.get("username"))
    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    return user

def require_admin(current_user: dict = Depends(get_current_user)):
    if current_user["role"] != "admin":
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Admin access required",
        )
    return current_user

# Protected Routes 

@app.get("/me")
def get_profile(current_user: dict = Depends(get_current_user)):
    return {
        "id": current_user["id"],
        "username": current_user["username"],
        "role": current_user["role"],
    }

@app.get("/admin/dashboard")
def admin_dashboard(admin: dict = Depends(require_admin)):
    return {"message": f"Welcome, {admin['username']}. Here's the admin dashboard."}

@app.get("/public")
def public_route():
    return {"message": "This route requires no authentication"}

Testing It

uvicorn main:app --reload

curl -X POST http://localhost:8000/auth/login \
  -H "Content-Type: application/json" \
  -d '{"username": "alice", "password": "secret123"}'

Use the access token:

curl http://localhost:8000/me \
  -H "Authorization: Bearer <your_access_token>"

Common Security Mistakes and How to Avoid Them

Using a weak or hardcoded secret key

# BAD
SECRET_KEY = "secret"

# GOOD — generate with: python -c "import secrets; print(secrets.token_hex(32))"
SECRET_KEY = os.environ["JWT_SECRET_KEY"]

Using the `none` algorithm

Some early JWT libraries had a bug where alg: none would skip signature verification. Always explicitly specify allowed algorithms:

jwt.decode(token, SECRET_KEY, algorithms=["HS256"])  # whitelist only

Storing JWTs in localStorage

localStorage is accessible to any JavaScript on the page, making it vulnerable to XSS attacks. Use httpOnly cookies for web apps, or keep tokens in memory for SPAs.

Long-lived access tokens without refresh tokens

If an access token is compromised, you want it to expire quickly. Use short-lived access tokens (15–30 minutes) and longer-lived refresh tokens (7–30 days).

Not validating the `exp` claim

The python-jose library validates expiry automatically when decoding, but make sure you're not using options={"verify_exp": False} in production.

Putting sensitive data in the payload

The payload is Base64-encoded, not encrypted. Decode any JWT at jwt.io to see this in action. Only store non-sensitive identifiers (user ID, role).

Token Revocation: The Hard Problem

JWTs are stateless by design - there's no built-in revocation mechanism. If a token is stolen, it remains valid until it expires. Solutions:

Short expiry - 15-minute access tokens minimize the damage window
Refresh token rotation - issue a new refresh token on every refresh, invalidate the old one
Denylist - maintain a set of revoked token IDs (the jti claim) in Redis. Defeats some of the stateless benefit but gives you real revocation.

import redis
import uuid

r = redis.Redis()

def create_access_token(data: dict) -> str:
    to_encode = data.copy()
    to_encode.update({
        "jti": str(uuid.uuid4()),  # unique token ID
        "exp": datetime.now(timezone.utc) + timedelta(minutes=15),
    })
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def is_token_revoked(jti: str) -> bool:
    return r.exists(f"revoked:{jti}")

def revoke_token(jti: str, ttl_seconds: int):
    r.setex(f"revoked:{jti}", ttl_seconds, "1")

Summary

JWT authentication is powerful when implemented correctly. The key things to remember:

Never store sensitive data in the payload
Use environment variables for secret keys
Keep access tokens short-lived, pair with refresh tokens
Whitelist algorithms when decoding
Understand the revocation tradeoff and design accordingly

The implementation above gives you a solid production foundation. From here, you can add email verification, OAuth2 social login, or 2FA on top of the same JWT infrastructure.

Uvicorn: The Lightning-Fast ASGI Server Powering Modern Python Apps

MEROLINE LIZLENT — Wed, 13 May 2026 08:22:57 +0000

If you have ever used FastAPI or Starlette, you've already heard of Uvicorn, as it is the server that runs your app. However the majority of tutorials only discuss it as a side effect of one-liner. In this article, we'll dive deep – into the internals of Uvicorn, into correct configuration of Uvicorn and into deployment of Uvicorn in production the right way.

What Is Uvicorn?
Uvicorn is an ASGI (Asynchronous Server Gateway Interface) server implementation for Python, based on uvloop and httptools. It's designed specifically for running async Python web frameworks like FastAPI, Starlette, Django Channels, and Quart.

Why WSGI vs ASGI?
The classic server interface used in python is WSGI (which is used in Flask and Django). It's a synchronous request, one by one per worker. It is good for conventional applications and doesn't work well with:

WebSockets
Long-polling
Async database calls
High-concurrency workloads ASGI solves this. Uvicorn is the fastest server implementation for it, its the async version of WSGI.

Traditional Stack:  Nginx → Gunicorn (WSGI) → Flask/Django
Modern Stack:       Nginx → Uvicorn (ASGI)  → FastAPI/Starlette

Installation

pip install uvicorn
pip install "uvicorn[standard]" //with high performance

The [standard] extras install:

uvloop: A drop-in replacement for Python's default event loop, written in Cython. ~2-4x faster.
httptools: A fast HTTP parser (used by Node.js). Much faster than the pure-Python fallback.
websockets: WebSocket support
watchfiles: File watching for --reload

# Run app:app (file: app.py, object: app)
uvicorn app:app

# With auto-reload for development
uvicorn app:app --reload

# Bind to a specific host and port
uvicorn app:app --host 0.0.0.0 --port 8080

# Verbose logging
uvicorn app:app --log-level debug

Your app.py just needs to expose an ASGI callable:

# app.py — raw ASGI, no framework needed
async def app(scope, receive, send):
    assert scope["type"] == "http"

    await send({
        "type": "http.response.start",
        "status": 200,
        "headers": [[b"content-type", b"text/plain"]],
    })
    await send({
        "type": "http.response.body",
        "body": b"Hello from raw ASGI!",
    })

Configuration

Workers

# Single worker (default)- fine for development
uvicorn app:app

# Multiple workers — for production on a single machine
uvicorn app:app --workers 4

The general rule of thumb for worker count: (2 × CPU cores) + 1. On a 4-core machine, that's 9 workers.

Timeouts and Limits

uvicorn app:app \
  --timeout-keep-alive 5 \   # seconds to wait for next request on keep-alive connection
  --limit-concurrency 100 \  # max concurrent connections before returning 503
  --limit-max-requests 1000  # restart worker after N requests (memory leak prevention)

Programmatic Configuration
When you just want to control Uvicorn from Python code, not the CLI; useful for custom startup logic, testing, or process managers.

import uvicorn

if __name__ == "__main__":
    uvicorn.run(
        "app:app",
        host="0.0.0.0",
        port=8000,
        workers=4,
        log_level="info",
        access_log=True,
        reload=False,
    )

Or with a Config object for more control:

import uvicorn
from uvicorn.config import Config
from uvicorn.main import Server

config = Config(
    app="app:app",
    host="0.0.0.0",
    port=8000,
    log_level="warning",
    workers=2,
)
server = Server(config=config)

# Now you can do things before/after serving
import asyncio
asyncio.run(server.serve())

Logging Configuration
Uvicorn's default logging is decent, but in production you'll want structured JSON logs:

import uvicorn
import logging

LOG_CONFIG = {
    "version": 1,
    "disable_existing_loggers": False,
    "formatters": {
        "json": {
            "()": "pythonjsonlogger.jsonlogger.JsonFormatter",
            "format": "%(asctime)s %(name)s %(levelname)s %(message)s",
        }
    },
    "handlers": {
        "default": {
            "class": "logging.StreamHandler",
            "formatter": "json",
            "stream": "ext://sys.stdout",
        }
    },
    "loggers": {
        "uvicorn": {"handlers": ["default"], "level": "INFO"},
        "uvicorn.error": {"handlers": ["default"], "level": "INFO"},
        "uvicorn.access": {"handlers": ["default"], "level": "INFO"},
    },
}

uvicorn.run("app:app", log_config=LOG_CONFIG)

Production Deployment Patterns

Pattern 1: Uvicorn + Gunicorn (Recommended for Most Cases)
Gunicorn is a battle-tested process manager. Use it to manage Uvicorn workers:

pip install gunicorn

gunicorn app:app \
  -w 4 \
  -k uvicorn.workers.UvicornWorker \
  --bind 0.0.0.0:8000 \
  --timeout 60 \
  --keep-alive 5 \
  --access-logfile - \
  --error-logfile -

This gives you Gunicorn's robust process management (signal handling, worker restarts, graceful shutdown) with Uvicorn's async performance.

Pattern 2: Uvicorn Behind Nginx

upstream fastapi_app {
    server 127.0.0.1:8000;
    server 127.0.0.1:8001;  # multiple Uvicorn instances
}

server {
    listen 80;
    server_name api.yourdomain.com;

    location / {
        proxy_pass http://fastapi_app;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;

        # WebSocket support
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
    }
}

Pattern 3: Systemd Service

# /etc/systemd/system/myapp.service
[Unit]
Description=My FastAPI App
After=network.target

[Service]
Type=exec
User=www-data
WorkingDirectory=/var/www/myapp
ExecStart=/var/www/myapp/venv/bin/gunicorn app:app \
    -w 4 \
    -k uvicorn.workers.UvicornWorker \
    --bind unix:/run/myapp.sock \
    --timeout 60
Restart=on-failure
RestartSec=5s

[Install]
WantedBy=multi-user.target

systemctl enable myapp
systemctl start myapp

Pattern 4: Docker

FROM python:3.12-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

WebSocket Support
Uvicorn handles WebSockets natively; no extra config needed:

from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Echo: {data}")

This works seamlessly because Uvicorn handles both HTTP and WebSocket connections through the ASGI interface.

Lifespan Events: Startup and Shutdown
Do you need to initialize a database pool or ML model when the server starts? Use lifespan events:

from contextlib import asynccontextmanager
from fastapi import FastAPI
import asyncpg

db_pool = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    global db_pool
    db_pool = await asyncpg.create_pool("postgresql://user:pass@localhost/db")
    print("Database pool created")
    yield
    # Shutdown
    await db_pool.close()
    print("Database pool closed")

app = FastAPI(lifespan=lifespan)

Uvicorn calls these at the right moment during server start and graceful shutdown.

Performance Tips

Always use uvicorn[standard] — the uvloop + httptools combo is a free 2-4x speedup
Use--limit-max-requests in production to prevent memory leaks from gradual bloat
Unix sockets over TCP when Nginx and Uvicorn are on the same machine — reduces overhead
Profile before scaling — adding workers helps CPU-bound work; for I/O-bound async apps, one worker with high concurrency often beats many workers

Summary
Uvicorn isn't just a "uvicorn main:app" one-liner. It's a production-grade ASGI server with rich configuration options, WebSocket support, lifecycle events, and proven deployment patterns.
Getting familiar with these options means you'll be ready to take your async Python apps from laptop to production confidently.

Building Production-Ready APIs with FastAPI: The Modern Python Framework You Should Be Using

MEROLINE LIZLENT — Wed, 13 May 2026 07:48:29 +0000

If you are still using Flask or Django REST Framework to develop REST APIs but have yet to discover FastAPI, you are missing out on some real performance and developer experience. For all the right reasons, FastAPI is one of the most starred Python projects on GitHub—and for a reason, it's more than just that.
Let's get very deep into how it works and let's make something real on the way.

** What Is FastAPI?**
FastAPI is a new, high-performance Python-based web framework designed to create APIs. It is built on top of Starlette (for web parts) and Pydantic (for data validation) and natively supports Python's async/await syntax.

Key selling points:
Benchmarks: On par with NodeJS and Go
Swagger UI and ReDoc: Automatic docs out of the box.
Type safety: Type annotations for validation and editor assistance in Python:
Async-first: First-class support for async request handlers.

Getting Started

pip install fastapi uvicorn[standard]

Your simplest possible app:

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"message": "Hello, FastAPI!"}

Run it:

uvicorn main:app --reload

Visit http://127.0.0.1:8000/docs — you already have interactive Swagger docs. No extra config needed.

The Power of Pydantic Models
That's where FastAPI can help. Your request/response schemas are defined in Pydantic models and FastAPI is responsible for validation, serialization, and documentation.

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr
from typing import Optional

app = FastAPI()

class UserCreate(BaseModel):
    name: str
    email: EmailStr
    age: Optional[int] = None

class UserResponse(BaseModel):
    id: int
    name: str
    email: str

@app.post("/users", response_model=UserResponse, status_code=201)
def create_user(user: UserCreate):
    # Simulate saving to DB
    return UserResponse(id=1, name=user.name, email=user.email)

FastAPI returns detailed responses for a malformed email or missing name (422 – Unprocessable Entity) – without any extra code.

Path Parameters, Query Parameters, and Dependency Injection

from fastapi import FastAPI, Query, Depends, HTTPException

app = FastAPI()

# Path parameter
@app.get("/items/{item_id}")
def get_item(item_id: int):
    if item_id > 100:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"item_id": item_id}

# Query parameters with validation
@app.get("/search")
def search_items(
    q: str = Query(..., min_length=3, max_length=50),
    skip: int = Query(0, ge=0),
    limit: int = Query(10, le=100),
):
    return {"query": q, "skip": skip, "limit": limit}

Notice ge=0 (greater than or equal to) and le=100 (less than or equal to) — FastAPI validates these automatically.

Dependency Injection
FastAPI has a clean, built-in dependency injection system. It's great for things like database sessions, authentication, and shared logic.

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from database import SessionLocal  # your SQLAlchemy setup

app = FastAPI()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/users/{user_id}")
def read_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

The get_db function is a dependency, FastAPI calls it before your route handler and cleans up after.

Background Tasks
Do you need to send an email or log something after returning a response? FastAPI's BackgroundTasks is perfect:

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def send_welcome_email(email: str):
    # Your email logic here
    print(f"Sending welcome email to {email}")

@app.post("/register")
def register(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_welcome_email, email)
    return {"message": "Registered successfully"}

The response is returned immediately the email is sent in the background.

Middleware
Add custom middleware for things like logging, CORS, or timing:

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
import time

app = FastAPI()

# CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://yourfrontend.com"],
    allow_methods=["*"],
    allow_headers=["*"],
)

# Custom timing middleware
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start = time.time()
    response = await call_next(request)
    duration = time.time() - start
    response.headers["X-Process-Time"] = str(duration)
    return response

Routers: Structuring Larger Applications
Don't put everything in main.py. Use APIRouter to split your app:

# routers/users.py
from fastapi import APIRouter

router = APIRouter(prefix="/users", tags=["users"])

@router.get("/")
def list_users():
    return [{"id": 1, "name": "Alice"}]

@router.get("/{user_id}")
def get_user(user_id: int):
    return {"id": user_id, "name": "Alice"}

# main.py
from fastapi import FastAPI
from routers import users, products, auth

app = FastAPI()
app.include_router(users.router)
app.include_router(products.router)
app.include_router(auth.router)

Testing FastAPI Apps

FastAPI integrates nicely with pytest via TestClient:

from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_create_user():
    response = client.post("/users", json={"name": "Bob", "email": "bob@example.com"})
    assert response.status_code == 201
    assert response.json()["name"] == "Bob"

Summary

There is a sweet spot to FastAPI: it's fast enough for production, ergonomic enough to get a prototype done fast, and typed enough to keep large bases maintainable. Saving hours of documentation work per project alone with the auto-generated docs.
As of today, unless you have a strong reason to do otherwise, FastAPI should be your top choice if you are building a new Python API project.

Exploratory Data Analysis (EDA) Workflow

MEROLINE LIZLENT — Mon, 20 Apr 2026 14:06:41 +0000

Introduction

Exploratory Data Analysis (EDA) is the crucial first step in any data project. It involves summarizing, visualizing, and understanding your dataset to uncover patterns, detect anomalies, and generate insights before modeling.

Why EDA Matters

Data is messier than ever (streaming sources, mixed types, missing values from APIs).
Models (including LLMs) are sensitive to data quality; bad EDA leads to garbage-in-garbage-out.
It's iterative and creative: ask questions, visualize, transform, repeat.

Core EDA Workflow

A solid, reusable EDA workflow typically follows these phases. It's not strictly linear you'll loop back often.

Understand the Business/Problem Context

What question are you trying to answer? (e.g., "Why is churn increasing?" or "What drives sales?")

Know the domain, key metrics, and stakeholders. This guides what to look for and prevents aimless exploration.
Load and Inspect the Data

Import your dataset and get a quick overview.
- Shape, data types, first/last rows, summary statistics.
- Check for duplicates, unexpected values, memory usage.
Data Cleaning & Quality Checks

Handle issues that would break downstream analysis:
- Missing values (percentage per column, patterns in missingness).
- Outliers and anomalies.
- Inconsistent formatting (dates, strings, categories).
- Incorrect data types (e.g., numeric stored as object).
Univariate Analysis (Understand individual variables)
- Numerical: histograms, box plots, summary stats (mean, median, std, skewness, kurtosis).
- Categorical: value counts, bar plots, frequency tables. Goal: Understand distributions and spot weirdness.
Bivariate & Multivariate Analysis Relationships between variables
- Numerical vs Numerical: scatter plots, correlation matrices/heatmaps.
- Numerical vs Categorical: box plots, violin plots, groupby aggregations.
- Categorical vs Categorical: stacked bars, crosstabs, chi-square if needed. Look for correlations, interactions, and potential confounders.
Feature Engineering & Transformations (during EDA)

Create new features (ratios, bins, groupings, time-based).

Apply log/power transforms for skewed data.

Test assumptions (normality, linearity).
Advanced Checks & Hypothesis Generation
- Segment analysis (group by key categories).
- Time-series specific (trends, seasonality if applicable).
- Dimensionality reduction (PCA, t-SNE) for high-dim data. Document insights, questions, and next steps.
Document & Communicate Findings

Create a clean notebook/report with key plots and takeaways.

Use storytelling: "We noticed X correlates strongly with Y, but only in segment Z."

Tools & Libraries

Core Stack:
- pandas + numpy — loading, cleaning, stats.
- matplotlib + seaborn — beautiful, statistical visualizations.
- plotly or altair — interactive plots.
Automation Boosters:
- pandas-profiling / ydata-profiling — one-line HTML report.
- Sweetviz, AutoViz, D-Tale — quick overviews.
- PandasAI — ask natural language questions about your DataFrame.
Others: scipy/statsmodels for deeper stats, missingno for missing data visualization.

Example starter imports:

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns

# Optional modern additions
import plotly.express as px
# from ydata_profiling import ProfileReport

Practical Code Examples (Titanic Dataset Style)

Assume you have a DataFrame df.

Step 2-3: Quick Inspection & Cleaning

print(df.shape)
print(df.info())
print(df.describe(include='all'))  # includes categorical

# Missing values
print(df.isnull().sum() / len(df) * 100)

# Quick profile (if using ydata-profiling)
# ProfileReport(df, title="EDA Report").to_file("report.html")

Univariate Visualization

# Numerical
sns.histplot(df['age'], kde=True)
plt.title('Age Distribution')
plt.show()

# Categorical
sns.countplot(data=df, x='sex')
plt.show()

Bivariate

# Correlation heatmap
numeric_df = df.select_dtypes(include=np.number)
sns.heatmap(numeric_df.corr(), annot=True, cmap='coolwarm')
plt.title('Correlation Matrix')
plt.show()

# Relationship
sns.boxplot(data=df, x='pclass', y='fare')
plt.show()

Handling Issues

# Impute example (mean/median/mode or advanced)
df['age'] = df['age'].fillna(df['age'].median())

# Outlier detection (simple IQR)
Q1 = df['fare'].quantile(0.25)
Q3 = df['fare'].quantile(0.75)
IQR = Q3 - Q1
df = df[~((df['fare'] < (Q1 - 1.5*IQR)) | (df['fare'] > (Q3 + 1.5*IQR)))]

Mastering Regular Expressions in Go

MEROLINE LIZLENT — Mon, 20 Apr 2026 12:05:28 +0000

Go has an inbuilt regexp package which supports regular expression by the RE2 engine. This is because of this single and self-sufficient design choice, Go regexps are safe, predictable and production-ready.
RE2 is guaranteed O(n) time with respect to input length, avoiding disastrous backtracking and ReDoS vulnerabilities - even with untrusted user input.

1. Compile vs MustCompile
Before matching, a pattern must be compiled into a *regexp.Regexp. Two functions handle this:

// Compile → returns an error. Use for dynamic / user-supplied patterns.
re, err := regexp.Compile(`\d{3}-\d{4}`)
if err != nil {
    log.Fatal("invalid pattern:", err)
}

// MustCompile → panics on invalid pattern. Use for known-good package-level vars.
var phoneRe = regexp.MustCompile(`\d{3}-\d{4}`)

// Invalid pattern example
_, badErr := regexp.Compile(`[invalid`)
// error parsing regexp: missing closing ]: `[invalid`

2. Basic Matching
Three levels of match output — boolean, first match, all matches:

var digitRe = regexp.MustCompile(`\d+`)

text := "Order #4821 placed on 2024-06-15, total: $99.99"

fmt.Println(digitRe.MatchString(text))      // true
fmt.Println(digitRe.FindString(text))       // "4821"
fmt.Println(digitRe.FindAllString(text, -1)) // [4821 2024 06 15 99 99]
fmt.Println(digitRe.FindAllString(text, 2))  // [4821 2024]  (first 2)

3. Positional (Index) Matching
Index methods return byte offsets instead of strings — useful when you need to reconstruct or replace around the match.

loc := digitRe.FindStringIndex(text)
// loc = [7, 11]  →  text[7:11] = "4821"

allLocs := digitRe.FindAllStringIndex(text, -1)
// [[7 11] [22 26] [27 29] [30 32] [42 44] [45 47]]

Capture Groups (Numbered) Use FindStringSubmatch to get the full match plus all capture group contents. Index 0 is always the full match.

var ipRe = regexp.MustCompile(`(\\d{1,3})\\.(\\d{1,3})\\.(\\d{1,3})\\.(\\d{1,3})`)

m := ipRe.FindStringSubmatch("Server: 192.168.1.42")
// m[0] = "192.168.1.42"   (full match)
// m[1] = "192"             (group 1)
// m[2] = "168"             (group 2)
// m[3] = "1"               (group 3)
// m[4] = "42"              (group 4)

// FindAllStringSubmatch — all matches with their groups
allMatches := ipRe.FindAllStringSubmatch(input, -1)

5. Named Capture Groups
Syntax: _(?P...) _ makes code far more readable and survives pattern refactoring.

var dateRe = regexp.MustCompile(
    `(?P<year>\\d{4})-(?P<month>\\d{2})-(?P<day>\\d{2})`)

func ParseDate(s string) map[string]string {
    match := dateRe.FindStringSubmatch(s)
    if match == nil { return nil }

    result := make(map[string]string)
    for i, name := range dateRe.SubexpNames() {
        if name != "" { result[name] = match[i] }
    }
    return result
}

// ParseDate("2024-11-28") → map[year:2024 month:11 day:28]

//  direct index lookup by name
yearIdx := dateRe.SubexpIndex("year")  // 1

6. Replace Methods

// 1. ReplaceAllString — static, supports $N back-references
wsRe   := regexp.MustCompile(`\\s+`)
wsRe.ReplaceAllString("foo   bar  baz", " ")  // "foo bar baz"

swapRe := regexp.MustCompile(`(\\w+)=(\\w+)`)
swapRe.ReplaceAllString("a=1 b=2", "$2=$1")  // "1=a 2=b"

// 2. ReplaceAllStringFunc — dynamic via a function
digitRe.ReplaceAllStringFunc("item1 qty5", func(s string) string {
    n, _ := strconv.Atoi(s)
    return strconv.Itoa(n * 2)
})  // "item2 qty10"

// 3. ReplaceAllLiteralString — $ signs are NOT interpreted
litRe  := regexp.MustCompile(`foo`)
litRe.ReplaceAllLiteralString("foobar", "$1")  // "$1bar"

7. Split

sepRe := regexp.MustCompile(`[,;\\s]+`)

sepRe.Split("one,two; three  four", -1)
// ["one" "two" "three" "four"]

sepRe.Split("a,b,c,d", 3)  // ["a" "b" "c,d"]  (n=3 → at most 3 pieces)

8. Longest match

re.Longest() switches the engine to leftmost-longest (POSIX) semantics before the first call.

greedyRe := regexp.MustCompile(`a+`)
greedyRe.Longest()
greedyRe.FindString("aaa")  // "aaa"  (not just "a")

9. LiteralPrefix
LiteralPrefix() returns the fixed-string prefix before the first metacharacter. The engine uses this to fast-skip non-matching positions.

urlRe              := regexp.MustCompile(`^https://`)
prefix, complete := urlRe.LiteralPrefix()
// prefix = "https://"   complete = true

partialRe          := regexp.MustCompile(`^https?://`)
p2, c2          := partialRe.LiteralPrefix()
// p2 = "http"   c2 = false  (the ? breaks the literal)

10. Real-World: Email pattern checker

package main

import (
    "fmt"
    "regexp"
)

func main() {
    email := "test@example.com"

    // Simple email regex pattern
    pattern := `^[a-zA-Z0-9._%+\-]+@[a-zA-Z0-9.\-]+\.[a-zA-Z]{2,}$`

    re := regexp.MustCompile(pattern)

    if re.MatchString(email) {
        fmt.Println("Valid email")
    } else {
        fmt.Println("Invalid email")
    }
}

When Should You Use RegExp?
Use it when:

Validating input (email, phone, etc.)
Parsing logs
Extracting structured data

Further Reading
Official docs: https://pkg.go.dev/regexp
RE2 syntax ref: https://pkg.go.dev/regexp/syntax
Run all examples: go run regexp_complete.go

Forem: MEROLINE LIZLENT

Python asyncio

The Core Problem: I/O Is Slow

The Event Loop

Coroutines: async def and await

asyncio.gather(): Run Coroutines Concurrently

Handling Errors in gather()

Tasks: Fire and Don't Wait

asyncio.wait() and as_completed()

Timeouts

Queues: Producer-Consumer Pattern

Semaphores: Limit Concurrency

Locks and Synchronization

Running Blocking Code in asyncio

Async Context Managers and Iterators

Common Mistakes

Calling a coroutine without awaiting it

Using time.sleep() instead of asyncio.sleep()

Forgetting asyncio.run() or running in wrong context

CPU-bound work on the event loop

When to Use asyncio (and When Not To)

Wrapping Up

Python's `ast`

What Is an Abstract Syntax Tree?

Parsing Code

Inspecting Nodes

Getting Line Numbers

NodeVisitor

Example: Find All Function Definitions

Real-World Analysis Examples

Find All Imports

Find All print() Calls

Detect Mutable Default Arguments

Count Complexity

ast.unparse()

NodeTransformer: Modifying the AST

Replace All print() with logger.info()

Safe Expression Evaluation: ast.literal_eval()

Building a Simple Linter

Summary

Python's `shutil`

Why shutil Over os?

Copying Files

shutil.copy() — Content + Permissions

shutil.copy2() — Content + Full Metadata

shutil.copyfile() — Content Only

shutil.copyfileobj() — File Object to File Object

Copying Entire Directory Trees

With Filters

Copy Into Existing Directory (Python 3.8+)

Moving Files and Directories

Deleting Directory Trees

Archiving: zip, tar, gzip, bz2

Creating Archives

Extracting Archives

Supported Formats

Disk Usage

Finding Executables

Terminal Size

Real-World Patterns

Timestamped Backups

Safe Directory Reset

Deployment Script

Archive and Rotate Old Logs

Quick Reference

Summary

GitPython

Installation

Opening a Repository

Repository Basics

Cloning a Repository

Staging and Committing

Commit with Custom Author

Branches

Working with Remotes

Reading Commit History

Filter by Author, Path, or Date

Diffs: What Changed?

Tags

Reading File Contents from Git

Coroutines: `async def` and `await`

`asyncio.gather()`: Run Coroutines Concurrently

`asyncio.wait()` and `as_completed()`

Using `time.sleep()` instead of `asyncio.sleep()`

Forgetting `asyncio.run()` or running in wrong context

Find All `print()` Calls

`ast.unparse()`

Replace All `print()` with `logger.info()`

Safe Expression Evaluation: `ast.literal_eval()`

Why `shutil` Over `os`?

`shutil.copy()` — Content + Permissions

`shutil.copy2()` — Content + Full Metadata

`shutil.copyfile()` — Content Only

`shutil.copyfileobj()` — File Object to File Object

`auth.py` — Core JWT Logic

`models.py`

`database.py` (Mock DB)

`main.py` — The Full Application

Using the `none` algorithm

Not validating the `exp` claim