Professional Python
Patterns, Concurrency & Applications

Table of Contents

I. Design Patterns in Python

1.1 Categories of Patterns

1.2 Singleton — Exactly One Instance

class ConfigManager:
    _instance = None

    def __new__(cls, *args, **kwargs):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance._initialized = False
        return cls._instance

    def __init__(self):
        if self._initialized:
            return
        self.settings = {"theme": "light", "lang": "en"}
        self._initialized = True

a = ConfigManager()
b = ConfigManager()

print(a is b)                    # True — same object
a.settings["theme"] = "dark"
print(b.settings["theme"])       # dark

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 Database(metaclass=SingletonMeta):
    def __init__(self):
        self.connection = "db://connected"

d1 = Database()
d2 = Database()
print(d1 is d2)      # True

1.3 Factory — Delegate Object Creation

from abc import ABC, abstractmethod

class Shape(ABC):
    @abstractmethod
    def area(self): pass

class Circle(Shape):
    def __init__(self, r): self.r = r
    def area(self): return 3.14159 * self.r ** 2

class Rectangle(Shape):
    def __init__(self, w, h): self.w, self.h = w, h
    def area(self): return self.w * self.h

class Triangle(Shape):
    def __init__(self, b, h): self.b, self.h = b, h
    def area(self): return 0.5 * self.b * self.h

class ShapeFactory:
    @staticmethod
    def create(kind, *args):
        registry = {
            "circle":    Circle,
            "rectangle": Rectangle,
            "triangle":  Triangle,
        }
        cls = registry.get(kind.lower())
        if cls is None:
            raise ValueError(f"Unknown shape: {kind}")
        return cls(*args)

for kind, args in [("circle", (5,)), ("rectangle", (4, 6)), ("triangle", (3, 8))]:
    shape = ShapeFactory.create(kind, *args)
    print(f"{kind:<10} area = {shape.area():.2f}")

circle     area = 78.54
rectangle  area = 24.00
triangle   area = 12.00

1.4 Observer — Publish/Subscribe

class Subject:
    def __init__(self):
        self._observers = []

    def attach(self, observer):
        self._observers.append(observer)

    def detach(self, observer):
        self._observers.remove(observer)

    def notify(self, event):
        for obs in self._observers:
            obs.update(event)

class EmailNotifier:
    def update(self, event):
        print(f"[EMAIL]   {event}")

class SMSNotifier:
    def update(self, event):
        print(f"[SMS]     {event}")

class LogNotifier:
    def update(self, event):
        print(f"[LOG]     {event}")

subject = Subject()
subject.attach(EmailNotifier())
subject.attach(SMSNotifier())
subject.attach(LogNotifier())

subject.notify("Order #101 shipped")

[EMAIL]   Order #101 shipped
[SMS]     Order #101 shipped
[LOG]     Order #101 shipped

1.5 Strategy — Pluggable Algorithms

class BubbleSort:
    def sort(self, data):
        arr = data[:]
        n = len(arr)
        for i in range(n - 1):
            for j in range(n - 1 - i):
                if arr[j] > arr[j + 1]:
                    arr[j], arr[j + 1] = arr[j + 1], arr[j]
        return arr

