Python Engineering & Data Structures Masterclass

01 Language Fundamentals & Python's Memory Model

Execution Engine

In Python, everything is an object. Variables do not store raw values; they are typed pointers referencing objects allocated on the heap. Python utilizes reference counting paired with a generational cycle detector for garbage collection.

Object Identity vs Value Equality

Use == to evaluate equality of values and is to evaluate memory identity (pointing to the exact same pointer address returned by id()).

# Identity vs Equality
a = [1, 2, 3]
b = [1, 2, 3]
c = a

print(a == b)  # True  (equivalent values)
print(a is b)  # False (distinct memory addresses)
print(a is c)  # True  (c references the exact same list)

# Small integer caching & interned strings (-5 to 256)
x = 256
y = 256
print(x is y)  # True (CPython singletons cache)

Pattern Matching & Walrus Operator

Python 3.10+ provides structural pattern matching with match / case. The walrus operator (:=) assigns values to variables as part of a larger expression.

# 1. Structural Pattern Matching
def parse_agent_action(event: dict) -> str:
    match event:
        case {"type": "tool_call", "tool": name, "args": dict(a)}:
            return f"Executing {name} with args {a}"
        case {"type": "message", "content": str(text)} if len(text) > 0:
            return f"Delivered message: {text}"
        case _:
            return "Unknown event format"

# 2. Walrus Operator in evaluation guards
import re
payload = "Token ID: agent_9921_beta"
if match := re.search(r"agent_(\d+)", payload):
    print("Found Agent ID:", match.group(1))

02 Built-in Data Structures & Big-O Complexity

Time & Space Guarantees

Choosing the wrong data structure degrades algorithmic throughput from milliseconds to seconds. The table below details average and worst-case time complexities in CPython:

Data Structure	Index Access	Search / Lookup	Append / Prepend	Insert / Delete (Middle)	Space Complexity
list (Dynamic Array)	O(1)	O(n)	O(1) amortized / O(n) prepend	O(n)	O(n)
tuple (Immutable Sequence)	O(1)	O(n)	N/A (immutable)	N/A	O(n) (compact)
dict (Hash Map)	N/A	O(1) avg / O(n) worst	O(1) key set	O(1) avg key deletion	O(n)
set (Hash Set)	N/A	O(1) avg / O(n) worst	O(1) add	O(1) avg remove	O(n)
collections.deque	O(n)	O(n)	O(1) both ends	O(n)	O(n)

Lists & Comprehensions

Lists are dynamic arrays over-allocated to provide amortized O(1) appends. List comprehensions run in optimized C bytecode loops.

# Slicing: [start:stop:step]
nums = [10, 20, 30, 40, 50, 60]
print(nums[::-1])      # Reverse copy: [60, 50, 40, 30, 20, 10]
print(nums[1:4])       # Sub-slice: [20, 30, 40]

# List comprehension with filtering & mapping
squared_evens = [x**2 for x in nums if x % 20 == 0]
# Result: [400, 1600, 3600]

# Sorting: In-place sort vs new sorted sequence
nums.sort(reverse=True)       # O(n log n) Timsort in place
ranked = sorted(nums, key=lambda v: -v)

Hash Maps (Dicts) & Sets

Python dictionaries preserve insertion order (since 3.7). Keys must be hashable objects (implementing __hash__ and __eq__).

# Dictionary Merging (PEP 584)
base_config = {"env": "prod", "retries": 3}
override = {"retries": 5, "timeout_ms": 1200}
merged = base_config | override  # {'env': 'prod', 'retries': 5, 'timeout_ms': 1200}

# Dict Comprehension
id_map = {f"agent_{i}": f"role_{i}" for i in range(3)}

# Set Operations (Union, Intersection, Difference)
set_a = {"read", "write", "execute"}
set_b = {"write", "audit"}
print(set_a & set_b)  # Intersection: {'write'}
print(set_a - set_b)  # Difference: {'read', 'execute'}

03 Advanced Standard Library Collections & Heaps

Batteries Included

1. collections.deque

A double-ended queue implemented as a doubly-linked list of 64-element blocks. Guarantees O(1) append and pop from both ends. Essential for sliding windows and BFS graph traversal.

from collections import deque

# Bounded queue auto-discards old items
history = deque(maxlen=3)
history.append("prompt_1")
history.append("prompt_2")
history.append("prompt_3")
history.append("prompt_4")
print(list(history)) 
# ['prompt_2', 'prompt_3', 'prompt_4']

history.appendleft("urgent")
print(history.pop())  # O(1) pop right

2. defaultdict & Counter

Eliminate defensive if key not in d: d[key] = [] boilerplate. Counter tallies frequencies and provides most_common(k) in O(n log k).

