Expert Python
Metaprogramming, Web & Production Systems

Table of Contents

I. Metaclasses & the Data Model

1.1 Everything is an Object

In Python, classes are themselves objects. Every class is an instance of some metaclass — normally type. Understanding this chain is the key to metaprogramming.

class Student:
    pass

s = Student()

print(type(s))          # <class '__main__.Student'>  — s is an instance of Student
print(type(Student))    # <class 'type'>              — Student is an instance of type
print(type(type))       # <class 'type'>              — type is an instance of itself

# The chain:
# object  ←  type
#   ↑         ↑
#   |         |
# Student ── instance ── s

1.2 Creating Classes Dynamically with type()

The type(name, bases, dict) form creates a new class at runtime.

# Normal class definition
class Dog:
    species = "Canis familiaris"
    def bark(self):
        return "Woof!"

# Equivalent dynamic creation
Dog2 = type("Dog2", (), {
    "species": "Canis familiaris",
    "bark": lambda self: "Woof!"
})

d = Dog2()
print(d.bark())              # Woof!
print(Dog2.__name__)         # Dog2
print(Dog2.__bases__)        # (<class 'object'>)

1.3 Writing a Custom Metaclass

class RegistryMeta(type):
    registry = {}

    def __new__(mcs, name, bases, namespace):
        cls = super().__new__(mcs, name, bases, namespace)
        if name != "Base":                 # skip the base itself
            RegistryMeta.registry[name] = cls
        return cls

class Base(metaclass=RegistryMeta):
    pass

class EmailSender(Base):
    def send(self): return "email sent"

class SMSSender(Base):
    def send(self): return "sms sent"

print(list(RegistryMeta.registry.keys()))
# ['EmailSender', 'SMSSender']

# Plugin pattern: pick a sender by name at runtime
sender_cls = RegistryMeta.registry["EmailSender"]
print(sender_cls().send())     # email sent

class NamingMeta(type):
    def __new__(mcs, name, bases, namespace):
        if not name[0].isupper():
            raise TypeError(f"Class name '{name}' must be PascalCase")
        for attr in namespace:
            if attr.startswith("get_") and not callable(namespace[attr]):
                raise TypeError(f"Method '{attr}' must be callable")
        return super().__new__(mcs, name, bases, namespace)

class GoodClass(metaclass=NamingMeta):        # OK
    def get_value(self): return 42

# class badClass(metaclass=NamingMeta):       # TypeError
#     pass

class SingletonMeta(type):
    _instances = {}

    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            cls._instances[cls] = super().__call__(*args, **kwargs)
        return cls._instances[cls]

class Logger(metaclass=SingletonMeta):
    def __init__(self):
        self.messages = []

a, b = Logger(), Logger()
print(a is b)              # True

1.4 __init_subclass__ — the Lighter Alternative

Since Python 3.6, most metaclass use-cases can be replaced by __init_subclass__, which is simpler and composes better.

class Base:
    registry = {}

    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)
        Base.registry[cls.__name__] = cls

class Alpha(Base): pass
class Beta(Base):  pass

print(Base.registry.keys())
# dict_keys(['Alpha', 'Beta'])

1.5 __set_name__ and Descriptor-Aware Classes

class Field:
    def __set_name__(self, owner, name):
        self.name = name

    def __get__(self, obj, objtype=None):
        if obj is None:
            return self
        return obj.__dict__.get(self.name)

    def __set__(self, obj, value):
        obj.__dict__[self.name] = value

class Model:
    name = Field()
    email = Field()

m = Model()
m.name = "Aarav"
m.email = "aarav@example.com"
print(m.name, m.email)      # Aarav aarav@example.com

1.6 Common Use Cases for Metaclasses

1.7 Metaclass Conflicts

class CombinedMeta(MetaA, MetaB):
    pass

class MyClass(BaseA, BaseB, metaclass=CombinedMeta):
    pass

II. Descriptors & __slots__

2.1 What is a Descriptor?

2.2 Data vs Non-Data Descriptors

Attribute lookup order: data descriptor → instance dict → non-data descriptor → class dict → __getattr__.

class PositiveNumber:
    """Descriptor that only accepts positive numbers."""
    def __set_name__(self, owner, name):
        self.name = name
        self.private = f"_{name}"

    def __get__(self, obj, objtype=None):
        if obj is None:
            return self
        return getattr(obj, self.private, None)

    def __set__(self, obj, value):
        if not isinstance(value, (int, float)):
            raise TypeError(f"{self.name} must be numeric")
        if value <= 0:
            raise ValueError(f"{self.name} must be positive")
        setattr(obj, self.private, value)

class Product:
    price = PositiveNumber()
    quantity = PositiveNumber()

    def __init__(self, name, price, quantity):
        self.name = name
        self.price = price
        self.quantity = quantity

    def total(self):
        return self.price * self.quantity

p = Product("Laptop", 50000, 3)
print(p.total())              # 150000

try:
    p.price = -100
except ValueError as e:
    print("Error:", e)         # Error: price must be positive

class Temperature:
    def __init__(self, celsius):
        self._celsius = celsius

    @property
    def celsius(self):
        return self._celsius

    @celsius.setter
    def celsius(self, value):
        if value < -273.15:
            raise ValueError("Below absolute zero")
        self._celsius = value

    @property
    def fahrenheit(self):
        return self._celsius * 9 / 5 + 32

t = Temperature(25)
print(t.fahrenheit)       # 77.0
t.celsius = 30
print(t.fahrenheit)       # 86.0

class CachedProperty:
    def __init__(self, func):
        self.func = func
        self.name = func.__name__

    def __set_name__(self, owner, name):
        self.name = name

    def __get__(self, obj, objtype=None):
        if obj is None:
            return self
        value = self.func(obj)
        obj.__dict__[self.name] = value     # cache on the instance
        return value

class Expensive:
    def __init__(self, n):
        self.n = n

    @CachedProperty
    def computed(self):
        print("Computing...")
        return sum(i ** 2 for i in range(self.n))

e = Expensive(1_000_000)
print(e.computed)     # Computing... then value
print(e.computed)     # no recompute (cached)
print(e.computed)     # cached

2.3 __slots__ — Memory-Optimised Classes

By default, every instance stores its attributes in a __dict__ (a hash table). For classes with millions of instances or a fixed set of attributes, __slots__ eliminates the per-instance dict, reducing memory and speeding up attribute access.

import sys

class PointDict:
    def __init__(self, x, y):
        self.x = x
        self.y = y

class PointSlots:
    __slots__ = ("x", "y")
    def __init__(self, x, y):
        self.x = x
        self.y = y

a = PointDict(1, 2)
b = PointSlots(1, 2)

print("dict-based:", sys.getsizeof(a) + sys.getsizeof(a.__dict__))
# ~152 bytes

print("slots-based:", sys.getsizeof(b))
# ~56 bytes — nearly 3× smaller

# a.z = 3        # OK
# b.z = 3        # AttributeError: 'PointSlots' object has no attribute 'z'

2.4 Slots Rules and Trade-offs

class Base:
    __slots__ = ("x",)

class Child(Base):
    __slots__ = ("y",)             # must redeclare; do NOT repeat "x"

c = Child()
c.x = 1
c.y = 2
print(c.x, c.y)                    # 1 2

# Child without __slots__ would reintroduce __dict__ and lose memory savings.

class BadChild(Base):
    pass                           # no __slots__ → has __dict__

b = BadChild()
b.z = 99                           # allowed (dict exists)
print(b.__dict__)                  # {'z': 99}

2.5 When to Use Slots and Descriptors

III. Memory Management & Garbage Collection

3.1 Reference Counting

CPython tracks the number of references to every object. When the count drops to zero, the object is deallocated immediately.

import sys

a = [1, 2, 3]
print(sys.getrefcount(a))    # 2 (a + temp reference inside getrefcount)

b = a
print(sys.getrefcount(a))    # 3

del b
print(sys.getrefcount(a))    # 2