class QuickSort:
    def sort(self, data):
        if len(data) <= 1:
            return data
        pivot = data[len(data) // 2]
        left  = [x for x in data if x < pivot]
        mid   = [x for x in data if x == pivot]
        right = [x for x in data if x > pivot]
        return self.sort(left) + mid + self.sort(right)

class BuiltinSort:
    def sort(self, data):
        return sorted(data)

class Sorter:
    def __init__(self, strategy):
        self.strategy = strategy

    def set_strategy(self, strategy):
        self.strategy = strategy

    def sort(self, data):
        return self.strategy.sort(data)

data = [38, 27, 43, 3, 9, 82, 10]

sorter = Sorter(BubbleSort())
print("Bubble  :", sorter.sort(data))

sorter.set_strategy(QuickSort())
print("Quick   :", sorter.sort(data))

sorter.set_strategy(BuiltinSort())
print("Built-in:", sorter.sort(data))

1.6 Pattern Comparison

II. Multithreading & Concurrency

2.1 Concurrency vs Parallelism

2.2 The Global Interpreter Lock (GIL)

2.3 Creating Threads

import threading
import time

def worker(name, delay):
    for i in range(3):
        print(f"{name}: step {i}")
        time.sleep(delay)

# Approach 1: pass a function
t1 = threading.Thread(target=worker, args=("Alpha", 0.2))
t2 = threading.Thread(target=worker, args=("Beta",  0.3))

t1.start()
t2.start()

t1.join()          # wait for t1 to finish
t2.join()
print("All threads done.")

Approach 2: subclass Thread

class MyThread(threading.Thread):
    def __init__(self, name, delay):
        super().__init__(name=name)
        self.delay = delay

    def run(self):
        for i in range(3):
            print(f"{self.name}: step {i}")
            time.sleep(self.delay)

t = MyThread("Gamma", 0.2)
t.start()
t.join()

2.4 Thread Synchronisation

When two threads access shared data, race conditions can occur. Python provides several synchronisation primitives.

import threading

counter = 0
lock = threading.Lock()

def unsafe_increment():
    global counter
    for _ in range(100_000):
        counter += 1          # NOT atomic — read, add, write

def safe_increment():
    global counter
    for _ in range(100_000):
        with lock:            # acquire and release automatically
            counter += 1

# Unsafe version — result is usually wrong
counter = 0
threads = [threading.Thread(target=unsafe_increment) for _ in range(5)]
for t in threads: t.start()
for t in threads: t.join()
print("Unsafe counter:", counter)      # often < 500000

# Safe version — correct
counter = 0
threads = [threading.Thread(target=safe_increment) for _ in range(5)]
for t in threads: t.start()
for t in threads: t.join()
print("Safe counter  :", counter)      # exactly 500000

2.5 Producer–Consumer with queue.Queue

import threading
import queue
import time
import random

def producer(q, n):
    for i in range(n):
        item = f"item-{i}"
        q.put(item)
        print(f"Produced {item}")
        time.sleep(random.uniform(0.05, 0.15))
    q.put(None)                # sentinel to stop consumers

def consumer(q, name):
    while True:
        item = q.get()
        if item is None:
            q.put(None)        # propagate sentinel to other consumers
            break
        print(f"{name} consumed {item}")
        q.task_done()

q = queue.Queue(maxsize=5)     # bounded queue → backpressure

prod = threading.Thread(target=producer, args=(q, 6))
cons1 = threading.Thread(target=consumer, args=(q, "C1"))
cons2 = threading.Thread(target=consumer, args=(q, "C2"))

prod.start(); cons1.start(); cons2.start()
prod.join(); cons1.join(); cons2.join()
print("Done.")

2.6 Thread Pools with concurrent.futures

Managing threads manually is tedious. ThreadPoolExecutor handles the pool lifecycle for you.

from concurrent.futures import ThreadPoolExecutor, as_completed
import time

def fetch(url, delay):
    time.sleep(delay)               # simulate network call
    return f"fetched {url} in {delay:.1f}s"

urls = [
    ("https://a.com", 0.5),
    ("https://b.com", 0.4),
    ("https://c.com", 0.6),
    ("https://d.com", 0.3),
]

with ThreadPoolExecutor(max_workers=4) as executor:
    futures = {executor.submit(fetch, url, delay): url
               for url, delay in urls}
    for future in as_completed(futures):
        print(future.result())

# Total time ≈ max(delay) = 0.6s instead of sum = 1.8s

2.7 Daemon Threads

import threading
import time

def background():
    while True:
        print("heartbeat...")
        time.sleep(1)

t = threading.Thread(target=background, daemon=True)
t.start()

time.sleep(3)
print("Main thread exiting — daemon will be killed automatically.")

2.8 Common Pitfalls

III. Multiprocessing & Parallelism

3.1 Why Multiprocessing?

The GIL prevents threads from running Python bytecode in parallel. Multiprocessing sidesteps this by spawning separate OS processes, each with its own Python interpreter and memory space — enabling true multi-core execution.

3.2 Creating Processes

from multiprocessing import Process
import os

def worker(name):
    print(f"Process {name} (PID {os.getpid()}) running")

if __name__ == "__main__":           # REQUIRED on Windows/macOS
    procs = [Process(target=worker, args=(f"W{i}",)) for i in range(4)]
    for p in procs: p.start()
    for p in procs: p.join()
    print("Main done.")

3.3 Process Pools

from multiprocessing import Pool
import time

def slow_square(n):
    total = 0
    for i in range(1, n + 1):
        total += i ** 2
    return total

numbers = [1_000_000, 2_000_000, 3_000_000, 4_000_000]

if __name__ == "__main__":
    # Sequential
    start = time.perf_counter()
    seq = [slow_square(n) for n in numbers]
    print(f"Sequential: {time.perf_counter() - start:.2f}s")

    # Parallel with a pool of 4 workers
    start = time.perf_counter()
    with Pool(processes=4) as pool:
        par = pool.map(slow_square, numbers)
    print(f"Parallel  : {time.perf_counter() - start:.2f}s")

    print("Results match:", seq == par)

3.4 Inter-Process Communication

(a) Queue

from multiprocessing import Process, Queue

def producer(q):
    for i in range(5):
        q.put(i)
    q.put(None)                     # sentinel

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

if __name__ == "__main__":
    q = Queue()
    p1 = Process(target=producer, args=(q,))
    p2 = Process(target=consumer, args=(q,))
    p1.start(); p2.start()
    p1.join();  p2.join()

(b) Pipe — bidirectional, two endpoints

from multiprocessing import Process, Pipe

def child(conn):
    conn.send("hello from child")
    print("Child received:", conn.recv())
    conn.close()

if __name__ == "__main__":
    parent_conn, child_conn = Pipe()
    p = Process(target=child, args=(child_conn,))
    p.start()
    print("Parent received:", parent_conn.recv())
    parent_conn.send("hello from parent")
    p.join()

(c) Shared memory

from multiprocessing import Process, Value, Array

def increment(val, arr):
    with val.get_lock():
        val.value += 1
    for i in range(len(arr)):
        arr[i] += 1

if __name__ == "__main__":
    counter = Value('i', 0)          # shared integer
    data    = Array('i', [0, 0, 0])  # shared array

    procs = [Process(target=increment, args=(counter, data)) for _ in range(3)]
    for p in procs: p.start()
    for p in procs: p.join()

    print("Counter:", counter.value)     # 3
    print("Array  :", list(data))        # [3, 3, 3]

3.5 concurrent.futures.ProcessPoolExecutor

The modern, high-level API — analogous to ThreadPoolExecutor.

from concurrent.futures import ProcessPoolExecutor
import math

def is_prime(n):
    if n < 2:
        return False
    for i in range(2, int(math.sqrt(n)) + 1):
        if n % i == 0:
            return False
    return True

if __name__ == "__main__":
    candidates = [10_000_019, 10_000_079, 10_000_103, 10_000_109]

    with ProcessPoolExecutor(max_workers=4) as executor:
        results = list(executor.map(is_prime, candidates))

    for n, prime in zip(candidates, results):
        print(f"{n} → {'prime' if prime else 'composite'}")

3.6 When to Use What

IV. Asynchronous Programming with asyncio

4.1 Synchronous vs Asynchronous

4.2 Coroutines — async def and await

import asyncio

async def greet(name, delay):
    await asyncio.sleep(delay)          # non-blocking sleep
    print(f"Hello, {name}!")
    return len(name)

async def main():
    # Sequential
    a = await greet("Aarav", 1)
    b = await greet("Diya",  1)
    print(a, b)

asyncio.run(main())

4.3 Running Coroutines Concurrently

import asyncio
import time

async def fetch(url, delay):
    await asyncio.sleep(delay)
    return f"{url} (took {delay}s)"

async def main():
    urls = [("a.com", 1), ("b.com", 2), ("c.com", 1.5)]

    start = time.perf_counter()

    # Sequential await
    seq = []
    for url, d in urls:
        seq.append(await fetch(url, d))
    print(f"Sequential: {time.perf_counter() - start:.2f}s")

    # Concurrent gather
    start = time.perf_counter()
    results = await asyncio.gather(*(fetch(u, d) for u, d in urls))
    print(f"Concurrent: {time.perf_counter() - start:.2f}s")
    for r in results:
        print(" ", r)

asyncio.run(main())

# Sequential: 4.50s
# Concurrent: 2.00s

4.4 Tasks and create_task

import asyncio

async def worker(name, delay):
    print(f"{name} starting")
    await asyncio.sleep(delay)
    print(f"{name} finished")
    return name

async def main():
    # Create tasks (they start running immediately)
    t1 = asyncio.create_task(worker("A", 1))
    t2 = asyncio.create_task(worker("B", 2))

    # Wait for both
    results = await asyncio.gather(t1, t2)
    print("Results:", results)

asyncio.run(main())

4.5 Timeouts and Cancellation

import asyncio

async def slow():
    await asyncio.sleep(10)
    return "done"

async def main():
    try:
        # Timeout after 2 seconds
        result = await asyncio.wait_for(slow(), timeout=2)
        print(result)
    except asyncio.TimeoutError:
        print("Timed out!")

asyncio.run(main())

4.6 Async Iteration and Async Context Managers

Async iteration

import asyncio

class AsyncCounter:
    def __init__(self, limit):
        self.limit = limit
        self.current = 0

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self.current >= self.limit:
            raise StopAsyncIteration
        await asyncio.sleep(0.2)
        self.current += 1
        return self.current

async def main():
    async for n in AsyncCounter(5):
        print(n, end=" ")
    print()

asyncio.run(main())
# 1 2 3 4 5

Async context manager

import asyncio

class AsyncDB:
    async def __aenter__(self):
        print("Connecting...")
        await asyncio.sleep(0.5)
        return self

    async def __aexit__(self, *exc):
        print("Disconnecting...")
        await asyncio.sleep(0.2)

    async def query(self, sql):
        await asyncio.sleep(0.3)
        return f"Result of: {sql}"

async def main():
    async with AsyncDB() as db:
        result = await db.query("SELECT * FROM students")
        print(result)

asyncio.run(main())

4.7 Async Queues — Producer/Consumer

import asyncio
import random

async def producer(q, n):
    for i in range(n):
        await asyncio.sleep(random.uniform(0.1, 0.4))
        await q.put(f"item-{i}")
        print(f"Produced item-{i}")
    await q.put(None)

async def consumer(q, name):
    while True:
        item = await q.get()
        if item is None:
            await q.put(None)
            break
        print(f"{name} got {item}")
        await asyncio.sleep(random.uniform(0.1, 0.3))

async def main():
    q = asyncio.Queue()
    await asyncio.gather(
        producer(q, 6),
        consumer(q, "C1"),
        consumer(q, "C2"),
    )

asyncio.run(main())

4.8 Common asyncio Functions

4.9 Choosing a Concurrency Model

V. Networking — Sockets & Protocols

5.1 What is a Socket?

5.2 TCP vs UDP

5.3 TCP Server and Client

# server.py
import socket

HOST, PORT = "127.0.0.1", 65432

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server:
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind((HOST, PORT))
    server.listen(5)
    print(f"Server listening on {HOST}:{PORT}")

    while True:
        conn, addr = server.accept()
        with conn:
            print(f"Connected by {addr}")
            while True:
                data = conn.recv(1024)
                if not data:
                    break
                print(f"Received: {data.decode()}")
                conn.sendall(b"Echo: " + data)

# client.py
import socket

HOST, PORT = "127.0.0.1", 65432

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as client:
    client.connect((HOST, PORT))
    for message in ["hello", "how are you?", "bye"]:
        client.sendall(message.encode())
        reply = client.recv(1024)
        print("Server:", reply.decode())

5.4 UDP Server and Client

# udp_server.py
import socket

with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as server:
    server.bind(("127.0.0.1", 9999))
    print("UDP server ready")

    while True:
        data, addr = server.recvfrom(1024)
        print(f"From {addr}: {data.decode()}")
        server.sendto(b"ACK: " + data, addr)

# udp_client.py
import socket

with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as client:
    client.sendto(b"ping", ("127.0.0.1", 9999))
    reply, _ = client.recvfrom(1024)
    print("Reply:", reply.decode())

5.5 Concurrency in Servers

A single-threaded server can handle only one client at a time. Use threads or asyncio for concurrent clients.

import socket
import threading

def handle(conn, addr):
    with conn:
        print(f"[{addr}] connected")
        while True:
            data = conn.recv(1024)
            if not data:
                break
            conn.sendall(b"Echo: " + data)
        print(f"[{addr}] disconnected")

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server:
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind(("127.0.0.1", 65432))
    server.listen()

    while True:
        conn, addr = server.accept()
        threading.Thread(target=handle, args=(conn, addr), daemon=True).start()

5.6 Common Socket Methods

5.7 DNS Resolution and IP Lookups

import socket

# Resolve a hostname to an IP
print(socket.gethostbyname("www.google.com"))     # e.g., 142.250.183.4

# Reverse lookup
print(socket.gethostbyaddr("8.8.8.8"))            # ('dns.google', ...)

# Get local hostname and IP
print(socket.gethostname())
print(socket.gethostbyname(socket.gethostname()))

VI. HTTP, REST APIs & the requests Library

6.1 HTTP Basics

HTTP (Hypertext Transfer Protocol) is the foundation of data exchange on the web. A client sends a request; the server sends a response.

6.2 The requests Library

pip install requests

import requests

response = requests.get(
    "https://api.github.com/search/repositories",
    params={"q": "python", "sort": "stars", "per_page": 3},
    timeout=10,
)

print("Status:", response.status_code)
data = response.json()              # parse JSON body

for repo in data["items"]:
    print(f"{repo['full_name']:<40} ⭐ {repo['stargazers_count']}")

import requests

payload = {"title": "New Post", "body": "Hello, world!", "userId": 1}

response = requests.post(
    "https://jsonplaceholder.typicode.com/posts",
    json=payload,
    timeout=10,
)

print(response.status_code)         # 201
print(response.json())              # the created post

6.3 Headers, Authentication and Sessions

import requests

# Custom headers
headers = {"User-Agent": "MyApp/1.0", "Accept": "application/json"}
response = requests.get("https://api.example.com/data", headers=headers)

# Bearer token authentication
headers = {"Authorization": "Bearer YOUR_TOKEN_HERE"}
response = requests.get("https://api.example.com/me", headers=headers)

# Basic authentication
response = requests.get("https://api.example.com/private",
                        auth=("username", "password"))

# Reusing a session (keeps cookies, connection pool)
with requests.Session() as session:
    session.headers.update({"User-Agent": "MyApp/1.0"})
    r1 = session.get("https://httpbin.org/cookies/set/user/aarav")
    r2 = session.get("https://httpbin.org/cookies")
    print(r2.json())

6.4 Robust Error Handling

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def make_session(retries=3, backoff=0.5):
    session = requests.Session()
    retry = Retry(
        total=retries,
        backoff_factor=backoff,
        status_forcelist=[429, 500, 502, 503, 504],
    )
    adapter = HTTPAdapter(max_retries=retry)
    session.mount("https://", adapter)
    session.mount("http://",  adapter)
    return session

session = make_session()
try:
    r = session.get("https://httpbin.org/status/503", timeout=10)
    r.raise_for_status()
except requests.HTTPError as e:
    print("HTTP error:", e)
except requests.ConnectionError:
    print("Network failure")
except requests.Timeout:
    print("Request timed out")
else:
    print("Success:", r.text[:80])

6.5 Consuming REST APIs

import requests

class WeatherClient:
    BASE = "https://api.open-meteo.com/v1/forecast"

    def __init__(self, timeout=10):
        self.timeout = timeout

    def current(self, lat, lon):
        params = {
            "latitude": lat,
            "longitude": lon,
            "current_weather": True,
        }
        r = requests.get(self.BASE, params=params, timeout=self.timeout)
        r.raise_for_status()
        return r.json()["current_weather"]

client = WeatherClient()
weather = client.current(28.61, 77.21)         # Delhi
print(f"Temperature: {weather['temperature']}°C")
print(f"Wind speed : {weather['windspeed']} km/h")

6.6 Rate Limiting & Pagination

import requests
import time

def fetch_all_pages(url, per_page=100):
    page = 1
    while True:
        r = requests.get(url, params={"page": page, "per_page": per_page},
                         timeout=10)
        r.raise_for_status()
        data = r.json()
        if not data:
            break
        yield from data
        page += 1
        time.sleep(0.5)         # be polite to the server

# for item in fetch_all_pages("https://api.example.com/items"):
#     print(item)

VII. Web Scraping with BeautifulSoup

7.1 What is Web Scraping?

7.2 Install and Basic Usage

pip install requests beautifulsoup4 lxml

import requests
from bs4 import BeautifulSoup

url = "https://quotes.toscrape.com/"
response = requests.get(url, timeout=10)
response.raise_for_status()

soup = BeautifulSoup(response.text, "html.parser")

# Get the page title
print(soup.title.string)

# Find all quote texts
for quote in soup.find_all("span", class_="text"):
    print(quote.get_text(strip=True))

7.3 Navigating the Parse Tree

7.4 CSS Selectors

import requests
from bs4 import BeautifulSoup
import time

BASE = "https://quotes.toscrape.com"

def scrape_quotes(max_pages=3):
    page = 1
    quotes = []

    while page <= max_pages:
        url = f"{BASE}/page/{page}/"
        r = requests.get(url, timeout=10)
        if r.status_code != 200:
            break

        soup = BeautifulSoup(r.text, "html.parser")
        for block in soup.find_all("div", class_="quote"):
            text   = block.find("span", class_="text").get_text(strip=True)
            author = block.find("small", class_="author").get_text(strip=True)
            tags   = [t.get_text(strip=True) for t in
                      block.find_all("a", class_="tag")]
            quotes.append({"text": text, "author": author, "tags": tags})

        page += 1
        time.sleep(1)               # polite delay

    return quotes

data = scrape_quotes(3)
print(f"Scraped {len(data)} quotes.")
for q in data[:3]:
    print(f"- {q['author']}: {q['text'][:60]}...")

import requests
from bs4 import BeautifulSoup
from urllib.parse import urljoin

url = "https://www.python.org/"
r = requests.get(url, timeout=10)
soup = BeautifulSoup(r.text, "html.parser")

for a in soup.find_all("a", href=True):
    absolute = urljoin(url, a["href"])
    print(absolute)

7.5 Saving Scraped Data

import csv, json

# As CSV
with open("quotes.csv", "w", newline="", encoding="utf-8") as f:
    writer = csv.DictWriter(f, fieldnames=["author", "text", "tags"])
    writer.writeheader()
    for q in data:
        q["tags"] = "|".join(q["tags"])     # flatten list
        writer.writerow(q)

# As JSON
with open("quotes.json", "w", encoding="utf-8") as f:
    json.dump(data, f, ensure_ascii=False, indent=2)

7.6 Handling Dynamic Pages

Content rendered by JavaScript is not in the initial HTML. Options:

• Look for an underlying JSON API (best).
• Use selenium or playwright to drive a real browser.
• Use requests-html for simple JS rendering.

from selenium import webdriver
from selenium.webdriver.common.by import By

driver = webdriver.Chrome()
driver.get("https://example.com/dynamic")

elements = driver.find_elements(By.CSS_SELECTOR, ".item")
for el in elements:
    print(el.text)

driver.quit()

VIII. Data Analysis with Pandas

8.1 Why Pandas?

Pandas provides two core data structures — Series (1-D labelled array) and DataFrame (2-D table) — with fast, expressive tools for loading, cleaning, transforming, aggregating and analysing tabular data.

pip install pandas

8.2 Series and DataFrame

import pandas as pd
import numpy as np

# Series — 1-D labelled array
s = pd.Series([10, 20, 30], index=["a", "b", "c"])
print(s["b"])                  # 20
print(s.mean())                # 20.0

# DataFrame — 2-D table
data = {
    "name":   ["Aarav", "Diya", "Kabir", "Meera", "Zara"],
    "branch": ["CSE", "ECE", "CSE", "MEC", "ECE"],
    "marks":  [88, 95, 72, 81, 90],
    "cgpa":   [8.5, 9.3, 7.1, 8.0, 8.8],
}
df = pd.DataFrame(data)
print(df)
print(df.shape)                # (5, 4)
print(df.dtypes)

8.3 Reading and Writing Data

# CSV
df = pd.read_csv("students.csv")
df.to_csv("out.csv", index=False)

# Excel
df = pd.read_excel("students.xlsx", sheet_name="Sheet1")
df.to_excel("out.xlsx", index=False)

# JSON
df = pd.read_json("students.json")

# SQL (SQLite example)
import sqlite3
with sqlite3.connect("school.db") as conn:
    df = pd.read_sql_query("SELECT * FROM students", conn)

8.4 Inspection

8.5 Selection and Filtering

# Column access
df["marks"]                    # Series
df[["name", "marks"]]          # DataFrame

# Row selection by position
df.iloc[0]                     # first row
df.iloc[0:3, 0:2]              # first 3 rows, first 2 cols

# Row selection by label
df.loc[0, "name"]              # row 0, column 'name'

# Boolean filtering
toppers = df[df["marks"] >= 85]
cse     = df[(df["branch"] == "CSE") & (df["marks"] > 80)]
failed  = df[df["marks"] < 40]

# Query syntax
df.query("marks > 80 and branch == 'ECE'")

# Sorting
df.sort_values("marks", ascending=False)
df.sort_values(["branch", "marks"], ascending=[True, False])

8.6 Data Cleaning

# Missing values
df.isnull().sum()                      # count missing per column
df.dropna()                            # drop rows with any NaN
df.dropna(subset=["marks"])            # drop rows where marks is NaN
df.fillna(0)                           # replace NaN with 0
df["marks"].fillna(df["marks"].mean(), inplace=True)

# Duplicates
df.duplicated().sum()
df.drop_duplicates(inplace=True)

# Type conversion
df["marks"] = df["marks"].astype(int)
df["date"]  = pd.to_datetime(df["date"])

# Rename columns
df.rename(columns={"name": "student_name"}, inplace=True)

# Apply a function
df["grade"] = df["marks"].apply(lambda m: "A" if m >= 90 else "B" if m >= 80 else "C")

8.7 Grouping and Aggregation

import pandas as pd

df = pd.DataFrame({
    "name":   ["Aarav", "Diya", "Kabir", "Meera", "Zara", "Rohan"],
    "branch": ["CSE", "ECE", "CSE", "MEC", "ECE", "CSE"],
    "marks":  [88, 95, 72, 81, 90, 65],
    "cgpa":   [8.5, 9.3, 7.1, 8.0, 8.8, 6.9],
})

summary = df.groupby("branch").agg(
    count   = ("name",  "count"),
    avg_marks = ("marks", "mean"),
    max_cgpa  = ("cgpa",  "max"),
).reset_index()

print(summary)

  branch  count  avg_marks  max_cgpa
0    CSE      3  75.000000       8.5
1    ECE      2  92.500000       9.3
2    MEC      1  81.000000       8.0

8.8 Merging, Joining, Concatenating

df1 = pd.DataFrame({"id": [1, 2, 3], "name": ["A", "B", "C"]})
df2 = pd.DataFrame({"id": [2, 3, 4], "marks": [88, 72, 91]})

# Inner join (default)
pd.merge(df1, df2, on="id", how="inner")

# Left, right, outer
pd.merge(df1, df2, on="id", how="left")
pd.merge(df1, df2, on="id", how="right")
pd.merge(df1, df2, on="id", how="outer")

# Concatenate vertically
pd.concat([df1, df2], ignore_index=True)

# Concatenate horizontally
pd.concat([df1, df2], axis=1)

8.9 Pivot Tables

df = pd.DataFrame({
    "date":   ["2024-01", "2024-01", "2024-02", "2024-02"],
    "branch": ["CSE", "ECE", "CSE", "ECE"],
    "sales":  [100, 80, 120, 90],
})

pivot = df.pivot_table(
    index="date", columns="branch", values="sales", aggfunc="sum"
)
print(pivot)

8.10 Time Series Basics

import pandas as pd

dates = pd.date_range("2024-01-01", periods=10, freq="D")
ts = pd.Series([100 + i * 5 for i in range(10)], index=dates)

print(ts["2024-01-05"])           # label-based access
print(ts.resample("W").sum())     # weekly aggregation
print(ts.rolling(window=3).mean())  # 3-day moving average

IX. Introduction to Machine Learning with scikit-learn

9.1 What is Machine Learning?

9.2 The ML Workflow

9.3 scikit-learn — the Standard Library

pip install scikit-learn

Every estimator in scikit-learn follows a consistent API:

• model.fit(X_train, y_train) — train
• model.predict(X_test) — predict
• model.score(X_test, y_test) — evaluate
• model.get_params() / set_params() — inspect/configure

from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score, classification_report

# 1. Load data
iris = load_iris()
X, y = iris.data, iris.target

# 2. Split
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42, stratify=y)