from collections import defaultdict, Counter

# Grouping without KeyError
grouped = defaultdict(list)
grouped["coder"].append("task_101")
grouped["critic"].append("task_102")

# Frequency Counting
tokens = ["llm", "rag", "llm", "agent", "llm", "rag"]
counts = Counter(tokens)
print(counts.most_common(2))
# [('llm', 3), ('rag', 2)]

3. heapq (Min-Heap)

Binary heap implementation. Keeps the smallest element at index 0. Insertion and removal run in O(log n) time.

import heapq

# Priority Queue (priority, task_name)
pq = []
heapq.heappush(pq, (3, "background_indexing"))
heapq.heappush(pq, (1, "security_hotfix"))
heapq.heappush(pq, (2, "generate_patch"))

# Always extract highest priority (lowest rank number)
prio, task = heapq.heappop(pq)
print(task) # 'security_hotfix'

04 Functions, Scopes & Advanced Decorator Mechanics

Functional Architecture

Python resolves variable names using the LEGB Rule (Local → Enclosing → Global → Built-in). Functions are first-class objects that can be passed as arguments, assigned to variables, and wrapped with decorators.

Production Retry Decorator

Decorators intercept function execution. Always wrap inner callables with @functools.wraps to preserve function name, docstrings, and signatures for introspection.

import time
from functools import wraps

def retry(attempts=3, backoff_factor=1.5):
    """Parameterized decorator for retrying flaky network/LLM calls."""
    def decorator(fn):
        @wraps(fn)
        def wrapper(*args, **kwargs):
            delay = 0.5
            for i in range(1, attempts + 1):
                try:
                    return fn(*args, **kwargs)
                except Exception as err:
                    if i == attempts:
                        raise err
                    print(f"[{fn.__name__}] Retry {i}/{attempts} after error: {err}")
                    time.sleep(delay)
                    delay *= backoff_factor
        return wrapper
    return decorator

@retry(attempts=3, backoff_factor=2)
def fetch_embeddings(text: str) -> list[float]:
    # Network call simulated here
    return [0.012, 0.451, -0.219]

Closures & Memory-Efficient Generators

Generators use yield to produce values lazily on demand. This allows processing gigabyte-scale datasets or unbounded streaming tokens with O(1) memory overhead.

# 1. State-Preserving Closure
def token_rate_limiter(max_per_minute: int):
    tokens_left = max_per_minute
    def consume(amount: int) -> bool:
        nonlocal tokens_left
        if tokens_left >= amount:
            tokens_left -= amount
            return True
        return False
    return consume

limiter = token_rate_limiter(100)
print(limiter(30))  # True (70 left)

# 2. Generator Stream (Lazy Evaluation)
def stream_chunks(items: list, chunk_size: int):
    for i in range(0, len(items), chunk_size):
        yield items[i:i + chunk_size]

for batch in stream_chunks(range(1000), 200):
    print("Processing batch of length:", len(batch))

05 Object-Oriented Design: Dataclasses & Structural Protocols

Type-Safe Architecture

Modern Python OOP balances explicit typing and protocols. Using dataclasses(slots=True) avoids creating an internal __dict__ per instance, reducing memory consumption by up to 50% while accelerating attribute access.

Memory-Optimized Dataclasses

Declare immutable domain entities with frozen=True to make instances hashable and eligible for use as set elements or dictionary keys.

from dataclasses import dataclass, field
from datetime import datetime

@dataclass(slots=True, frozen=True)
class AgentStep:
    step_id: str
    role: str
    content: str
    timestamp: datetime = field(default_factory=datetime.utcnow)

    def summary(self) -> str:
        return f"[{self.role.upper()}]: {self.content[:40]}..."

step = AgentStep("s-101", "coder", "Synthesizing unit test suite")
print(step.summary())
# Immutable: step.role = 'critic' raises FrozenInstanceError

Static Duck Typing via Protocol

Rather than enforcing rigid subclass hierarchies via abc.ABC, typing.Protocol enables structural subtyping: any class that satisfies the method signatures is accepted.

from typing import Protocol, runtime_checkable

@runtime_checkable
class VectorStore(Protocol):
    def insert(self, key: str, vector: list[float]) -> None: ...
    def query(self, vector: list[float], top_k: int) -> list[str]: ...

class PineconeAdapter:
    def insert(self, key: str, vector: list[float]) -> None:
        print(f"Persisting {key} to Pinecone index.")

    def query(self, vector: list[float], top_k: int) -> list[str]:
        return ["doc_12", "doc_84"]

# Verification without explicit inheritance!
store: VectorStore = PineconeAdapter()
print(isinstance(store, VectorStore))  # True