3.2 The Cycle Problem and Generational GC

import gc

# Check GC is enabled
print(gc.isenabled())          # True

# The three generations
print(gc.get_threshold())      # (700, 10, 10) default

# Force a collection
collected = gc.collect()
print("Collected:", collected)

# Disable (rarely needed)
gc.disable()
gc.enable()

# Inspect objects tracked by GC
gc.set_debug(gc.DEBUG_LEAK)

3.3 weakref — Non-Owning References

A weak reference lets you reference an object without preventing its garbage collection. Ideal for caches, observer lists and back-references.

import weakref

class Node:
    def __init__(self, name):
        self.name = name

n = Node("root")
r = weakref.ref(n)

print(r())              # <Node object at ...>
print(r().name)         # root

del n
print(r())              # None — object was collected

# WeakValueDictionary — auto-removes dead entries
cache = weakref.WeakValueDictionary()

class Cached:
    def __init__(self, key): self.key = key

c = Cached("a")
cache["a"] = c
print(len(cache))       # 1
del c
print(len(cache))       # 0 (auto-removed)

import weakref

class Child:
    def __init__(self, parent, name):
        self.parent = weakref.ref(parent)     # weak back-reference
        self.name = name

class Parent:
    def __init__(self, name):
        self.name = name
        self.children = []

    def add(self, child_name):
        c = Child(self, child_name)
        self.children.append(c)
        return c

p = Parent("root")
p.add("child1")
p.add("child2")

# Even after deleting p, no cycle keeps it alive
# del p          # parent and children are freed together

3.4 __del__ and Finalisers

import weakref

class Resource:
    def __init__(self, name):
        self.name = name
        self.finalizer = weakref.finalize(self, self._cleanup, name)

    @staticmethod
    def _cleanup(name):
        print(f"Cleaning up {name}")

r = Resource("db-connection")
del r        # Clean up printed exactly once, deterministically

3.5 Measuring Memory

import sys, tracemalloc

# Object size
print(sys.getsizeof([1, 2, 3]))          # ~80 bytes for the list container

# Total heap usage with tracemalloc
tracemalloc.start()

data = [str(i) for i in range(100_000)]
snapshot = tracemalloc.take_snapshot()

for stat in snapshot.statistics("lineno")[:5]:
    print(stat)

tracemalloc.stop()

3.6 Common Memory Leaks in Python

IV. Profiling & Performance Optimisation

4.1 Measure Before You Optimise

4.2 Timing with timeit

import timeit

# Time a single expression (runs it 1,000,000 times)
t = timeit.timeit("'-'.join(str(n) for n in range(100))", number=10_000)
print(f"{t:.4f}s")

# Compare two implementations
setup = "data = list(range(1000))"
a = timeit.timeit("sum(x**2 for x in data)", setup=setup, number=1000)
b = timeit.timeit("[x**2 for x in data]", setup=setup, number=1000)
print(f"generator: {a:.4f}s, list: {b:.4f}s")

# Command-line
# python -m timeit -n 10000 -s "data=list(range(1000))" "sum(x for x in data)"

4.3 cProfile — Function-Level Profiling

import cProfile

def slow():
    return sum(i ** 2 for i in range(1_000_000))

def fast():
    return sum(i * i for i in range(1_000_000))

def main():
    for _ in range(3):
        slow(); fast()

cProfile.run("main()", sort="cumulative")

# Command line
# python -m cProfile -s cumtime script.py

4.4 line_profiler — Line-by-Line Timing

pip install line_profiler

# In code:
from line_profiler import LineProfiler

def process(data):
    result = []
    for x in data:
        result.append(x ** 2)
    return sum(result)

profiler = LineProfiler()
profiler.add_function(process)
profiler.enable_by_count()
process(list(range(100_000)))
profiler.print_stats()

# Or use the kernel: kernprof -l -v script.py

4.5 memory_profiler — Line-by-Line Memory

pip install memory_profiler

# Decorate a function
from memory_profiler import profile

@profile
def build_large():
    a = [i for i in range(1_000_000)]
    b = [i * 2 for i in a]
    del a
    return b

build_large()

# Command-line
# python -m memory_profiler script.py

4.6 Optimisation Checklist

import timeit

# 1. Join vs += in a loop
setup = "words = ['a'] * 1000"
t1 = timeit.timeit("s=''\nfor w in words: s += w", setup=setup, number=1000)
t2 = timeit.timeit("''.join(words)", setup=setup, number=1000)
print(f"+=: {t1:.4f}s   join: {t2:.4f}s")   # join is 5–10× faster

# 2. Set vs list membership
setup = "data = list(range(10_000)); target = 9999"
t1 = timeit.timeit("target in data", setup=setup, number=10_000)
t2 = timeit.timeit("target in set(data)", setup=setup + "\ns = set(data)",
                   number=10_000)
print(f"list: {t1:.4f}s   set: {t2:.4f}s")

# 3. Local vs global name lookup
setup = "x = 0\n"
t1 = timeit.timeit("for _ in range(1000): x", setup=setup, number=1000)
t2 = timeit.timeit("def f():\n    y=0\n    for _ in range(1000): y\nf()",
                   setup=setup, number=1000)

from functools import lru_cache
import timeit

def fib_plain(n):
    if n <= 1: return n
    return fib_plain(n - 1) + fib_plain(n - 2)

@lru_cache(maxsize=None)
def fib_cached(n):
    if n <= 1: return n
    return fib_cached(n - 1) + fib_cached(n - 2)

# fib_plain(30) would take seconds; fib_cached(30) is instant
print(timeit.timeit("fib_cached(30)", globals=globals(), number=1000))
print(fib_cached.cache_info())

4.7 Amdahl's Law

If only 50% of the code is parallelisable (\(p=0.5\)), the maximum speedup with infinite cores is \(2\times\). This is why profiling must guide parallelisation efforts.

V. C-Extensions, Cython & Numba

5.1 Why Extend Python?

Pure Python is fast enough for most tasks, but CPU-bound hot loops can be 10–100× slower than C. Python provides several escape hatches:

5.2 ctypes — Calling C Without Compiling Python Code

# Step 1: write a small C library (libmath.c)
# int square(int x) { return x * x; }

# Compile:
# gcc -shared -fPIC -o libmath.so libmath.c       (Linux/macOS)
# cl /LD libmath.c                                (Windows)

# Step 2: call it from Python
import ctypes

lib = ctypes.CDLL("./libmath.so")
lib.square.argtypes = [ctypes.c_int]
lib.square.restype  = ctypes.c_int

print(lib.square(12))          # 144

# Calling libc functions directly
libc = ctypes.CDLL(None)
print(libc.strlen(b"hello"))   # 5

5.3 Cython — Python Syntax, C Speed

Cython is a superset of Python that compiles to C. You can start with unmodified Python code and progressively add type declarations for speed.

# File: primes.pyx
def count_primes(int limit):
    cdef int n, d, count = 0
    cdef bint is_prime
    for n in range(2, limit):
        is_prime = True
        for d in range(2, int(n ** 0.5) + 1):
            if n % d == 0:
                is_prime = False
                break
        if is_prime:
            count += 1
    return count

# Build: create setup.py
# from setuptools import setup
# from Cython.Build import cythonize
# setup(ext_modules=cythonize("primes.pyx"))
#
# python setup.py build_ext --inplace
#
# Then: import primes; primes.count_primes(1_000_000)

5.4 Numba — JIT for Numeric Code

pip install numba

from numba import njit
import numpy as np
import time

@njit
def sum_squares(n):
    total = 0
    for i in range(n):
        total += i * i
    return total

# First call compiles; subsequent calls are native speed
start = time.perf_counter()
print(sum_squares(10_000_000))
print(f"JIT: {time.perf_counter() - start:.3f}s")

# Compare with pure Python
def sum_squares_py(n):
    total = 0
    for i in range(n):
        total += i * i
    return total

