Advanced Topics 

Asynchronous programming (asyncio, await)

 What async is and why it matters

Asynchronous programming is a technique for writing programs that can start work on an operation, then switch to do something else while that operation is waiting (I/O, timers, etc.). The goal is higher throughput and responsiveness for I/O-bound workloads (network requests, file I/O, many concurrent sockets) without the overhead of many OS threads.

Key ideas:

• Concurrency vs parallelism: Concurrency is managing multiple tasks that appear to run at the same time; parallelism runs tasks simultaneously on multiple CPUs. Async provides concurrency for I/O-bound tasks; it does not bypass the GIL for CPU-bound work.
• Non-blocking I/O: Async code yields control when waiting, allowing the event loop to run other tasks.
• Cooperative multitasking: Async tasks yield explicitly (via await) rather than being preempted by the scheduler.
• Single-threaded, high concurrency: An async program often runs a single event loop thread that multiplexes thousands of coroutines efficiently.
• Use cases: web servers (FastAPI, aiohttp), scraping many URLs, multiplexed sockets, real-time systems, lightweight background tasks.

Runtime model (Python asyncio):

• Coroutines (declared with async def) are awaitable objects.
• await suspends a coroutine until an awaitable completes (another coroutine, Future, Task, I/O operation).
• Event loop drives execution: schedules tasks, runs callbacks, polls I/O.
• Tasks wrap coroutines, scheduled to run on the loop (via asyncio.create_task).
• Futures represent a result that will be available later.

Tradeoffs:

• Async code requires async libraries (e.g., aiohttp instead of requests).
• Debugging stack traces across awaits can be slightly different.
• Mixing sync and async code requires care (blocking code must be run in executor).

 Basic asyncio examples

Example 1: Simple concurrent HTTP fetch (using aiohttp)

# requires: pip install aiohttp
import asyncio
import aiohttp
import time

URLS = [
    "https://example.com",
    "https://httpbin.org/get",
    "https://httpbin.org/delay/1",
]

async def fetch(session, url):
    async with session.get(url) as resp:
        text = await resp.text()
        return url, resp.status, len(text)

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [asyncio.create_task(fetch(session, u)) for u in URLS]
        for task in asyncio.as_completed(tasks):
            url, status, length = await task
            print(f"{url} -> {status}, {length} bytes")

if __name__ == "__main__":
    start = time.time()
    asyncio.run(main())
    print("Elapsed:", time.time() - start)

Notes:

• aiohttp is async; requests is blocking — do not use requests inside async code.
• asyncio.as_completed yields tasks as they finish; gather collects in original order.

Example 2: Using asyncio.gather with timeouts and cancellation

import asyncio

async def slow(n):
    await asyncio.sleep(n)
    return n

async def main():
    try:
        results = await asyncio.wait_for(asyncio.gather(slow(1), slow(3)), timeout=2.0)
        print("Results", results)
    except asyncio.TimeoutError:
        print("Timed out - cancelling tasks")
        # tasks created by gather are cancelled automatically on timeout if not awaited further

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

Example 3: Running blocking code in an executor

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor

def blocking_task(x):
    time.sleep(2)
    return x * 2

async def run_blocking(loop):
    executor = ThreadPoolExecutor()
    result = await loop.run_in_executor(executor, blocking_task, 10)
    print("Result from blocking:", result)

async def main():
    loop = asyncio.get_running_loop()
    await run_blocking(loop)

asyncio.run(main())

Use run_in_executor when you must call blocking libraries (e.g., CPU work or blocking DB drivers).

Coroutines and event loops

Coroutines, Tasks, Futures and the Event Loop

• Coroutine: An async def function returns a coroutine object when called. The coroutine must be scheduled for execution by an event loop (via await or asyncio.create_task).
• Task: A Task is a wrapper around a coroutine that is scheduled on the loop and drives execution. Tasks are subclasses of Future.
• Future: A low-level object representing a result not yet available; tasks are higher-level constructs built on futures.
• Event loop: Single object managing I/O, scheduled callbacks, timers. You usually run a loop via asyncio.run() (which creates and closes the loop). In advanced contexts (embedding or library code), you get the loop via asyncio.get_running_loop().

Event loop operations:

• Polling OS-level I/O (sockets) using selectors (select, epoll, kqueue)
• Scheduling timers and callbacks
• Running tasks and coroutines
• Handling cancellation and exceptions

Important patterns:

• Producer/consumer with asyncio.Queue (async-safe)
• Cancellation flows: when cancelling tasks, the coroutine receives asyncio.CancelledError; ensure cleanup in finally.
• Backpressure: use bounded queues to prevent unbounded memory growth.

Patterns and examples

Example 1: Producer/consumer with asyncio.Queue

import asyncio
import random

async def producer(q, n):
    for i in range(n):
        await asyncio.sleep(random.random() * 0.2)
        item = f"item-{i}"
        await q.put(item)
        print("Produced", item)
    await q.put(None)  # sentinel to signal completion