# 3. Scale (optional for tree-based models, essential for others)
scaler = StandardScaler()
X_train = scaler.fit_transform(X_train)
X_test  = scaler.transform(X_test)

# 4. Train
clf = RandomForestClassifier(n_estimators=100, random_state=42)
clf.fit(X_train, y_train)

# 5. Predict
y_pred = clf.predict(X_test)

# 6. Evaluate
print("Accuracy:", accuracy_score(y_test, y_pred))
print(classification_report(y_test, y_pred,
                            target_names=iris.target_names))

Accuracy: 0.9

              precision    recall  f1-score   support
      setosa       1.00      1.00      1.00        10
  versicolor       0.82      0.90      0.86        10
   virginica       0.89      0.80      0.84        10

9.4 Regression Example

from sklearn.datasets import load_diabetes
from sklearn.linear_model import LinearRegression
from sklearn.model_selection import train_test_split
from sklearn.metrics import mean_squared_error, r2_score

X, y = load_diabetes(return_X_y=True)
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42)

model = LinearRegression()
model.fit(X_train, y_train)

y_pred = model.predict(X_test)
print(f"RMSE: {mean_squared_error(y_test, y_pred, squared=False):.2f}")
print(f"R²  : {r2_score(y_test, y_pred):.3f}")

9.5 Unsupervised — K-Means Clustering