start = time.perf_counter()
print(sum_squares_py(10_000_000))
print(f"Python: {time.perf_counter() - start:.3f}s")

Typical speedup: 50–200× for numeric loops.

5.5 Choosing an Approach

import numpy as np
import time
from numba import njit

N = 10_000_000

# Pure Python
def pure(n):
    s = 0
    for i in range(n):
        s += i * i
    return s

# NumPy vectorised
def numpy_version(n):
    arr = np.arange(n, dtype=np.int64)
    return int((arr * arr).sum())

# Numba JIT
@njit
def numba_version(n):
    s = 0
    for i in range(n):
        s += i * i
    return s

numba_version(1)  # warm up

for name, fn in [("Pure", pure), ("NumPy", numpy_version), ("Numba", numba_version)]:
    start = time.perf_counter()
    fn(N)
    print(f"{name:<6}: {time.perf_counter() - start:.3f}s")

# Typical results on a laptop:
# Pure : 1.20 s
# NumPy: 0.06 s
# Numba: 0.015 s

VI. Web Development with Flask

6.1 What is a Web Framework?

6.2 Your First Flask App

pip install flask

# app.py
from flask import Flask

app = Flask(__name__)

@app.route("/")
def home():
    return "<h1>Hello, Flask!</h1>"

@app.route("/about")
def about():
    return {"course": "INT108", "topic": "Flask"}

if __name__ == "__main__":
    app.run(debug=True, host="0.0.0.0", port=5000)

Run with python app.py and visit http://localhost:5000/.

6.3 Routing and URL Parameters

from flask import Flask
app = Flask(__name__)

# Path parameter (string by default)
@app.route("/user/<name>")
def user(name):
    return f"Hello, {name}!"

# Typed path parameter
@app.route("/post/<int:post_id>")
def show_post(post_id):
    return f"Post #{post_id}"

# Float and path converters
@app.route("/price/<float:amount>")
def price(amount):
    return f"Price: {amount}"

@app.route("/files/<path:subpath>")
def files(subpath):
    return f"File path: {subpath}"

# HTTP methods
@app.route("/items", methods=["GET", "POST"])
def items():
    from flask import request
    if request.method == "POST":
        return {"status": "created"}, 201
    return {"items": []}

6.4 Request and Response

from flask import Flask, request, jsonify, make_response

app = Flask(__name__)

@app.route("/search")
def search():
    # Query string: /search?q=python&limit=10
    q     = request.args.get("q", "")
    limit = request.args.get("limit", 10, type=int)
    return {"query": q, "limit": limit}

@app.route("/submit", methods=["POST"])
def submit():
    # JSON body
    data = request.get_json(silent=True) or {}
    name = data.get("name")
    return jsonify({"received": name}), 201

@app.route("/custom")
def custom():
    response = make_response("Custom body", 202)
    response.headers["X-Course"] = "INT108"
    response.set_cookie("session_id", "abc123")
    return response

6.5 Templates with Jinja2

# templates/hello.html
<!DOCTYPE html>
<html>
<head><title>{{ title }}</title></head>
<body>
  <h1>Hello, {{ name }}!</h1>
  <ul>
    {% for item in items %}
      <li>{{ item }}</li>
    {% endfor %}
  </ul>
  <p>Course: {{ course }}</p>
</body>
</html>

from flask import Flask, render_template

app = Flask(__name__)

@app.route("/hello/<name>")
def hello(name):
    return render_template(
        "hello.html",
        title="Welcome",
        name=name,
        items=["Python", "Flask", "Jinja2"],
        course="INT108",
    )

6.6 Blueprints — Modular Applications

# auth.py
from flask import Blueprint, jsonify

auth_bp = Blueprint("auth", __name__, url_prefix="/auth")

@auth_bp.route("/login", methods=["POST"])
def login():
    return jsonify({"status": "logged in"})

@auth_bp.route("/logout", methods=["POST"])
def logout():
    return jsonify({"status": "logged out"})

# app.py
from flask import Flask
from auth import auth_bp

app = Flask(__name__)
app.register_blueprint(auth_bp)
# Routes available at /auth/login, /auth/logout

6.7 Configuration and Environment

import os
from flask import Flask

class Config:
    SECRET_KEY = os.getenv("SECRET_KEY", "dev-key")
    DEBUG      = False
    DB_URL     = os.getenv("DATABASE_URL", "sqlite:///app.db")

class DevConfig(Config):
    DEBUG = True

class ProdConfig(Config):
    DEBUG = False

app = Flask(__name__)
app.config.from_object(ProdConfig if os.getenv("ENV") == "prod" else DevConfig)

print(app.config["DEBUG"])

6.8 Simple REST API with Flask

from flask import Flask, request, jsonify, abort

app = Flask(__name__)

tasks = {}
next_id = 1

@app.route("/tasks", methods=["GET"])
def list_tasks():
    return jsonify(list(tasks.values()))

@app.route("/tasks/<int:task_id>", methods=["GET"])
def get_task(task_id):
    task = tasks.get(task_id)
    if not task:
        abort(404, description="Task not found")
    return jsonify(task)

@app.route("/tasks", methods=["POST"])
def create_task():
    global next_id
    data = request.get_json(silent=True) or {}
    if "title" not in data:
        abort(400, description="title is required")
    task = {"id": next_id, "title": data["title"], "done": False}
    tasks[next_id] = task
    next_id += 1
    return jsonify(task), 201

@app.route("/tasks/<int:task_id>", methods=["PUT"])
def update_task(task_id):
    task = tasks.get(task_id)
    if not task:
        abort(404)
    data = request.get_json(silent=True) or {}
    task.update({k: v for k, v in data.items() if k in {"title", "done"}})
    return jsonify(task)

@app.route("/tasks/<int:task_id>", methods=["DELETE"])
def delete_task(task_id):
    if task_id not in tasks:
        abort(404)
    del tasks[task_id]
    return "", 204

@app.errorhandler(404)
def not_found(e):
    return jsonify({"error": str(e)}), 404

@app.errorhandler(400)
def bad_request(e):
    return jsonify({"error": str(e)}), 400

if __name__ == "__main__":
    app.run(debug=True)

6.9 Production Server

Flask's dev server is not for production. Use gunicorn (Linux) or waitress (Windows).

pip install gunicorn

# Run with 4 worker processes
gunicorn -w 4 -b 0.0.0.0:8000 app:app

# Waitress (cross-platform)
pip install waitress
waitress-serve --port=8000 app:app

VII. REST APIs with FastAPI

7.1 Why FastAPI?

pip install fastapi uvicorn[standard]

7.2 Hello World

from fastapi import FastAPI

app = FastAPI(title="INT108 API", version="1.0.0")

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

# Run: uvicorn main:app --reload
# Docs:  http://127.0.0.1:8000/docs
# ReDoc: http://127.0.0.1:8000/redoc

7.3 Path and Query Parameters

from fastapi import FastAPI, Query, Path
from typing import Optional

app = FastAPI()

@app.get("/items/{item_id}")
async def get_item(
    item_id: int = Path(..., gt=0, description="Positive item id"),
    q: Optional[str] = Query(None, max_length=50),
    limit: int = Query(10, ge=1, le=100),
):
    return {"item_id": item_id, "q": q, "limit": limit}

7.4 Pydantic Models for Request/Response

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

app = FastAPI()

class StudentCreate(BaseModel):
    name:  str = Field(..., min_length=2, max_length=50)
    email: EmailStr
    age:   int = Field(..., ge=15, le=100)
    branch: str = "CSE"

class StudentOut(StudentCreate):
    id: int

# In-memory store
students: dict[int, StudentOut] = {}
next_id = 1

@app.post("/students", response_model=StudentOut, status_code=201)
async def create_student(payload: StudentCreate):
    global next_id
    student = StudentOut(id=next_id, **payload.model_dump())
    students[next_id] = student
    next_id += 1
    return student

@app.get("/students", response_model=list[StudentOut])
async def list_students():
    return list(students.values())