async def consumer(q):
    while True:
        item = await q.get()
        if item is None:
            q.task_done()
            break
        print("Consumed", item)
        q.task_done()

async def main():
    q = asyncio.Queue()
    await asyncio.gather(producer(q, 10), consumer(q))
    await q.join()

asyncio.run(main())

Example 2: Proper cancellation and cleanup

import asyncio

async def worker():
    try:
        while True:
            print("Working...")
            await asyncio.sleep(1)
    except asyncio.CancelledError:
        print("Worker was cancelled, cleaning up")
        # do cleanup here
        raise

async def main():
    t = asyncio.create_task(worker())
    await asyncio.sleep(3)
    t.cancel()
    try:
        await t
    except asyncio.CancelledError:
        print("Task confirmed cancelled")

asyncio.run(main())

Memory profiling

 Why profile memory, what to measure

Memory profiling helps find leaks, excessive allocations, and high memory usage. Key metrics:

• Resident Set Size (RSS): OS memory occupied by the process.
• Heap allocations: Python object allocations tracked by tracemalloc.
• Object lifetimes: Which objects are kept alive unexpectedly (reference cycles or global caches).
• Peak memory usage: Useful in batch processing.

Tools and approaches:

• tracemalloc (built-in): tracks allocations, can show top memory-consuming lines and compare snapshots.
• objgraph: visualize object graphs and find referents.
• Heapy / guppy: heap analysis.
• Memory profilers: memory_profiler (line-by-line memory usage using decorator @profile), psutil (process memory), or OS tools (top, htop).
• Sampling vs tracing: tracemalloc traces allocations (higher overhead), sampling profilers sample RSS over time (lower overhead).

Common causes of leaks:

• Global caches that grow without bounds (list, dict)
• Circular references with objects that implement __del__
• References held in closures, frames, or third-party libraries

 Using tracemalloc and memory_profiler

Example 1: tracemalloc basic usage

import tracemalloc

def make_data(n):
    return [list(range(1000)) for _ in range(n)]

tracemalloc.start()

snap1 = tracemalloc.take_snapshot()
data = make_data(50)
snap2 = tracemalloc.take_snapshot()

top_stats = snap2.compare_to(snap1, 'lineno')
for stat in top_stats[:10]:
    print(stat)

This prints allocation differences by file/line.

Example 2: memory_profiler line-by-line

Install: pip install memory_profiler psutil and run via mprof or use decorator.

# save as memtest.py
from memory_profiler import profile

@profile
def f():
    a = [0] * (10**6)   # allocate ~8 MB
    b = [0] * (2 * 10**6)
    del b
    return a

if __name__ == "__main__":
    f()

Run:

python -m memory_profiler memtest.py

It prints memory usage per line.

Example 3: using psutil to monitor RSS over time

import psutil, time, os

pid = os.getpid()
p = psutil.Process(pid)

for i in range(5):
    print("RSS:", p.memory_info().rss)
    time.sleep(1)

Performance optimization techniques

 Principles and workflow

Performance optimization should follow measurement: measure → identify hotspot → optimize → measure again.

Categories of optimization:

• Algorithmic improvements: change O(n^2) to O(n log n).
• Data structure choices: use set for membership, deque for append/pop left.
• Avoiding repeated work: memoization (functools.lru_cache), caching results.
• I/O batching: reduce syscalls by batching writes/reads.
• Vectorized operations: use NumPy/Pandas for numeric workloads.
• Concurrency: threads for I/O-bound, processes for CPU-bound (multiprocessing or process pools).
• Minimize allocations: reuse objects or preallocate lists.
• Avoiding Python-level loops: move heavy loops to C extensions or use builtins; list comprehensions are often faster than loops.

Tools:

• cProfile / profile: CPU profiler.
• py-spy / vprof / scalene: sampling profilers with low overhead.
• line_profiler for line-level timing.
• timeit for microbenchmarks.

Optimization tradeoffs:

• Readability vs speed: prefer clarity unless performance is critical.
• Memory vs speed: caching speeds up but uses memory.
• Complexity vs maintainability.

Examples of common optimizations

Example 1: Replace list in checks with set

# slow
def find_common(a, b):
    res = []
    for x in a:
        if x in b:
            res.append(x)
    return res

# faster: convert b to set
def find_common_fast(a, b):
    bset = set(b)
    return [x for x in a if x in bset]

Example 2: Use itertools for efficient iteration

from itertools import islice

# process in chunks from iterator
def chunked(iterable, size):
    it = iter(iterable)
    while True:
        chunk = list(islice(it, size))
        if not chunk:
            break
        yield chunk

Example 3: Use lru_cache for expensive pure functions

from functools import lru_cache
@lru_cache(maxsize=1024)
def fib(n):
    if n < 2:
        return n
    return fib(n-1) + fib(n-2)

Example 4: Use comprehension vs append in loop

# slower
res = []
for i in range(1000):
    res.append(i*i)

# faster
res = [i*i for i in range(1000)]

Example 5: Profile with cProfile and view top callers

import cProfile, pstats