from sklearn.datasets import make_blobs
from sklearn.cluster import KMeans
import matplotlib.pyplot as plt

X, _ = make_blobs(n_samples=300, centers=4, random_state=42)

kmeans = KMeans(n_clusters=4, random_state=42, n_init=10)
labels = kmeans.fit_predict(X)

plt.scatter(X[:, 0], X[:, 1], c=labels, cmap="viridis", s=20)
plt.scatter(kmeans.cluster_centers_[:, 0],
            kmeans.cluster_centers_[:, 1],
            marker="X", s=200, c="red", label="Centroids")
plt.title("K-Means Clustering")
plt.legend()
plt.tight_layout()
plt.show()

9.6 Model Evaluation Metrics

9.7 Avoiding Overfitting

from sklearn.datasets import load_iris
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import cross_val_score

X, y = load_iris(return_X_y=True)
clf = RandomForestClassifier(n_estimators=100, random_state=42)

scores = cross_val_score(clf, X, y, cv=5, scoring="accuracy")
print(f"Fold scores : {scores.round(3)}")
print(f"Mean ± std  : {scores.mean():.3f} ± {scores.std():.3f}")

X. Logging, Configuration & Environment

10.1 Why logging over print?

10.2 Basic Logging

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)

logging.debug("Detailed diagnostic")
logging.info("Program started")
logging.warning("Low disk space")
logging.error("Failed to connect")
logging.critical("System shutdown imminent")