@app.get("/students/{student_id}", response_model=StudentOut)
async def get_student(student_id: int):
    if student_id not in students:
        raise HTTPException(status_code=404, detail="Student not found")
    return students[student_id]

7.5 Async Endpoints and Concurrency

import asyncio
from fastapi import FastAPI

app = FastAPI()

@app.get("/slow/{n}")
async def slow(n: int):
    await asyncio.sleep(n)             # non-blocking
    return {"slept": n}

@app.get("/parallel")
async def parallel():
    results = await asyncio.gather(
        asyncio.sleep(1, result="A"),
        asyncio.sleep(1, result="B"),
        asyncio.sleep(1, result="C"),
    )
    return {"results": results}        # ~1s total, not 3s

7.6 Dependency Injection

from fastapi import FastAPI, Depends, Header, HTTPException

app = FastAPI()

def verify_token(x_token: str = Header(...)):
    if x_token != "secret-token":
        raise HTTPException(status_code=401, detail="Invalid token")
    return x_token

def get_db():
    # In a real app: yield a DB session
    db = {"connected": True}
    try:
        yield db
    finally:
        db["connected"] = False

@app.get("/protected", dependencies=[Depends(verify_token)])
async def protected(db=Depends(get_db)):
    return {"db": db}

7.7 Response Models and Status Codes

from fastapi import FastAPI, status, HTTPException
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float

@app.post("/items", status_code=status.HTTP_201_CREATED)
async def create(item: Item):
    return {"created": item.name}

@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete(item_id: int):
    return None

@app.get("/items/{item_id}", responses={404: {"description": "Not found"}})
async def get(item_id: int):
    if item_id > 100:
        raise HTTPException(404, "Item not found")
    return {"id": item_id}

7.8 Background Tasks and Lifespan

from fastapi import FastAPI, BackgroundTasks
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    print("Starting up: connect to DB")
    app.state.cache = {}
    yield
    # Shutdown
    print("Shutting down: close DB")
    app.state.cache = None

app = FastAPI(lifespan=lifespan)

def send_email(to: str):
    print(f"Sending email to {to}")

@app.post("/signup")
async def signup(email: str, bg: BackgroundTasks):
    bg.add_task(send_email, email)
    return {"status": "queued"}

7.9 Database Integration with SQLAlchemy

from fastapi import FastAPI, Depends
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, sessionmaker, Session

DATABASE_URL = "sqlite:///./app.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine)
Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    id    = Column(Integer, primary_key=True)
    name  = Column(String, nullable=False)
    email = Column(String, unique=True, nullable=False)

Base.metadata.create_all(engine)

app = FastAPI()

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

@app.post("/users")
def create_user(name: str, email: str, db: Session = Depends(get_db)):
    user = User(name=name, email=email)
    db.add(user); db.commit(); db.refresh(user)
    return {"id": user.id, "name": user.name}

7.10 Deploying FastAPI

# Development
uvicorn main:app --reload --host 0.0.0.0 --port 8000

# Production with multiple workers
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

# Behind Gunicorn (Linux) with uvicorn workers
gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker -b 0.0.0.0:8000

VIII. Testing at Scale — pytest, Mocking, TDD

8.1 Why pytest over unittest?

pip install pytest pytest-cov pytest-mock

8.2 Basic Tests

# test_math.py
import pytest

def add(a, b):
    return a + b

def divide(a, b):
    if b == 0:
        raise ZeroDivisionError("Cannot divide by zero")
    return a / b

def test_add():
    assert add(2, 3) == 5

def test_add_negative():
    assert add(-1, -1) == -2

def test_divide():
    assert divide(10, 2) == 5

def test_divide_by_zero():
    with pytest.raises(ZeroDivisionError):
        divide(10, 0)

def test_divide_message():
    with pytest.raises(ZeroDivisionError, match="Cannot divide"):
        divide(10, 0)

# Run all tests
pytest -v

# With coverage
pytest --cov=mypackage --cov-report=term-missing

# Only failed from last run
pytest --lf

# Stop after first failure
pytest -x

8.3 Parametrised Tests

import pytest

@pytest.mark.parametrize("a, b, expected", [
    (2, 3, 5),
    (0, 0, 0),
    (-1, 1, 0),
    (100, -50, 50),
])
def test_add(a, b, expected):
    assert a + b == expected

@pytest.mark.parametrize("s, expected", [
    ("madam", True),
    ("racecar", True),
    ("hello", False),
    ("", True),
])
def test_palindrome(s, expected):
    assert (s == s[::-1]) == expected

8.4 Fixtures

import pytest

@pytest.fixture
def sample_students():
    return [
        {"name": "Aarav", "marks": [88, 92, 79]},
        {"name": "Diya",  "marks": [95, 81, 90]},
    ]

@pytest.fixture
def db_connection(tmp_path):
    """Uses tmp_path (built-in) for an isolated temporary DB per test."""
    db_file = tmp_path / "test.db"
    conn = {"file": str(db_file), "open": True}
    yield conn
    conn["open"] = False       # teardown

def test_count(sample_students):
    assert len(sample_students) == 2

def test_first_student(sample_students):
    assert sample_students[0]["name"] == "Aarav"

def test_db_uses_tmp(db_connection):
    assert db_connection["open"]
    assert db_connection["file"].endswith("test.db")

Fixture scopes

8.5 Mocking with unittest.mock

from unittest.mock import patch, MagicMock
import requests

def get_user_name(user_id):
    response = requests.get(f"https://api.example.com/users/{user_id}", timeout=10)
    response.raise_for_status()
    return response.json()["name"]

# --- Test ---
def test_get_user_name():
    with patch("requests.get") as mock_get:
        mock_response = MagicMock()
        mock_response.json.return_value = {"name": "Aarav"}
        mock_response.raise_for_status.return_value = None
        mock_get.return_value = mock_response

        assert get_user_name(1) == "Aarav"
        mock_get.assert_called_once_with(
            "https://api.example.com/users/1", timeout=10)

8.6 Test-Driven Development (TDD)

# Step 1 (RED): write tests before code
import pytest

def validate_password(pwd):
    raise NotImplementedError

@pytest.mark.parametrize("pwd, ok", [
    ("short",   False),
    ("alllowercase1!", False),
    ("ALLUPPERCASE1!", False),
    ("NoDigits!", False),
    ("GoodPass123!", True),
])
def test_password_validation(pwd, ok):
    assert validate_password(pwd) == ok

# Step 2 (GREEN): implement minimum code
import re

def validate_password(pwd: str) -> bool:
    if len(pwd) < 8:
        return False
    if not re.search(r"[a-z]", pwd):
        return False
    if not re.search(r"[A-Z]", pwd):
        return False
    if not re.search(r"\d", pwd):
        return False
    if not re.search(r"[!@#$%^&*]", pwd):
        return False
    return True

# Step 3 (REFACTOR): clean up if needed

8.7 Testing Web APIs

# FastAPI - TestClient
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_root():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello, FastAPI!"}

def test_create_student():
    payload = {"name": "Aarav", "email": "a@x.com", "age": 19}
    r = client.post("/students", json=payload)
    assert r.status_code == 201
    data = r.json()
    assert data["name"] == "Aarav"
    assert "id" in data

# Flask - test client
import pytest
from app import app

@pytest.fixture
def client():
    app.config["TESTING"] = True
    with app.test_client() as c:
        yield c

def test_home(client):
    r = client.get("/")
    assert r.status_code == 200

def test_create_task(client):
    r = client.post("/tasks", json={"title": "Study"})
    assert r.status_code == 201
    assert r.get_json()["title"] == "Study"

8.8 Code Coverage

pytest --cov=mypackage --cov-report=html

# Opens htmlcov/index.html — shows line-by-line coverage.
# Aim for >80% coverage on critical modules, but never chase 100%
# at the cost of meaningful tests.

8.9 Test Organisation