def main():
    # code to profile
    pass

if __name__ == "__main__":
    profiler = cProfile.Profile()
    profiler.runcall(main)
    stats = pstats.Stats(profiler).sort_stats('cumtime')
    stats.print_stats(20)

Packaging Python projects (setuptools, pip)

Packaging fundamentals

Packaging lets you distribute Python code as installable artifacts. Essential components:

• Source distribution (sdist): a tarball of your source code.
• Wheel: built, binary distribution format — faster installs and preferred.
• Metadata: package name, version, dependencies (install_requires), classifiers.
• Entry points: console scripts for command-line tools.
• Dependency management: requirements.txt, pyproject.toml (newer standard), setup.cfg.
• Building tools: historically setuptools, now often build (python -m build) with pyproject.toml specifying build backend (setuptools or flit, poetry).
• Publishing: upload to PyPI using twine.
• Virtual environments: always test installs in clean venv.

Best practice: use pyproject.toml to define build-system and metadata (PEP 517/518). Keep code readable and tests in package.

 Minimal packaging with setuptools and pyproject.toml

Project layout

myproject/
  src/
    mypackage/
      __init__.py
      cli.py
  tests/
  pyproject.toml
  setup.cfg
  README.md

pyproject.toml (build backend)

[build-system]
requires = ["setuptools>=61.0", "wheel"]
build-backend = "setuptools.build_meta"

setup.cfg (metadata + options)

[metadata]
name = mypackage
version = 0.1.0
description = Example package
long_description = file: README.md
long_description_content_type = text/markdown
author = Your Name
license = MIT
classifiers =
    Programming Language :: Python :: 3
    License :: OSI Approved :: MIT License

[options]
package_dir =
    = src
packages = find:
install_requires =
    requests

[options.entry_points]
console_scripts =
    mycli = mypackage.cli:main

src/mypackage/cli.py

def main():
    print("Hello from mycli")

Build and install locally

python -m pip install --upgrade build twine
python -m build            # creates dist/*.whl and sdist
python -m pip install dist/mypackage-0.1.0-py3-none-any.whl

Publish to PyPI (test first)

python -m twine upload --repository testpypi dist/*
# then to real PyPI after verification

Notes:

• Use setuptools.find_packages() or packages = find: in setup.cfg when using src/ layout.
• For modern workflows, tools like Poetry handle dependency and packaging, but understanding setuptools is essential.

Python internals (bytecode, CPython)

CPython architecture and bytecode

CPython is the canonical Python implementation written in C. Key internals:

• Interpreter loop: CPython compiles source to bytecode (.pyc) then interprets bytecode in a virtual machine (evaluation loop).
• Bytecode: low-level operations such as LOAD_FAST, STORE_FAST, BINARY_ADD. Bytecode is what the interpreter executes.
• Frame objects: executed contexts containing local/global variables, instruction pointer, and evaluation stack.
• Reference counting + GC: CPython uses ref-counting for immediate deallocation plus a generational cyclic garbage collector for cycles.
• Object model: every value is a PyObject* (pointer to C struct) with type and reference count. Types define behavior (methods, descriptors).
• Method resolution and attribute lookup: attribute access uses descriptors (__get__, __set__) and type dictionaries; method lookup can invoke descriptors that return bound methods.
• GIL (Global Interpreter Lock): CPython’s mutex preventing multiple threads from executing Python bytecode concurrently. Makes single-threaded CPython safe for certain object models but restricts CPU-bound threading.

Why understanding bytecode helps:

• Helps reason about performance (e.g., attribute access vs local variable).
• Understand how constructs compile (list comp, generator).
• Build tools that manipulate or generate code (e.g., decorators, AST transforms).

Inspecting bytecode and small CPython experiments

Example 1: Inspect bytecode with dis

import dis

def f(x, y):
    z = x + y
    return z * 2

dis.dis(f)

Sample output shows instructions and stack effects. Use this to see what Python actually does.

Example 2: Timing local variable vs attribute access

import timeit

setup = "class A: pass\nobj = A(); obj.x = 1\n"
stmt_attr = "obj.x"
stmt_local = "x = obj.x; x"

print("attr:", timeit.timeit(stmt_attr, setup=setup, number=1000000))
print("local:", timeit.timeit(stmt_local, setup=setup, number=1000000))

You’ll typically see local variable access is faster than attribute access.

Example 3: Inspect .__code__ object

def f(a, b=2):
    return a + b

print(f.__code__.co_varnames)
print("argcount:", f.__code__.co_argcount)

__code__ contains bytecode, variable names, argcounts, constants — useful for metaprogramming.

Final notes and recommended follow-ups

These advanced topics are broad. For practical learning path:

• Build a small async app (e.g., aiohttp crawler) to internalize asyncio patterns.
• Profile memory and CPU of real scripts — practice reading profiler output.
• Optimize measured hotspots with algorithmic and micro-optimizations.
• Package a toy CLI app and publish to Test PyPI.
• Inspect bytecode for code you optimize frequently.

Next Blog- Python in AI