10.3 Logging to a File with Rotation

import logging
from logging.handlers import RotatingFileHandler

logger = logging.getLogger("myapp")
logger.setLevel(logging.DEBUG)

handler = RotatingFileHandler(
    "app.log", maxBytes=1_000_000, backupCount=5)

formatter = logging.Formatter(
    "%(asctime)s | %(levelname)s | %(message)s")
handler.setFormatter(formatter)

logger.addHandler(handler)

for i in range(100):
    logger.info("Iteration %d", i)

10.4 Structured Configuration with configparser

A common pattern: read settings from an INI file.

# config.ini
[app]
name = MyApp
debug = true

[database]
host = localhost
port = 5432
user = admin

import configparser

config = configparser.ConfigParser()
config.read("config.ini")

print(config["app"]["name"])                       # MyApp
print(config.getboolean("app", "debug"))           # True
print(config.getint("database", "port"))           # 5432

10.5 Environment Variables

import os

# Reading
db_url = os.environ.get("DATABASE_URL", "sqlite:///local.db")
api_key = os.getenv("API_KEY", "")
debug   = os.getenv("DEBUG", "false").lower() == "true"

# Setting (within the process)
os.environ["APP_MODE"] = "production"

10.6 Reading .env Files with python-dotenv

# .env
DATABASE_URL=postgres://localhost/mydb
API_KEY=secret123
DEBUG=true

pip install python-dotenv

from dotenv import load_dotenv
import os

load_dotenv()               # reads .env into os.environ

print(os.getenv("DATABASE_URL"))
print(os.getenv("DEBUG") == "true")

10.7 Logging Best Practices

import logging.config

LOGGING = {
    "version": 1,
    "disable_existing_loggers": False,
    "formatters": {
        "standard": {
            "format": "%(asctime)s [%(levelname)s] %(name)s: %(message)s"
        },
    },
    "handlers": {
        "console": {
            "class": "logging.StreamHandler",
            "formatter": "standard",
            "level": "INFO",
        },
        "file": {
            "class": "logging.handlers.RotatingFileHandler",
            "formatter": "standard",
            "level": "DEBUG",
            "filename": "app.log",
            "maxBytes": 1_000_000,
            "backupCount": 3,
        },
    },
    "root": {
        "handlers": ["console", "file"],
        "level": "DEBUG",
    },
}

logging.config.dictConfig(LOGGING)
logger = logging.getLogger("myapp")

logger.info("Application started")
logger.error("Something failed")

XI. Type Hints, Static Analysis & Code Quality

11.1 Why Type Hints?

Python is dynamically typed, but type hints (PEP 484) let you document expected types. They are not enforced at runtime — they serve as documentation and enable static analysis tools to catch bugs before execution.

11.2 Basic Type Hints

def greet(name: str) -> str:
    return f"Hello, {name}"

def add(a: int, b: int) -> int:
    return a + b

def average(numbers: list[float]) -> float:
    return sum(numbers) / len(numbers)

# Variables
count: int = 0
prices: dict[str, float] = {"apple": 50.0}
point: tuple[int, int] = (3, 4)

11.3 Optional, Union and None

from typing import Optional, Union

def find_user(user_id: int) -> Optional[str]:
    """Returns a name or None if not found."""
    if user_id < 0:
        return None
    return "Aarav"

def parse(value: Union[int, str]) -> str:
    return str(value)

# Python 3.10+ shorthand
def find_user2(user_id: int) -> str | None:
    return "Diya" if user_id > 0 else None

11.4 Collections and Callables

from typing import Callable, Iterable, Sequence

def apply_all(xs: Iterable[int], f: Callable[[int], int]) -> list[int]:
    return [f(x) for x in xs]

def first(items: Sequence[str]) -> str:
    return items[0]

# Literal types (3.8+)
from typing import Literal
def set_mode(mode: Literal["r", "w", "a"]) -> None:
    print(f"Mode: {mode}")

# TypedDict (3.8+)
from typing import TypedDict
class Student(TypedDict):
    name: str
    roll: int
    marks: list[int]

11.5 Generics with TypeVar

from typing import TypeVar

T = TypeVar("T")

def first_or_default(items: list[T], default: T) -> T:
    return items[0] if items else default

print(first_or_default([1, 2, 3], 0))         # 1
print(first_or_default([], "none"))           # none

11.6 Dataclasses with Types

from dataclasses import dataclass, field

@dataclass
class Student:
    name: str
    roll: int
    marks: list[int] = field(default_factory=list)

    def average(self) -> float:
        return sum(self.marks) / len(self.marks) if self.marks else 0.0

s = Student("Aarav", 101, [88, 92, 79])
print(s.average())     # 86.333...

11.7 Static Analysis Tools

pip install mypy black ruff

# Format and check
black .
ruff check .
mypy src/

11.8 PEP 8 — Style Guide Essentials

from dataclasses import dataclass, field
from typing import Optional