project/
├── src/mypackage/
│   ├── __init__.py
│   └── core.py
└── tests/
    ├── conftest.py           # shared fixtures
    ├── test_core.py          # unit tests
    ├── test_api.py           # integration tests
    └── test_e2e.py           # end-to-end tests

IX. Docker, CI/CD & Cloud Deployment

9.1 Why Docker?

9.2 A Production-Grade Dockerfile

# --- Build stage ---
FROM python:3.12-slim AS builder

WORKDIR /app

RUN apt-get update && apt-get install -y --no-install-recommends \
        build-essential && rm -rf /var/lib/apt/lists/*

COPY requirements.txt .
RUN pip install --user --no-cache-dir -r requirements.txt

# --- Runtime stage ---
FROM python:3.12-slim

WORKDIR /app

# Non-root user for security
RUN useradd --create-home appuser

COPY --from=builder /root/.local /home/appuser/.local
COPY src/ ./src/

ENV PATH=/home/appuser/.local/bin:$PATH \
    PYTHONUNBUFFERED=1 \
    PYTHONDONTWRITEBYTECODE=1

USER appuser

EXPOSE 8000
HEALTHCHECK --interval=30s --timeout=3s CMD \
    python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"

CMD ["uvicorn", "src.main:app", "--host", "0.0.0.0", "--port", "8000"]

9.3 docker-compose.yml — Multi-Service Apps

version: "3.9"

services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://user:pass@db:5432/mydb
    depends_on:
      db:
        condition: service_healthy
    restart: unless-stopped

  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
      POSTGRES_DB: mydb
    volumes:
      - pgdata:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U user"]
      interval: 10s
      retries: 5

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"

volumes:
  pgdata:

docker compose up --build
docker compose down -v

9.4 Docker Best Practices

# .dockerignore
.git
.github
__pycache__
*.pyc
.venv
.env
tests/
*.md
.pytest_cache
.mypy_cache
.ruff_cache

9.5 CI/CD with GitHub Actions

# .github/workflows/ci-cd.yml
name: CI/CD

on:
  push:
    branches: [main]
  pull_request:

jobs:
  test:
    runs-on: ubuntu-latest
    strategy:
      matrix:
        python-version: ["3.10", "3.11", "3.12"]
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: ${{ matrix.python-version }}
          cache: pip
      - run: pip install -e ".[dev]"
      - run: ruff check .
      - run: mypy src/
      - run: pytest --cov=src --cov-fail-under=80

  build-and-push:
    needs: test
    if: github.ref == 'refs/heads/main'
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: docker/setup-buildx-action@v3
      - uses: docker/login-action@v3
        with:
          registry: ghcr.io
          username: ${{ github.actor }}
          password: ${{ secrets.GITHUB_TOKEN }}
      - uses: docker/build-push-action@v5
        with:
          push: true
          tags: ghcr.io/${{ github.repository }}:latest

9.6 Deployment Targets

9.7 Environment Configuration in Production

# Never hardcode. Use secrets from the platform.
# 12-factor app principles:
#   - Config in environment
#   - Stateless processes
#   - Logs as event streams
#   - Dev/prod parity

import os
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str
    redis_url: str = "redis://localhost:6379"
    secret_key: str
    debug: bool = False
    log_level: str = "INFO"

    class Config:
        env_file = ".env"

settings = Settings()
print(settings.database_url)

9.8 Zero-Downtime Deployments

X. Security in Python Applications

10.1 The OWASP Top 10 in Python

10.2 Password Hashing

pip install passlib[bcrypt]

from passlib.context import CryptContext

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

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

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

h = hash_password("MySecret123!")
print(verify_password("MySecret123!", h))   # True
print(verify_password("wrong", h))          # False

10.3 JWT Authentication in FastAPI

pip install "python-jose[cryptography]" "passlib[bcrypt]"

from datetime import datetime, timedelta, timezone
from jose import jwt, JWTError
from passlib.context import CryptContext
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

SECRET_KEY = "change-me-in-prod"
ALGORITHM = "HS256"
EXPIRE_MIN = 60

pwd_ctx = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2  = OAuth2PasswordBearer(tokenUrl="token")

app = FastAPI()
fake_db = {"aarav": {"username": "aarav",
                     "password": pwd_ctx.hash("secret")}}

def create_token(username: str) -> str:
    payload = {
        "sub": username,
        "exp": datetime.now(timezone.utc) + timedelta(minutes=EXPIRE_MIN),
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

def current_user(token: str = Depends(oauth2)) -> str:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        if not username:
            raise JWTError()
        return username
    except JWTError:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Invalid token")

@app.post("/token")
def login(form: OAuth2PasswordRequestForm = Depends()):
    user = fake_db.get(form.username)
    if not user or not pwd_ctx.verify(form.password, user["password"]):
        raise HTTPException(401, "Bad credentials")
    return {"access_token": create_token(form.username), "token_type": "bearer"}

@app.get("/me")
def me(user: str = Depends(current_user)):
    return {"user": user}

10.4 Input Validation

from pydantic import BaseModel, Field, EmailStr, constr, conint

class Registration(BaseModel):
    username: constr(min_length=3, max_length=30, regex=r"^[a-zA-Z0-9_]+$")
    email: EmailStr
    age: conint(ge=18, le=120)
    password: constr(min_length=12)

# FastAPI uses this automatically:
# - returns 422 for invalid payloads
# - rejects SQL-injection attempts at the schema level for typed fields

10.5 Rate Limiting

pip install slowapi

from slowapi import Limiter
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from fastapi import FastAPI, Request

limiter = Limiter(key_func=get_remote_address)
app = FastAPI()
app.state.limiter = limiter

@app.exception_handler(RateLimitExceeded)
async def ratelimit_handler(request: Request, exc: RateLimitExceeded):
    from fastapi.responses import JSONResponse
    return JSONResponse({"error": "rate limit exceeded"}, status_code=429)

@app.get("/api/data")
@limiter.limit("100/minute")
async def data(request: Request):
    return {"ok": True}

10.6 Secrets Management

# .env (never commit)
DATABASE_URL=postgresql://...
JWT_SECRET=...

# Load with pydantic-settings or python-dotenv
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str
    jwt_secret: str

    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"

settings = Settings()

10.7 Dependency Auditing

pip install pip-audit safety

# Scan installed packages
pip-audit

# Scan requirements file
pip-audit -r requirements.txt

# Safety (alternative)
safety check

10.8 Security Checklist

XI. Observability — Metrics, Traces, Logs

11.1 The Three Pillars

11.2 Structured Logging

import logging
import json
from datetime import datetime, timezone

class JSONFormatter(logging.Formatter):
    def format(self, record):
        payload = {
            "ts":      datetime.now(timezone.utc).isoformat(),
            "level":   record.levelname,
            "logger":  record.name,
            "message": record.getMessage(),
        }
        if record.exc_info:
            payload["exc"] = self.formatException(record.exc_info)
        return json.dumps(payload)

handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())

logger = logging.getLogger("api")
logger.setLevel(logging.INFO)
logger.addHandler(handler)

logger.info("user signed in", extra={"user_id": 42})

Structured (JSON) logs are queryable in Loki/CloudWatch/Datadog, unlike free-form strings.

11.3 Metrics with Prometheus

pip install prometheus-client

from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time

REQUESTS = Counter(
    "http_requests_total", "Total HTTP requests",
    ["method", "endpoint", "status"])

LATENCY = Histogram(
    "http_request_duration_seconds", "Request latency",
    ["endpoint"])

INFLIGHT = Gauge("http_inflight_requests", "In-flight requests")

# Start a /metrics endpoint on port 8000
start_http_server(8000)

def handle_request(endpoint):
    INFLIGHT.inc()
    start = time.perf_counter()
    try:
        time.sleep(0.05)                     # simulate work
        REQUESTS.labels("GET", endpoint, "200").inc()
    finally:
        LATENCY.labels(endpoint).observe(time.perf_counter() - start)
        INFLIGHT.dec()

for _ in range(50):
    handle_request("/api/data")

11.4 FastAPI Instrumentation with OpenTelemetry

pip install opentelemetry-api opentelemetry-sdk \
    opentelemetry-instrumentation-fastapi \
    opentelemetry-exporter-otlp

from fastapi import FastAPI
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor

provider = TracerProvider()
provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
trace.set_tracer_provider(provider)

app = FastAPI()
FastAPIInstrumentor.instrument_app(app)

@app.get("/")
async def root():
    with trace.get_tracer(__name__).start_as_current_span("custom-work"):
        return {"message": "traced"}

11.5 Health Checks and Readiness

from fastapi import FastAPI, Response, status

app = FastAPI()
_state = {"db": True, "cache": True}

@app.get("/health")
async def health():
    """Liveness: is the process alive?"""
    return {"status": "ok"}

@app.get("/ready")
async def ready(response: Response):
    """Readiness: can we serve traffic?"""
    if not all(_state.values()):
        response.status_code = status.HTTP_503_SERVICE_UNAVAILABLE
        return {"status": "not ready", "deps": _state}
    return {"status": "ready"}

11.6 Distributed Tracing Concepts

11.7 Alerting Best Practices

XII. Capstone Project — Production-Ready API

The following project ties together everything in Unit V: FastAPI + Pydantic + SQLAlchemy + pytest + Docker + structured logging + health checks.

12.1 Project Structure

task-api/
├── pyproject.toml
├── Dockerfile
├── docker-compose.yml
├── .env.example
├── .dockerignore
├── src/
│   └── taskapi/
│       ├── __init__.py
│       ├── main.py           # FastAPI app + routes
│       ├── models.py         # SQLAlchemy models
│       ├── schemas.py        # Pydantic models
│       ├── database.py       # Engine, SessionLocal, Base
│       ├── config.py         # Settings (pydantic-settings)
│       ├── logging_config.py # Structured logging
│       └── auth.py           # JWT helpers
└── tests/
    ├── conftest.py
    ├── test_api.py
    └── test_auth.py

12.2 Configuration

# src/taskapi/config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str = "sqlite:///./app.db"
    jwt_secret:   str = "dev-secret"
    jwt_algorithm: str = "HS256"
    jwt_expire_min: int = 60
    log_level: str = "INFO"

    class Config:
        env_file = ".env"

settings = Settings()

12.3 Database Layer

# src/taskapi/database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base
from .config import settings

engine = create_engine(
    settings.database_url,
    connect_args={"check_same_thread": False}
        if settings.database_url.startswith("sqlite") else {},
)

SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)
Base = declarative_base()

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

12.4 Models and Schemas

# src/taskapi/models.py
from sqlalchemy import Column, Integer, String, Boolean, DateTime
from sqlalchemy.sql import func
from .database import Base

class Task(Base):
    __tablename__ = "tasks"
    id         = Column(Integer, primary_key=True)
    title      = Column(String(200), nullable=False)
    done       = Column(Boolean, default=False)
    owner      = Column(String(50), nullable=False)
    created_at = Column(DateTime(timezone=True), server_default=func.now())

# src/taskapi/schemas.py
from pydantic import BaseModel, Field
from datetime import datetime

class TaskCreate(BaseModel):
    title: str = Field(..., min_length=1, max_length=200)

class TaskUpdate(BaseModel):
    title: str | None = None
    done:  bool | None = None

class TaskOut(BaseModel):
    id: int
    title: str
    done: bool
    owner: str
    created_at: datetime

    class Config:
        from_attributes = True

12.5 Main Application

# src/taskapi/main.py
from fastapi import FastAPI, Depends, HTTPException, status, Request
from sqlalchemy.orm import Session
from .database import Base, engine, get_db
from .models import Task
from .schemas import TaskCreate, TaskUpdate, TaskOut
from .auth import current_user
from .logging_config import configure_logging, logger

Base.metadata.create_all(engine)
configure_logging()

app = FastAPI(title="Task API", version="1.0.0")

@app.middleware("http")
async def log_requests(request: Request, call_next):
    logger.info("request.start",
                extra={"method": request.method, "path": request.url.path})
    response = await call_next(request)
    logger.info("request.end",
                extra={"status": response.status_code, "path": request.url.path})
    return response

@app.get("/health")
def health():
    return {"status": "ok"}

@app.post("/tasks", response_model=TaskOut, status_code=201)
def create_task(payload: TaskCreate,
                db: Session = Depends(get_db),
                user: str = Depends(current_user)):
    task = Task(**payload.model_dump(), owner=user)
    db.add(task); db.commit(); db.refresh(task)
    return task

@app.get("/tasks", response_model=list[TaskOut])
def list_tasks(db: Session = Depends(get_db),
               user: str = Depends(current_user)):
    return db.query(Task).filter(Task.owner == user).all()

@app.get("/tasks/{task_id}", response_model=TaskOut)
def get_task(task_id: int,
             db: Session = Depends(get_db),
             user: str = Depends(current_user)):
    task = db.query(Task).filter(Task.id == task_id,
                                 Task.owner == user).first()
    if not task:
        raise HTTPException(404, "Task not found")
    return task

@app.patch("/tasks/{task_id}", response_model=TaskOut)
def update_task(task_id: int, patch: TaskUpdate,
                db: Session = Depends(get_db),
                user: str = Depends(current_user)):
    task = db.query(Task).filter(Task.id == task_id,
                                 Task.owner == user).first()
    if not task:
        raise HTTPException(404, "Task not found")
    for k, v in patch.model_dump(exclude_unset=True).items():
        setattr(task, k, v)
    db.commit(); db.refresh(task)
    return task

@app.delete("/tasks/{task_id}", status_code=204)
def delete_task(task_id: int,
                db: Session = Depends(get_db),
                user: str = Depends(current_user)):
    task = db.query(Task).filter(Task.id == task_id,
                                 Task.owner == user).first()
    if not task:
        raise HTTPException(404, "Task not found")
    db.delete(task); db.commit()

12.6 Logging Configuration

# src/taskapi/logging_config.py
import logging, json
from datetime import datetime, timezone
from .config import settings

class JSONFormatter(logging.Formatter):
    def format(self, record):
        out = {
            "ts":    datetime.now(timezone.utc).isoformat(),
            "level": record.levelname,
            "msg":   record.getMessage(),
            "name":  record.name,
        }
        for key in ("method", "path", "status", "user_id"):
            if hasattr(record, key):
                out[key] = getattr(record, key)
        if record.exc_info:
            out["exc"] = self.formatException(record.exc_info)
        return json.dumps(out)

def configure_logging():
    handler = logging.StreamHandler()
    handler.setFormatter(JSONFormatter())
    root = logging.getLogger()
    root.handlers.clear()
    root.addHandler(handler)
    root.setLevel(settings.log_level)

logger = logging.getLogger("taskapi")

12.7 Tests

# tests/conftest.py
import pytest
from fastapi.testclient import TestClient
from src.taskapi.main import app
from src.taskapi.database import Base, engine, SessionLocal

@pytest.fixture(autouse=True)
def fresh_db():
    Base.metadata.drop_all(engine)
    Base.metadata.create_all(engine)
    yield
    Base.metadata.drop_all(engine)

@pytest.fixture
def client():
    return TestClient(app)

@pytest.fixture
def auth_token(client):
    # assumes /token endpoint from auth.py
    r = client.post("/token", data={"username": "aarav",
                                     "password": "secret"})
    return r.json()["access_token"]

# tests/test_api.py
def test_health(client):
    assert client.get("/health").json() == {"status": "ok"}

def test_create_task(client, auth_token):
    headers = {"Authorization": f"Bearer {auth_token}"}
    r = client.post("/tasks", json={"title": "Study FastAPI"}, headers=headers)
    assert r.status_code == 201
    assert r.json()["title"] == "Study FastAPI"

def test_list_tasks(client, auth_token):
    headers = {"Authorization": f"Bearer {auth_token}"}
    client.post("/tasks", json={"title": "A"}, headers=headers)
    client.post("/tasks", json={"title": "B"}, headers=headers)
    r = client.get("/tasks", headers=headers)
    assert len(r.json()) == 2

def test_missing_auth(client):
    r = client.get("/tasks")
    assert r.status_code == 401

12.8 Dockerisation

# Dockerfile
FROM python:3.12-slim AS builder
WORKDIR /app
COPY pyproject.toml ./
RUN pip install --user --no-cache-dir -e .

FROM python:3.12-slim
WORKDIR /app
RUN useradd --create-home appuser
COPY --from=builder /root/.local /home/appuser/.local
COPY src/ ./src/
ENV PATH=/home/appuser/.local/bin:$PATH PYTHONUNBUFFERED=1
USER appuser
EXPOSE 8000
HEALTHCHECK CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"
CMD ["uvicorn", "src.taskapi.main:app", "--host", "0.0.0.0", "--port", "8000"]

12.9 Deployment Checklist

XIII. Summary & Quick Reference Sheet

13.1 Metaprogramming Cheat Sheet

13.2 Performance Cheat Sheet

13.3 Web Framework Cheat Sheet

13.4 Testing Cheat Sheet

13.5 Docker & Deploy Cheat Sheet

13.6 Security Cheat Sheet

13.7 Observability Cheat Sheet

XIV. Exam Tips & Practice Questions

Top 12 Exam Tips

1. Metaclasses are just subclasses of type. Say this first, then show a small registry example.
2. Prefer __init_subclass__ over metaclasses when possible — it composes better and is easier to reason about.
3. Data descriptors beat instance __dict__; non-data descriptors lose to it. State this clearly in any descriptor question.
4. __slots__ saves memory by removing __dict__. Mention that a subclass without slots reintroduces the dict.
5. CPython uses reference counting + generational GC. Use weakref to break cycles in caches and back-references.
6. Always profile before optimising. cProfile for functions, timeit for micro-benchmarks, tracemalloc for memory.
7. NumPy and Numba are the fastest wins for numeric code. Show a benchmark comparison.
8. Use FastAPI for new APIs — automatic validation, docs, and async. Use Flask for simple apps and rapid prototypes.
9. Never use the Flask dev server in production. Gunicorn/waitress/uvicorn with multiple workers.
10. Test with pytest, mock external I/O, measure coverage. Structure tests as unit / integration / e2e.
11. Docker: multi-stage, non-root, .dockerignore, healthcheck. These four signals tell an examiner you understand production.
12. Security basics matter: bcrypt for passwords, JWT with short TTL, parameterised SQL, secret manager, pip-audit in CI.

Practice Questions

XV. Full Solutions to Practice Questions

Solution 1 — object, type, custom metaclass

Every class is an instance of a metaclass (type by default). object is the root of the class hierarchy; type is a subclass of object and its own metaclass.

class LoggingMeta(type):
    def __new__(mcs, name, bases, namespace):
        print(f"[meta] creating class {name}")
        return super().__new__(mcs, name, bases, namespace)

class A(metaclass=LoggingMeta): pass
class B(A, metaclass=LoggingMeta): pass

print(type(A))          # <class '__main__.LoggingMeta'>
print(type(LoggingMeta))# <class 'type'>
print(type(type))       # <class 'type'>

Solution 2 — PositiveInt descriptor

class PositiveInt:
    def __set_name__(self, owner, name):
        self.name = name
        self.private = f"_{name}"

    def __get__(self, obj, objtype=None):
        if obj is None:
            return self
        return getattr(obj, self.private, None)

    def __set__(self, obj, value):
        if not isinstance(value, int):
            raise TypeError(f"{self.name} must be int")
        if value <= 0:
            raise ValueError(f"{self.name} must be positive")
        setattr(obj, self.private, value)

class Product:
    price = PositiveInt()
    quantity = PositiveInt()

    def __init__(self, price, quantity):
        self.price = price
        self.quantity = quantity

    def total(self):
        return self.price * self.quantity

p = Product(100, 5)
print(p.total())     # 500
try:
    p.price = -1
except ValueError as e:
    print("Err:", e)

Solution 3 — __slots__ memory benchmark

import sys

class WithDict:
    def __init__(self, x, y):
        self.x = x
        self.y = y

class WithSlots:
    __slots__ = ("x", "y")
    def __init__(self, x, y):
        self.x = x
        self.y = y

N = 1_000_000

dict_objs  = [WithDict(i, i) for i in range(N)]
slots_objs = [WithSlots(i, i) for i in range(N)]

# Approximate memory per instance
print("dict-based (per instance):",
      sys.getsizeof(dict_objs[0]) + sys.getsizeof(dict_objs[0].__dict__))
print("slots-based (per instance):",
      sys.getsizeof(slots_objs[0]))

# Expected: ~152 B vs ~56 B per instance → ~100 MB saved at 1M objects

# Limitations:
# 1. No dynamic attributes outside the slots list.
# 2. Subclasses without __slots__ reintroduce __dict__ and lose savings.

Solution 4 — Reference cycles and GC

import gc, sys

class Node:
    def __init__(self, name):
        self.name = name
        self.peer = None

a = Node("A")
b = Node("B")
a.peer = b
b.peer = a            # cycle

print("Before del:", sys.getrefcount(a))
del a, b
# At this point both objects are unreachable but still alive because of the cycle.

collected = gc.collect()
print("Collected:", collected)   # 2 (or more)

Reference counting frees objects immediately when the count hits zero, but cannot free cycles. The generational GC periodically walks all objects tracked by the GC and breaks cycles by finding objects unreachable from any root.

Solution 5 — Profiling primes

import cProfile

def primes_naive(n):
    result = []
    for num in range(2, n):
        ok = True
        for d in range(2, num):
            if num % d == 0:
                ok = False
                break
        if ok:
            result.append(num)
    return result

def primes_optimised(n):
    result = []
    for num in range(2, n):
        if all(num % d for d in range(2, int(num ** 0.5) + 1)):
            result.append(num)
    return result

cProfile.run("primes_naive(20000)", sort="cumtime")
cProfile.run("primes_optimised(20000)", sort="cumtime")

# The optimised version only checks up to sqrt(n) and uses short-circuit all(),
# giving ~10× speedup at n=20000.

Solution 6 — Pure Python vs NumPy vs Numba

import numpy as np
import time
from numba import njit

N = 10_000_000

def pure(n):
    s = 0
    for i in range(n):
        s += i * i
    return s

def numpy_fn(n):
    arr = np.arange(n, dtype=np.int64)
    return int((arr * arr).sum())

@njit
def numba_fn(n):
    s = 0
    for i in range(n):
        s += i * i
    return s

numba_fn(1)          # warm-up

for label, fn in [("Pure", pure), ("NumPy", numpy_fn), ("Numba", numba_fn)]:
    t = time.perf_counter()
    fn(N)
    print(f"{label:<6}: {time.perf_counter() - t:.3f}s")

• Pure Python runs bytecode per iteration — ~1s.
• NumPy vectorises the operation in C — ~50 ms.
• Numba compiles the loop to native code — ~15 ms.

Solution 7 — Minimal Flask app

from flask import Flask, jsonify, request

app = Flask(__name__)

@app.route("/")
def home():
    return "Hello, Flask!"

@app.route("/user/<name>")
def user(name):
    return f"Hello, {name}!"

@app.route("/api/status")
def status():
    return jsonify({"status": "ok", "version": "1.0"})

@app.route binds a URL rule to a view function. Flask extracts path parameters by name and passes them as arguments.

Solution 8 — FastAPI notes CRUD

from fastapi import FastAPI, HTTPException, Depends, status
from pydantic import BaseModel
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, sessionmaker, Session

DATABASE_URL = "sqlite:///./notes.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine)
Base = declarative_base()

class Note(Base):
    __tablename__ = "notes"
    id    = Column(Integer, primary_key=True)
    title = Column(String, nullable=False)
    body  = Column(String, default="")

Base.metadata.create_all(engine)

class NoteIn(BaseModel):
    title: str
    body:  str = ""

class NoteOut(NoteIn):
    id: int
    class Config:
        from_attributes = True

app = FastAPI()

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

@app.post("/notes", response_model=NoteOut, status_code=201)
def create(payload: NoteIn, db: Session = Depends(get_db)):
    note = Note(**payload.model_dump())
    db.add(note); db.commit(); db.refresh(note)
    return note

@app.get("/notes/{note_id}", response_model=NoteOut)
def read(note_id: int, db: Session = Depends(get_db)):
    note = db.get(Note, note_id)
    if not note:
        raise HTTPException(404, "Not found")
    return note

@app.put("/notes/{note_id}", response_model=NoteOut)
def update(note_id: int, payload: NoteIn, db: Session = Depends(get_db)):
    note = db.get(Note, note_id)
    if not note:
        raise HTTPException(404, "Not found")
    note.title, note.body = payload.title, payload.body
    db.commit(); db.refresh(note)
    return note

@app.delete("/notes/{note_id}", status_code=204)
def delete(note_id: int, db: Session = Depends(get_db)):
    note = db.get(Note, note_id)
    if not note:
        raise HTTPException(404, "Not found")
    db.delete(note); db.commit()

Solution 9 — pytest for discount

import pytest

def calculate_discount(price, percent):
    if price < 0 or percent < 0:
        raise ValueError("Inputs must be non-negative")
    if percent > 100:
        raise ValueError("Percent must be <= 100")
    return price * (1 - percent / 100)

@pytest.fixture
def base_price():
    return 1000

@pytest.mark.parametrize("percent, expected", [
    (0,   1000),
    (10,  900),
    (25,  750),
    (100, 0),
])
def test_discount(base_price, percent, expected):
    assert calculate_discount(base_price, percent) == expected

def test_negative_price():
    with pytest.raises(ValueError, match="non-negative"):
        calculate_discount(-1, 10)

def test_percent_over_100():
    with pytest.raises(ValueError, match="<= 100"):
        calculate_discount(100, 150)

Solution 10 — Dockerfile + Compose

# Dockerfile
FROM python:3.12-slim AS builder
WORKDIR /app
COPY requirements.txt .
RUN pip install --user --no-cache-dir -r requirements.txt

FROM python:3.12-slim
WORKDIR /app
RUN useradd --create-home appuser
COPY --from=builder /root/.local /home/appuser/.local
COPY src/ ./src/
ENV PATH=/home/appuser/.local/bin:$PATH PYTHONUNBUFFERED=1
USER appuser
EXPOSE 8000
HEALTHCHECK --interval=30s CMD \
  python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"
CMD ["uvicorn", "src.main:app", "--host", "0.0.0.0", "--port", "8000"]

# docker-compose.yml
version: "3.9"
services:
  api:
    build: .
    ports: ["8000:8000"]
    environment:
      - DATABASE_URL=postgresql://user:pass@db:5432/mydb
    depends_on:
      db:
        condition: service_healthy
  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
      POSTGRES_DB: mydb
    volumes:
      - pgdata:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U user"]
      interval: 10s
volumes:
  pgdata:

Solution 11 — OWASP Top 10 with Python mitigations

Solution 12 — Structured logging + Prometheus metrics

import json, logging, time
from prometheus_client import Counter, Histogram, start_http_server

class JSONFormatter(logging.Formatter):
    def format(self, record):
        return json.dumps({
            "ts":    self.formatTime(record),
            "level": record.levelname,
            "msg":   record.getMessage(),
        })

logger = logging.getLogger("api")
logger.setLevel(logging.INFO)
h = logging.StreamHandler()
h.setFormatter(JSONFormatter())
logger.addHandler(h)

REQUESTS = Counter("requests_total", "Total requests", ["path"])
LATENCY  = Histogram("request_latency_seconds", "Latency", ["path"])

start_http_server(8000)

def handle(path):
    REQUESTS.labels(path).inc()
    with LATENCY.labels(path).time():
        time.sleep(0.05)              # simulated work
    logger.info(f"handled {path}")

for _ in range(20):
    handle("/api/data")

Solution 13 — Three pillars & endpoint design

from fastapi import FastAPI, Response, status
from prometheus_client import generate_latest, CONTENT_TYPE_LATEST

app = FastAPI()
_state = {"db": True}

@app.get("/health")
def health():
    return {"status": "ok"}

@app.get("/ready")
def ready(response: Response):
    if not all(_state.values()):
        response.status_code = status.HTTP_503_SERVICE_UNAVAILABLE
        return {"status": "not ready"}
    return {"status": "ready"}

@app.get("/metrics")
def metrics():
    return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)

Pillars: logs (structured JSON events), metrics (Prometheus counters/histograms), traces (OpenTelemetry spans). Each answers a distinct operational question.

Solution 14 — GitHub Actions CI/CD

# .github/workflows/ci-cd.yml
name: CI/CD
on:
  push: { branches: [main] }
  pull_request:

jobs:
  test:
    runs-on: ubuntu-latest
    strategy:
      matrix:
        python-version: ["3.10", "3.11", "3.12"]
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: ${{ matrix.python-version }}
          cache: pip
      - run: pip install -e ".[dev]"
      - run: ruff check .
      - run: mypy src/
      - run: pytest --cov=src --cov-fail-under=80

  build-and-push:
    needs: test
    if: github.ref == 'refs/heads/main'
    runs-on: ubuntu-latest
    permissions:
      contents: read
      packages: write
    steps:
      - uses: actions/checkout@v4
      - uses: docker/setup-buildx-action@v3
      - uses: docker/login-action@v3
        with:
          registry: ghcr.io
          username: ${{ github.actor }}
          password: ${{ secrets.GITHUB_TOKEN }}
      - uses: docker/build-push-action@v5
        with:
          push: true
          tags: ghcr.io/${{ github.repository }}:${{ github.sha }}

XVI. References, Key Takeaways & CO Mapping

16.1 Textbooks and References

16.2 Relevant Web Resources

16.3 Key Takeaways

1. Metaclasses are classes whose instances are classes; they let you hook into class creation to enforce rules, auto-register, and inject behaviour.
2. __init_subclass__ covers most metaclass use cases with less complexity — prefer it when possible.
3. Descriptors give fine-grained control over attribute access; property is the built-in data descriptor you use most often.
4. Data descriptors beat instance __dict__; non-data descriptors lose to it — this ordering enables cached-property patterns.
5. __slots__ eliminates per-instance dicts, reducing memory and speeding attribute access, at the cost of dynamic attributes.
6. CPython uses reference counting + generational GC. Use weakref to avoid cycles in caches, listeners and back-references.
7. Measure before optimising. timeit for micro-benchmarks, cProfile for functions, tracemalloc/memory_profiler for memory.
8. Escape hatches for CPU-bound code: NumPy (vectorise), Numba (JIT), Cython (compile), ctypes (call C), Rust (rewrite).
9. Flask is a micro-framework — small, explicit, great for APIs and prototypes; use Gunicorn/waitress in production.
10. FastAPI adds async, Pydantic validation, and auto-generated OpenAPI docs; it is the modern choice for new REST services.
11. Test with pytest: plain assert, fixtures, parametrisation, mocking external I/O, coverage gates in CI.
12. Docker packages apps reproducibly: multi-stage builds, non-root user, .dockerignore, healthchecks.
13. CI/CD runs linters, type checks and tests on every push, then builds and pushes images from main.
14. Security fundamentals: bcrypt/argon2 for passwords, JWT with short TTL, parameterised SQL, validated input, secrets in env vars, pip-audit in CI.
15. Observability has three pillars: metrics, logs, traces. Instrument with Prometheus, JSON logging, and OpenTelemetry.

16.4 CO Mapping

16.5 Learning Path

16.6 Further Reading