06 Asynchronous Programming & Structured Concurrency

Async / Await & Event Loops

AI agents spend the majority of their wall-clock time waiting on network sockets (LLM API inferences, database queries, web scraping). asyncio runs cooperative multitasking on a single OS thread via non-blocking sockets.

Modern Structured Concurrency with TaskGroup (Python 3.11+)

asyncio.TaskGroup

Unlike older asyncio.gather(), asyncio.TaskGroup guarantees that if any concurrent sub-task raises an exception, all other ongoing tasks are immediately canceled, preventing orphaned zombie operations.

import asyncio

async def call_llm_worker(agent_id: str, prompt: str) -> str:
    print(f"[{agent_id}] Dispatching inference request...")
    await asyncio.sleep(0.1)  # Simulating network latency
    return f"{agent_id} completed: Evaluated '{prompt}'"

async def orchestrate_swarm():
    # Structured concurrency with TaskGroup
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(call_llm_worker("researcher", "Scan security CVEs"))
        t2 = tg.create_task(call_llm_worker("coder", "Generate SQL patch"))
        t3 = tg.create_task(call_llm_worker("critic", "Verify unit test coverage"))

    # When context exits, all tasks are guaranteed complete
    results = [t1.result(), t2.result(), t3.result()]
    for res in results:
        print("->", res)

if __name__ == "__main__":
    asyncio.run(orchestrate_swarm())

07 Real-World Patterns: Building Agent Tooling in Python

Production Engineering

Production agent frameworks (LangGraph, CrewAI, AutoGen, PydanticAI) rely on decorators, type inspection, and schema serialization to extract function definitions for Model Context Protocol (MCP) and JSON-RPC dispatch:

Autonomous Tool Registry Implementation

import inspect
from typing import Callable, Any

class AgentToolRegistry:
    def __init__(self):
        self._tools: dict[str, dict[str, Any]] = {}

    def register(self, description: str):
        """Decorator that extracts parameter types and registers executable tools."""
        def decorator(func: Callable):
            sig = inspect.signature(func)
            params = {
                param.name: param.annotation.__name__
                for param in sig.parameters.values()
            }
            self._tools[func.__name__] = {
                "name": func.__name__,
                "desc": description,
                "parameters": params,
                "callable": func
            }
            return func
        return decorator

    def execute(self, name: str, **kwargs) -> Any:
        if name not in self._tools:
            raise ValueError(f"Tool '{name}' not found in registry.")
        return self._tools[name]["callable"](**kwargs)

# Usage
registry = AgentToolRegistry()

@registry.register("Search technical documentation repository for semantic matches")
def search_docs(query: str, top_k: int = 3) -> list[str]:
    return [f"Match {i} for '{query}'" for i in range(1, top_k + 1)]

# Execution triggered by an LLM Tool Call:
output = registry.execute("search_docs", query="FastAPI checkpointers", top_k=2)
print("Agent Tool Output:", output)

08 Frequently Asked Technical Questions

Interview & Production Prep

What is the difference between a list and collections.deque in Python? →

A Python list is a contiguous dynamic array. Appending to the end is amortized O(1), but inserting or popping from the front is O(n) because all subsequent elements must be shifted in memory. In contrast, collections.deque is a doubly-linked list of blocks that provides guaranteed O(1) appends and pops from both ends, making it the ideal choice for queues and sliding windows.

How does the Global Interpreter Lock (GIL) affect Python concurrency? →

The GIL is a mutex that prevents multiple native OS threads from executing Python bytecodes simultaneously in CPython. For I/O-bound tasks (HTTP requests, LLM API calls, database operations), threads and asyncio yield execution while waiting for sockets, allowing thousands of concurrent requests. For CPU-bound tasks (matrix multiplications, image rendering), developers use multiprocessing, C/Rust extensions, or the new free-threaded mode introduced in Python 3.13 (PEP 703).

What are the benefits of using __slots__ in Python classes? →

By default, Python stores an object's instance attributes in a dynamic dictionary (__dict__). Adding __slots__ = ('attr1', 'attr2') (or using @dataclass(slots=True)) allocates a fixed array of pointers instead. This reduces memory usage by 40–50% when instantiating millions of records and speeds up attribute read/write operations by avoiding hash table lookups.

How does Python's garbage collector detect circular references? →

Python combines reference counting with a cyclic garbage collector. Reference counting instantly deallocates any object whose reference count drops to 0. To resolve cyclic references (e.g. Node A referencing Node B, and Node B referencing Node A), Python's cyclic GC periodically tracks container objects across 3 generational heuristics (Generation 0, 1, and 2), finding and breaking unreachable isolated reference loops.

Python Engineering, Data Structures & Agent Systems