@dataclass
class Account:
    owner: str
    balance: float = 0.0
    history: list[tuple[str, float]] = field(default_factory=list)

    def deposit(self, amount: float) -> None:
        if amount <= 0:
            raise ValueError("Deposit must be positive")
        self.balance += amount
        self.history.append(("deposit", amount))

    def withdraw(self, amount: float) -> bool:
        if amount <= 0 or amount > self.balance:
            return False
        self.balance -= amount
        self.history.append(("withdraw", amount))
        return True

    def last_transaction(self) -> Optional[tuple[str, float]]:
        return self.history[-1] if self.history else None

acc = Account("Aarav", 5000)
acc.deposit(1500)
print(acc.withdraw(2000))       # True
print(acc.balance)              # 4500.0
print(acc.last_transaction())   # ('withdraw', 2000)

XII. Virtual Environments, Packaging & Distribution

12.1 Why Virtual Environments?

Different projects need different library versions. Installing everything globally leads to dependency conflicts. A virtual environment is an isolated Python installation per project.

12.2 venv — the Standard Tool

# Create a virtual environment
python -m venv .venv

# Activate
# Linux / macOS:
source .venv/bin/activate

# Windows (PowerShell):
.venv\Scripts\Activate.ps1

# Windows (cmd):
.venv\Scripts\activate.bat

# Deactivate
deactivate

12.3 Managing Dependencies

# Install a package
pip install requests

# Install a specific version
pip install "requests==2.31.0"

# Install from requirements.txt
pip install -r requirements.txt

# Freeze current environment
pip freeze > requirements.txt

# List installed packages
pip list

# Upgrade pip
python -m pip install --upgrade pip

Example requirements.txt

requests>=2.31.0,<3.0.0
beautifulsoup4==4.12.2
pandas>=2.0
numpy
python-dotenv

12.4 pyproject.toml — Modern Project Metadata

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

[project]
name = "my-python-app"
version = "0.1.0"
description = "A sample Python project"
readme = "README.md"
requires-python = ">=3.10"
authors = [{name = "Aarav Sharma", email = "aarav@example.com"}]
dependencies = [
    "requests>=2.31.0",
    "python-dotenv>=1.0",
]

[project.optional-dependencies]
dev = [
    "pytest>=7.0",
    "black",
    "ruff",
    "mypy",
]

[project.scripts]
myapp = "myapp.cli:main"

12.5 Project Layout — the src Layout

my-python-app/
├── pyproject.toml
├── README.md
├── .gitignore
├── .env.example
├── src/
│   └── myapp/
│       ├── __init__.py
│       ├── cli.py
│       ├── core.py
│       └── utils.py
└── tests/
    ├── test_core.py
    └── test_utils.py

12.6 .gitignore Essentials

__pycache__/
*.py[cod]
.venv/
venv/
.env
*.egg-info/
dist/
build/
.pytest_cache/
.mypy_cache/
.ruff_cache/
.coverage
htmlcov/
*.log
.DS_Store

12.7 Building and Distributing

# Install build tool
pip install build

# Build distribution (source + wheel)
python -m build

# Output:
# dist/my_python_app-0.1.0-py3-none-any.whl
# dist/my_python_app-0.1.0.tar.gz

# Upload to PyPI (test first)
pip install twine
twine upload --repository testpypi dist/*
twine upload dist/*

12.8 Dockerising a Python App

# Dockerfile
FROM python:3.12-slim

WORKDIR /app

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

COPY src/ ./src/

ENV PYTHONUNBUFFERED=1
CMD ["python", "-m", "src.myapp"]

docker build -t myapp:latest .
docker run --rm -e API_KEY=abc123 myapp:latest

12.9 Continuous Integration (CI) Snippet

# .github/workflows/ci.yml
name: CI
on: [push, 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 }}
      - run: pip install -e ".[dev]"
      - run: ruff check .
      - run: mypy src/
      - run: pytest

XIII. Capstone Projects

These projects integrate the entire course — from Units I to IV. Use them as templates for the final practical exam and for portfolio submissions.

13.1 Weather Dashboard (API + Data + Visualization)

import requests
import pandas as pd
import matplotlib.pyplot as plt

class WeatherDashboard:
    URL = "https://api.open-meteo.com/v1/forecast"

    def fetch(self, lat, lon, days=7):
        params = {
            "latitude": lat, "longitude": lon,
            "daily": "temperature_2m_max,temperature_2m_min",
            "forecast_days": days,
            "timezone": "auto",
        }
        r = requests.get(self.URL, params=params, timeout=10)
        r.raise_for_status()
        data = r.json()["daily"]
        return pd.DataFrame({
            "date": data["time"],
            "max":  data["temperature_2m_max"],
            "min":  data["temperature_2m_min"],
        })

    def plot(self, df, city="Delhi"):
        df["date"] = pd.to_datetime(df["date"])
        plt.figure(figsize=(9, 4))
        plt.plot(df["date"], df["max"], marker="o", label="Max °C", color="crimson")
        plt.plot(df["date"], df["min"], marker="o", label="Min °C", color="steelblue")
        plt.fill_between(df["date"], df["min"], df["max"], alpha=0.15, color="teal")
        plt.title(f"7-Day Weather Forecast — {city}")
        plt.ylabel("Temperature (°C)")
        plt.legend()
        plt.grid(alpha=0.3)
        plt.tight_layout()
        plt.savefig("weather_dashboard.png", dpi=120)
        plt.show()

dash = WeatherDashboard()
df = dash.fetch(28.61, 77.21)
print(df)
dash.plot(df)

13.2 Concurrent Web Scraper

import requests
import csv
from bs4 import BeautifulSoup
from concurrent.futures import ThreadPoolExecutor, as_completed

BASE = "https://quotes.toscrape.com/page/{}/"

def scrape_page(page):
    r = requests.get(BASE.format(page), timeout=10)
    if r.status_code != 200:
        return []
    soup = BeautifulSoup(r.text, "html.parser")
    return [
        {
            "page":   page,
            "author": b.find("small", class_="author").text,
            "text":   b.find("span", class_="text").text,
        }
        for b in soup.find_all("div", class_="quote")
    ]

def scrape_all(max_pages=5):
    rows = []
    with ThreadPoolExecutor(max_workers=5) as ex:
        futures = {ex.submit(scrape_page, p): p for p in range(1, max_pages + 1)}
        for f in as_completed(futures):
            rows.extend(f.result())
    return rows

rows = scrape_all(5)
with open("quotes.csv", "w", newline="", encoding="utf-8") as f:
    writer = csv.DictWriter(f, fieldnames=["page", "author", "text"])
    writer.writeheader()
    writer.writerows(rows)

print(f"Saved {len(rows)} quotes.")

13.3 Async Chat Server

import asyncio

clients = set()

async def broadcast(message, sender):
    for c in list(clients):
        if c is not sender:
            c.write(message.encode())
            await c.drain()

async def handle(reader, writer):
    clients.add(writer)
    addr = writer.get_extra_info("peername")
    print(f"[+] {addr} joined ({len(clients)} online)")
    try:
        while True:
            data = await reader.read(1024)
            if not data:
                break
            msg = data.decode().strip()
            print(f"{addr}: {msg}")
            await broadcast(f"{addr}: {msg}\n", writer)
    finally:
        clients.discard(writer)
        writer.close()
        await writer.wait_closed()
        print(f"[-] {addr} left")

async def main():
    server = await asyncio.start_server(handle, "127.0.0.1", 8888)
    print("Chat server on 127.0.0.1:8888")
    async with server:
        await server.serve_forever()

# asyncio.run(main())

13.4 Sales Analytics Pipeline

import pandas as pd
import matplotlib.pyplot as plt

class SalesAnalytics:
    def __init__(self, path):
        self.df = pd.read_csv(path, parse_dates=["date"])

    def clean(self):
        self.df.dropna(subset=["amount"], inplace=True)
        self.df = self.df[self.df["amount"] > 0]
        self.df["month"] = self.df["date"].dt.to_period("M")
        return self

    def summary(self):
        return {
            "total_revenue": self.df["amount"].sum(),
            "orders":        len(self.df),
            "avg_order":     self.df["amount"].mean(),
            "top_region":    self.df.groupby("region")["amount"].sum().idxmax(),
        }

    def plot(self):
        monthly = self.df.groupby("month")["amount"].sum()
        by_region = self.df.groupby("region")["amount"].sum()

        fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(12, 4))
        monthly.plot(kind="line", marker="o", ax=ax1, color="teal")
        ax1.set_title("Monthly Revenue")
        ax1.set_ylabel("Revenue (₹)")

        by_region.plot(kind="bar", ax=ax2, color="coral")
        ax2.set_title("Revenue by Region")
        ax2.set_ylabel("Revenue (₹)")

        plt.tight_layout()
        plt.savefig("sales_dashboard.png", dpi=120)
        plt.show()

# sales = SalesAnalytics("sales.csv").clean()
# print(sales.summary())
# sales.plot()

13.5 Sentiment Analysis with scikit-learn

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.naive_bayes import MultinomialNB
from sklearn.pipeline import Pipeline
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report

texts = [
    "I love this product", "Absolutely fantastic service",
    "Best purchase ever", "Highly recommend",
    "Terrible experience", "Worst customer support",
    "Waste of money", "Never buying again",
    "Amazing quality", "Very disappointing",
]
labels = [1, 1, 1, 1, 0, 0, 0, 0, 1, 0]

X_train, X_test, y_train, y_test = train_test_split(
    texts, labels, test_size=0.3, random_state=42)

model = Pipeline([
    ("tfidf", TfidfVectorizer(stop_words="english")),
    ("clf",   MultinomialNB()),
])

model.fit(X_train, y_train)
predictions = model.predict(X_test)
print(classification_report(y_test, predictions))

# Predict new sentences
print(model.predict(["I really enjoyed it", "Awful quality"]))

XIV. Summary & Quick Reference Sheet

14.1 Design Patterns Cheat Sheet

14.2 Concurrency Cheat Sheet

14.3 Networking & APIs Cheat Sheet

14.4 Data Analysis Cheat Sheet

14.5 ML Cheat Sheet

14.6 Code Quality Cheat Sheet

14.7 Packaging Cheat Sheet

XV. Exam Tips & Practice Questions

Top 10 Exam Tips

1. Always state the GIL when contrasting threads and processes. Threads for I/O, processes for CPU.
2. Guard multiprocessing with if __name__ == "__main__": on Windows and macOS.
3. Use asyncio.gather for concurrency in coroutines; never call blocking functions inside an event loop.
4. Always pass timeout= to requests calls — omitting it is a top production bug.
5. Use parameterised SQL queries (?) to prevent SQL injection — never f-strings in SQL.
6. Respect scraping ethics: check robots.txt, add delays, prefer APIs.
7. For pandas questions, show the full pipeline: read → inspect → clean → transform → aggregate → save.
8. For ML questions, use cross-validation and report both train and test metrics — that demonstrates awareness of overfitting.
9. Type hints are documentation, not enforcement. Mention mypy and ruff for static checks.
10. Structure project answers cleanly: imports, class/functions, driver with sample I/O. Add comments and docstrings.

Practice Questions

XVI. Full Solutions to Practice Questions

Solution 1 — Singleton pattern

class Logger:
    _instance = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance.messages = []
        return cls._instance

    def log(self, msg):
        self.messages.append(msg)
        print(f"[LOG] {msg}")

a = Logger()
b = Logger()
a.log("hello")
print(a is b)                # True
print(b.messages)            # ['hello']

Real-world uses: (1) application-wide logger, (2) database connection pool, (3) configuration manager, (4) thread pool.

Solution 2 — Concurrent downloads

import time
from concurrent.futures import ThreadPoolExecutor, as_completed

def fetch(url, delay):
    time.sleep(delay)
    return f"{url} done"

urls = [("a.com", 1.0), ("b.com", 1.5), ("c.com", 1.2), ("d.com", 0.8), ("e.com", 1.3)]

# Sequential
start = time.perf_counter()
for u, d in urls:
    fetch(u, d)
print(f"Sequential: {time.perf_counter() - start:.2f}s")

# Concurrent with 5 workers
start = time.perf_counter()
with ThreadPoolExecutor(max_workers=5) as ex:
    futures = {ex.submit(fetch, u, d): u for u, d in urls}
    for f in as_completed(futures):
        print(f.result())
print(f"Concurrent: {time.perf_counter() - start:.2f}s")

Result: sequential ≈ 5.8s, concurrent ≈ 1.5s (bounded by the slowest task).

Solution 3 — Race condition and Lock

import threading

counter = 0
lock = threading.Lock()

def unsafe():
    global counter
    for _ in range(200_000):
        counter += 1

def safe():
    global counter
    for _ in range(200_000):
        with lock:
            counter += 1

# Unsafe
counter = 0
threads = [threading.Thread(target=unsafe) for _ in range(4)]
for t in threads: t.start()
for t in threads: t.join()
print("Unsafe:", counter)      # usually < 800000

# Safe
counter = 0
threads = [threading.Thread(target=safe) for _ in range(4)]
for t in threads: t.start()
for t in threads: t.join()
print("Safe  :", counter)      # 800000

Why it works: counter += 1 is three bytecode operations (read, add, write). Without a lock, two threads can interleave and lose updates. The lock makes the read-modify-write sequence atomic.

Solution 4 — Parallel squares with multiprocessing.Pool

from multiprocessing import Pool
import time

def square(n):
    return n * n

if __name__ == "__main__":
    nums = list(range(1, 21))

    start = time.perf_counter()
    seq = [square(n) for n in nums]
    print(f"Sequential: {time.perf_counter() - start:.4f}s")

    start = time.perf_counter()
    with Pool(processes=4) as pool:
        par = pool.map(square, nums)
    print(f"Parallel  : {time.perf_counter() - start:.4f}s")
    print("Equal:", seq == par)

Solution 5 — Concurrent fetches with asyncio

import asyncio
import time

async def fetch(name, delay):
    await asyncio.sleep(delay)
    return f"{name} finished in {delay}s"

async def main():
    tasks = [fetch("a", 1.0), fetch("b", 2.0), fetch("c", 1.5)]

    # Sequential
    start = time.perf_counter()
    for t in tasks:
        pass                    # already created — recreate for sequential
    seq_tasks = [fetch("a", 1.0), fetch("b", 2.0), fetch("c", 1.5)]
    start = time.perf_counter()
    for t in seq_tasks:
        await t
    print(f"Sequential: {time.perf_counter() - start:.2f}s")

    # Concurrent
    start = time.perf_counter()
    results = await asyncio.gather(fetch("a", 1.0), fetch("b", 2.0), fetch("c", 1.5))
    print(f"Concurrent: {time.perf_counter() - start:.2f}s")
    for r in results:
        print(" ", r)

asyncio.run(main())

Result: sequential ≈ 4.5s, concurrent ≈ 2.0s (bounded by the slowest).

Solution 6 — TCP echo server & client

# server.py
import socket

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
    s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    s.bind(("127.0.0.1", 65432))      # associate with address
    s.listen(5)                       # enter listening mode
    print("Server ready")
    conn, addr = s.accept()           # block until client connects
    with conn:
        while True:
            data = conn.recv(1024)
            if not data:
                break
            conn.sendall(b"Echo: " + data)

# client.py
import socket

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as c:
    c.connect(("127.0.0.1", 65432))   # initiate connection
    c.sendall(b"hello")
    print(c.recv(1024).decode())      # Echo: hello

Roles: bind associates the socket with a local address; listen queues incoming connections; accept blocks until a client connects and returns a new socket; connect is the client-side counterpart that initiates the handshake.

Solution 7 — GitHub API client

import requests

url = "https://api.github.com/search/repositories"
params = {"q": "python", "sort": "stars", "per_page": 5}

try:
    r = requests.get(url, params=params, timeout=10)
    r.raise_for_status()
except requests.HTTPError as e:
    print("HTTP error:", e)
else:
    for repo in r.json()["items"]:
        print(f"{repo['full_name']:<45} ⭐ {repo['stargazers_count']}")

Solution 8 — News headline scraper

import requests
from bs4 import BeautifulSoup
import csv

url = "https://www.thehindu.com/news/national/"
r = requests.get(url, timeout=10)
r.raise_for_status()

soup = BeautifulSoup(r.text, "html.parser")
headlines = [h.get_text(strip=True) for h in soup.find_all("h3")]

with open("headlines.csv", "w", newline="", encoding="utf-8") as f:
    writer = csv.writer(f)
    writer.writerow(["headline"])
    for h in headlines:
        writer.writerow([h])

print(f"Saved {len(headlines)} headlines.")

Solution 9 — Employee analytics with pandas

import pandas as pd

df = pd.read_csv("employees.csv")

# Average salary per department
avg = df.groupby("department")["salary"].mean().reset_index()
avg.columns = ["department", "avg_salary"]

# Highest-paid employee per department
idx = df.groupby("department")["salary"].idxmax()
top = df.loc[idx, ["department", "name", "salary"]]
top.columns = ["department", "top_earner", "top_salary"]

# Merge and export
summary = avg.merge(top, on="department")
summary.to_excel("employee_report.xlsx", index=False)
print(summary)

Solution 10 — Iris classification with cross-validation

from sklearn.datasets import load_iris
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split, cross_val_score
from sklearn.metrics import classification_report

X, y = load_iris(return_X_y=True)
X_tr, X_te, y_tr, y_te = train_test_split(X, y, test_size=0.2,
                                          random_state=42, stratify=y)

clf = RandomForestClassifier(n_estimators=200, random_state=42)
clf.fit(X_tr, y_tr)

print("Test metrics:")
print(classification_report(y_te, clf.predict(X_te),
                            target_names=load_iris().target_names))

scores = cross_val_score(clf, X, y, cv=5, scoring="accuracy")
print(f"5-fold CV: mean={scores.mean():.3f}, std={scores.std():.3f}")
print("Folds:", scores.round(3))

Solution 11 — Logging with dictConfig

import logging.config

LOGGING = {
    "version": 1,
    "disable_existing_loggers": False,
    "formatters": {
        "standard": {
            "format": "%(asctime)s | %(levelname)-8s | %(name)s | %(message)s"
        },
    },
    "handlers": {
        "console": {
            "class": "logging.StreamHandler",
            "formatter": "standard",
            "level": "INFO",
        },
        "file": {
            "class": "logging.handlers.RotatingFileHandler",
            "formatter": "standard",
            "level": "DEBUG",
            "filename": "app.log",
            "maxBytes": 1_000_000,
            "backupCount": 3,
        },
    },
    "root": {"handlers": ["console", "file"], "level": "DEBUG"},
}

logging.config.dictConfig(LOGGING)
log = logging.getLogger("demo")

log.debug("debug → file only")
log.info("info  → file + console")
log.error("error → both")

Solution 12 — Type-hinted Stack with generics

from typing import Generic, TypeVar, Optional

T = TypeVar("T")

class Stack(Generic[T]):
    """A LIFO stack supporting any hashable type."""

    def __init__(self) -> None:
        self._items: list[T] = []

    def push(self, item: T) -> None:
        """Push an item onto the top of the stack."""
        self._items.append(item)

    def pop(self) -> Optional[T]:
        """Remove and return the top item, or None if empty."""
        return self._items.pop() if self._items else None

    def peek(self) -> Optional[T]:
        """Return the top item without removing it."""
        return self._items[-1] if self._items else None

    def is_empty(self) -> bool:
        return not self._items

    def __len__(self) -> int:
        return len(self._items)

s: Stack[int] = Stack()
s.push(10); s.push(20)
print(s.pop())       # 20
print(s.peek())      # 10
print(len(s))        # 1

Solution 13 — venv setup and project layout

# 1. Create environment
python -m venv .venv

# 2. Activate
source .venv/bin/activate          # Linux/macOS
# .venv\Scripts\activate           # Windows

# 3. Install dependencies
pip install requests pandas

# 4. Freeze
pip freeze > requirements.txt

# 5. Project layout (src-based)
my-project/
├── pyproject.toml
├── README.md
├── .gitignore
├── requirements.txt
├── src/
│   └── myproject/
│       ├── __init__.py
│       ├── cli.py
│       └── core.py
└── tests/
    ├── test_core.py
    └── test_cli.py

Solution 14 — Concurrency vs parallelism

• Threading: many simultaneous file, network, DB operations; GIL is released during I/O.
• Multiprocessing: heavy numeric computation, image processing, data pipelines.
• asyncio: thousands of concurrent sockets or HTTP connections — single-threaded event loop, cooperative scheduling.

XVII. References, Key Takeaways & CO Mapping

17.1 Textbooks and References

17.2 Relevant Web Resources

17.3 Key Takeaways

1. Design patterns are reusable templates for common software problems — Singleton (one instance), Factory (decouple creation), Observer (notify many), Strategy (interchangeable algorithms).
2. The GIL prevents Python threads from executing bytecode in parallel; therefore threads help with I/O-bound tasks but not CPU-bound tasks.
3. Multiprocessing achieves true parallelism by spawning separate processes; always guard with if __name__ == "__main__".
4. asyncio uses a single-threaded event loop with cooperative scheduling — ideal for thousands of concurrent I/O operations.
5. Sockets are the low-level foundation of networking: TCP for reliability, UDP for speed.
6. HTTP and REST APIs are consumed with requests — always pass timeout and handle errors with raise_for_status().
7. Web scraping with requests + BeautifulSoup should respect robots.txt, be rate-limited, and prefer official APIs.
8. pandas provides Series and DataFrame with fast, expressive tools for cleaning, aggregating, joining and analysing tabular data.
9. scikit-learn offers a consistent API across all estimators: fit, predict, score; use cross-validation to detect overfitting.
10. Logging is production-grade, unlike print. Configure with dictConfig for centralised, environment-specific output.
11. Type hints document intent and enable static analysis (mypy, pyright) — they are not enforced at runtime.
12. Virtual environments (venv) isolate dependencies; pyproject.toml is the modern project metadata file.
13. Professional Python combines clean code (PEP 8, black, ruff), tests (unittest, pytest), type checks (mypy), and CI/CD pipelines.

17.4 CO Mapping

17.5 Recommended Learning Path

17.6 Common Pitfalls — Recap