Python Journey β€” Beginner-Friendly Learning Guide
βŒƒK Search

🐍Welcome to Python Journey

This guide is written for complete beginners β€” even if you have never written a single line of code in your life, you can follow along. Every concept is explained in plain English first, with real-world analogies, before showing you the code. By the end, you will also be interview-ready for Python, LangChain, LangGraph, and FastAPI questions.

How to use this guide: Read the explanation section carefully before looking at the code. The code then shows you how the idea translates into actual Python syntax. Pay attention to the πŸ’‘ analogy boxes, ⚠️ common mistake boxes, and πŸ“Œ beginner notes β€” these are where the most important lessons hide.

18
Sections
100+
Code Examples
60+
Interview Q&As
4
Major Frameworks
🐍
Python Core
Basics + Advanced
β†’
πŸ”—
LangChain
LLM Pipelines / RAG
β†’
πŸ•ΈοΈ
LangGraph
Agent State Graphs
β†’
⚑
FastAPI
Async Web APIs

πŸ“¦Variables & Data Types

What is a Variable?

Imagine your computer's memory as a huge warehouse full of empty boxes. When you want to store a piece of information β€” like someone's name or a price β€” Python creates a box, puts the information inside, and sticks a label on the outside. That label is the variable name.

When you write name = "Alice", Python does three things: creates a box in memory, puts the text "Alice" inside it, and labels the box "name". Whenever you write name later in your code, Python goes to the warehouse, finds the box labelled "name", and gives you back what's inside.

Variable = Labelled box. You can change what's inside the box at any time (that's why it's called a variable β€” its value can vary). The label stays the same, but you can swap the contents whenever you want.

Python is dynamically typed, which means the box can hold anything β€” a number today, a word tomorrow. Compare this to languages like Java or C++, where each box can only hold one specific type, like "this box can only hold integers".

The Basic Data Types

Python has several built-in types for different kinds of data:

  • int β€” whole numbers (no decimal point): 42, -7, 1000000
  • float β€” numbers with a decimal point: 3.14, -0.5, 99.99
  • str β€” text (called a "string"): "Hello", 'Alice'
  • bool β€” only two possible values: True or False
  • None β€” represents "nothing" or "no value yet". Like an empty box with no contents.
python β€” variables and types
# The = sign means "put this value into this variable"
# It is NOT asking "are these equal?" β€” that uses ==

name    = "Alice"       # str  β€” text goes in quotes
age     = 30            # int  β€” whole number, no quotes
height  = 1.75          # float β€” has a decimal point
is_active = True        # bool β€” capital T, no quotes!
address = None          # None β€” this person has no address yet

# You can check what type a variable holds
print(type(name))       # <class 'str'>
print(type(age))        # <class 'int'>
print(type(height))     # <class 'float'>
print(type(is_active))  # <class 'bool'>

# isinstance() checks if a value is a certain type β€” returns True or False
print(isinstance(age, int))     # True
print(isinstance(name, str))    # True
print(isinstance(age, float))   # False β€” 30 is an int, not a float

# Multiple assignment on one line (Python allows this!)
x, y, z = 1, 2, 3    # x=1, y=2, z=3

# Assign the same value to multiple variables
a = b = c = 0         # a=0, b=0, c=0

# You can change a variable's value at any time
age = 30
age = 31              # now age is 31 β€” the old value (30) is gone
age = "thirty-one"    # Python lets you even change the TYPE (dynamic typing!)

Why does None exist? Imagine a form where you fill in your address β€” but some people haven't filled it in yet. You can't just leave the variable as "" (empty string) because that might mean something different. None explicitly means "this has no value yet". It's the programmer's way of saying "nothing here".

The Walrus Operator := (Python 3.8+)

The walrus operator := lets you assign a value to a variable inside an expression. This is useful when you want to check something AND save the result at the same time, without writing two separate lines.

python β€” walrus operator
import re  # Python's built-in pattern-matching library

# WITHOUT walrus β€” two steps
match = re.search(r"\d+", "user42")  # look for digits in "user42"
if match:
    print(match.group())             # prints "42"

# WITH walrus β€” one step! Assign AND check in the same line
# re.search() runs, its result goes into 'm', and 'm' is checked for truth
if m := re.search(r"\d+", "user42"):
    print(m.group())   # also prints "42"

# Another common use: reading lines from a file
# while line := file.readline():    β€” reads a line, stops when empty
#     process(line)

Don't confuse = and ==!

= is assignment β€” "put this value into this variable".
== is comparison β€” "are these two things equal?" (returns True or False).

Writing if age = 30: is a syntax error. You must write if age == 30: to check.

πŸ“Strings

What is a String?

A string is simply a sequence of characters β€” letters, digits, spaces, punctuation β€” all strung together. In Python, you create a string by putting text inside quotes (single ' or double ", both work the same).

Think of a string like a sequence of beads on a necklace. Each bead is a character. The first bead is at position 0 (Python counts from zero!), the second at position 1, and so on.

Strings are immutable β€” like printed pages. Once a page is printed, you can't erase one letter and replace it. To "change" the text, you have to print a brand new page. In Python, every time you "modify" a string, Python actually creates a brand new string and throws away the old one. The variable then points to the new string.

python β€” string basics and indexing
# Creating strings β€” single or double quotes, both work
greeting = "Hello, World!"
name     = 'Alice'

# Strings have positions (indices) starting at 0
# "Python"
#  P y t h o n
#  0 1 2 3 4 5   β€” counting from the left (positive indices)
# -6-5-4-3-2-1   β€” counting from the right (negative indices)

word = "Python"
print(word[0])    # "P" β€” first character
print(word[1])    # "y" β€” second character
print(word[-1])   # "n" β€” last character (count from right)
print(word[-2])   # "o" β€” second-to-last character

# SLICING β€” getting a piece of the string
# Format: string[start : stop : step]
# 'stop' is NOT included (it stops just before that index)
print(word[0:3])   # "Pyt" β€” indices 0, 1, 2 (NOT 3)
print(word[2:5])   # "tho" β€” indices 2, 3, 4
print(word[:3])    # "Pyt" β€” from the start up to (not including) index 3
print(word[3:])    # "hon" β€” from index 3 to the end
print(word[::-1])  # "nohtyP" β€” reversed! step=-1 goes backwards

# Length of a string
print(len(word))   # 6 β€” there are 6 characters in "Python"

f-strings β€” The Modern Way to Build Text

Very often you need to build a message that combines fixed text with variable values β€” like "Hello Alice, you have 5 messages". The old way was clunky. The modern way is f-strings (introduced in Python 3.6). Just put an f before the opening quote, and then put your variable names inside curly braces {}.

python β€” f-strings and string methods
name  = "Alice"
score = 99.5
count = 5

# f-string β€” put an 'f' before the quote, then use {} for variables
message = f"Hello {name}, your score is {score}"
# Python replaces {name} with "Alice" and {score} with 99.5
# Result: "Hello Alice, your score is 99.5"

# You can control how numbers are formatted
# :.2f means "show 2 decimal places as a float"
formatted = f"Score: {score:.2f}"     # "Score: 99.50"
# :d means integer, :,d adds comma separators for thousands
big_num   = f"Total: {1234567:,}"     # "Total: 1,234,567"
# :% converts to a percentage
ratio     = f"Pass rate: {0.875:.1%}" # "Pass rate: 87.5%"

# ──────────────────────────────────────────────────────
# COMMON STRING METHODS
# A "method" is a function that belongs to a string β€” you call it with a dot
# ──────────────────────────────────────────────────────

s = "  Hello, World!  "

s.strip()              # "Hello, World!" β€” removes spaces from both ends
s.lstrip()             # "Hello, World!  " β€” removes only left spaces
s.rstrip()             # "  Hello, World!" β€” removes only right spaces
s.upper()              # "  HELLO, WORLD!  " β€” ALL CAPS
s.lower()              # "  hello, world!  " β€” all lowercase
s.title()              # "  Hello, World!  " β€” Title Case

# replace(old, new) β€” swap one piece of text for another
"Hello World".replace("World", "Python")   # "Hello Python"

# split(separator) β€” break a string into a list of parts
"a,b,c".split(",")           # ["a", "b", "c"]
"hello world".split()        # ["hello", "world"] β€” splits on any whitespace

# join(list) β€” glue a list of strings together with a separator
",".join(["a", "b", "c"])    # "a,b,c"
" ".join(["Hello", "World"]) # "Hello World"

# startswith / endswith β€” check the beginning or end
"hello".startswith("hel")    # True
"image.png".endswith(".png") # True

# find / index β€” where does this substring appear?
"hello".find("ll")           # 2 β€” found at index 2
"hello".find("xyz")          # -1 β€” not found (returns -1)

# in / not in β€” does this string contain another?
"py" in "python"             # True
"xyz" in "python"            # False

# count β€” how many times does a substring appear?
"banana".count("a")          # 3

Strings are immutable β€” methods don't change the original. When you call s.upper(), Python does NOT change s. It returns a NEW string. You must save the result: s = s.upper() or upper_s = s.upper(). If you just write s.upper() without saving it, the result is thrown away.

πŸ“šCollections β€” Lists, Dicts, Sets, Tuples

Often you don't want to store just one value β€” you want to store many. Python has four main types of collections, each with a different purpose. Think of them like different kinds of storage containers:

  • List β€” An ordered, numbered shopping list. Items have positions (0, 1, 2...). You can add, remove, and change items.
  • Tuple β€” A locked list. Once created, the contents cannot change. Good for fixed data like coordinates (x, y) or RGB colours.
  • Dict β€” A contact book. You look up by name (key) to get the phone number (value). Super fast lookups.
  • Set β€” A bag where duplicates automatically disappear. Good for membership testing and removing duplicates.

Lists β€” Ordered, Changeable Collections

A list is like a numbered shopping list. Each item has a position number starting at 0. You can add new items, remove existing ones, and change any item. Lists are created with square brackets [].

Imagine a train with numbered carriages: carriage 0, carriage 1, carriage 2... You can add a carriage to the end, remove a carriage, or look inside a specific carriage by its number. That's a list.

python β€” lists
# Creating a list β€” square brackets, items separated by commas
fruits = ["apple", "banana", "cherry"]
numbers = [3, 1, 4, 1, 5, 9, 2, 6]
mixed = [42, "hello", True, 3.14]  # lists can hold different types

# Accessing items by index (position) β€” starts at 0!
print(fruits[0])   # "apple"  β€” first item
print(fruits[1])   # "banana" β€” second item
print(fruits[-1])  # "cherry" β€” last item (negative counts from end)

# Changing an item
fruits[1] = "mango"   # replace "banana" with "mango"
# fruits is now ["apple", "mango", "cherry"]

# ─── ADDING ITEMS ──────────────────────────────────────────────
fruits.append("grape")        # add to the END: ["apple","mango","cherry","grape"]
fruits.insert(1, "orange")    # insert at position 1, shifting others right
fruits.extend(["kiwi","plum"])# add multiple items from another list

# ─── REMOVING ITEMS ────────────────────────────────────────────
fruits.remove("mango")        # removes the FIRST occurrence of "mango"
last = fruits.pop()           # removes AND returns the last item
second = fruits.pop(1)        # removes AND returns item at index 1

# ─── SORTING ───────────────────────────────────────────────────
nums = [3, 1, 4, 1, 5, 9]
nums.sort()                      # modifies the list IN-PLACE: [1,1,3,4,5,9]
sorted_desc = sorted(nums, reverse=True)  # returns a NEW list: [9,5,4,3,1,1]
# nums is still [1,1,3,4,5,9] β€” sorted() doesn't touch the original

# ─── USEFUL OPERATIONS ─────────────────────────────────────────
print(len(fruits))          # number of items
print(3 in numbers)         # True β€” 3 is in the list?
print(numbers.count(1))     # 2 β€” how many times does 1 appear?
numbers.reverse()           # reverse in-place

# ─── SLICING β€” getting a sub-list ──────────────────────────────
nums = [10, 20, 30, 40, 50]
print(nums[1:3])    # [20, 30] β€” indices 1 and 2 (NOT 3)
print(nums[:3])     # [10, 20, 30] β€” first 3 items
print(nums[2:])     # [30, 40, 50] β€” from index 2 to the end
print(nums[::2])    # [10, 30, 50] β€” every 2nd item

# ─── LIST COMPREHENSIONS β€” a powerful shortcut ─────────────────
# Normal way to make a list of squares:
squares = []
for x in range(10):
    squares.append(x**2)

# List comprehension β€” same result, one line:
squares = [x**2 for x in range(10)]
# Read as: "make a list of x squared, for each x from 0 to 9"

# With a condition β€” only include even numbers
evens = [x for x in range(20) if x % 2 == 0]
# [0, 2, 4, 6, 8, 10, 12, 14, 16, 18]

# Flatten a nested list
nested = [[1, 2], [3, 4], [5, 6]]
flat = [x for row in nested for x in row]  # [1, 2, 3, 4, 5, 6]

Tuples β€” Immutable (Locked) Sequences

A tuple looks like a list but uses parentheses () and, crucially, cannot be changed after creation. Once you put items in a tuple, they are locked in forever.

Imagine the coordinates of a city on a map β€” (latitude, longitude). These are fixed facts. You'd never want someone to accidentally change them. A tuple enforces this β€” it's a guarantee that the data won't be modified. Use tuples for data that should never change: coordinates, RGB colours, database column names, function return values.

python β€” tuples
# Creating a tuple β€” parentheses (or even just commas)
point = (3, 4)               # x=3, y=4
rgb_red = (255, 0, 0)        # red colour in RGB
single = (42,)               # single-item tuple MUST have a trailing comma!
also_tuple = 1, 2, 3         # parentheses are optional β€” commas create tuples

# Accessing items β€” same as lists, by index
print(point[0])   # 3 β€” x coordinate
print(point[1])   # 4 β€” y coordinate

# UNPACKING β€” pulling items out into separate variables
x, y = point     # x=3, y=4
r, g, b = rgb_red  # r=255, g=0, b=0

# Tuples cannot be changed (this would cause a TypeError)
# point[0] = 10  # TypeError: 'tuple' object does not support item assignment

# ─── WHY USE TUPLES INSTEAD OF LISTS? ──────────────────────────
# 1. They signal "this data should not change" β€” good for code clarity
# 2. They are slightly faster than lists
# 3. They can be used as dictionary keys (lists cannot!)
# 4. Functions often return multiple values as a tuple

def min_max(numbers):
    """Return the minimum and maximum of a list."""
    return min(numbers), max(numbers)    # Python returns a tuple

low, high = min_max([3, 1, 4, 1, 5])
print(f"Min: {low}, Max: {high}")  # Min: 1, Max: 5

# ─── NAMED TUPLES β€” tuples with labels ──────────────────────────
from collections import namedtuple

# Create a "template" called Color with fields r, g, b
Color = namedtuple("Color", ["r", "g", "b"])
red = Color(255, 0, 0)
print(red.r)      # 255 β€” access by name (much clearer than red[0])
print(red.g)      # 0
print(red)        # Color(r=255, g=0, b=0)

Dictionaries β€” Key-Value Lookup Tables

A dictionary stores data as key-value pairs. Think of it like a real dictionary β€” you look up a word (the key) to find its definition (the value). Or like a contact book β€” you look up a person's name (key) to find their phone number (value).

Dictionaries use curly braces {}, and each entry is written as key: value.

A Python dict is like a filing cabinet with labelled folders. Instead of searching through every folder one by one (slow), you go directly to the folder labelled "Alice" (instant). This is why dict lookups are O(1) β€” they take the same time no matter how many items are in the dict.

python β€” dictionaries
# Creating a dictionary
user = {
    "name": "Alice",     # "name" is the key, "Alice" is the value
    "age": 30,           # "age" is the key, 30 is the value
    "email": "alice@example.com"
}

# ─── READING VALUES ─────────────────────────────────────────────
print(user["name"])           # "Alice" β€” access by key
# user["phone"]               # KeyError! "phone" doesn't exist

# .get() is SAFER β€” returns None (or a default) if key doesn't exist
print(user.get("phone"))           # None β€” no error, just None
print(user.get("phone", "N/A"))    # "N/A" β€” custom default value

# ─── ADDING / UPDATING VALUES ──────────────────────────────────
user["city"] = "New York"     # add a new key-value pair
user["age"]  = 31             # update an existing value (age was 30, now 31)
user.update({"age": 32, "country": "USA"})  # update multiple at once

# ─── REMOVING VALUES ───────────────────────────────────────────
del user["city"]              # remove key "city" and its value
removed = user.pop("country", None)  # remove and return value; None if not found

# ─── ITERATING (LOOPING) ───────────────────────────────────────
for key in user:              # loop over all keys
    print(key)                # prints: name, age, email

for value in user.values():   # loop over all values
    print(value)

for key, value in user.items():  # loop over key-value pairs
    print(f"{key} = {value}")    # name = Alice, age = 32, etc.

# ─── CHECKING EXISTENCE ─────────────────────────────────────────
print("name" in user)      # True β€” does key "name" exist?
print("phone" in user)     # False β€” no "phone" key

# ─── DICT COMPREHENSION ─────────────────────────────────────────
# Create a dict of numbers and their squares
squares = {x: x**2 for x in range(6)}
# {0:0, 1:1, 2:4, 3:9, 4:16, 5:25}

# ─── defaultdict β€” no KeyError when accessing missing keys ──────
from collections import defaultdict
word_count = defaultdict(int)  # missing keys default to 0
for word in ["apple", "banana", "apple", "cherry", "apple"]:
    word_count[word] += 1      # no KeyError even on first access!
# {"apple": 3, "banana": 1, "cherry": 1}

# ─── Merging two dicts (Python 3.9+) ────────────────────────────
defaults = {"timeout": 30, "retries": 3}
custom   = {"host": "localhost", "retries": 5}
merged   = {**defaults, **custom}   # custom values override defaults
# OR in Python 3.9+:
merged   = defaults | custom        # same result, cleaner syntax

Sets β€” Unique Item Collections

A set is like a bag where duplicates are automatically removed. If you put the number 5 in twice, the set only keeps one copy. Sets are great for removing duplicates from a list and for membership testing (checking if something is in the set).

Imagine a guest list at a party. If the same person tries to sign in twice, the doorman says "you're already on the list!" That's a set β€” every entry is unique. Also, the order doesn't matter (sets are unordered), just like a guest list doesn't care about the order people arrived.

python β€” sets
# Creating a set β€” curly braces (but no key:value pairs β€” that would be a dict!)
fruits = {"apple", "banana", "cherry", "apple"}  # "apple" appears twice
print(fruits)   # {"apple", "banana", "cherry"} β€” only ONE apple!
# Note: the order may differ each time you run it β€” sets are UNORDERED

# Creating a set from a list (great for removing duplicates)
numbers = [1, 2, 2, 3, 3, 3, 4, 4, 4, 4]
unique = set(numbers)       # {1, 2, 3, 4}
unique_list = list(unique)  # convert back to list if needed

# ─── ADDING AND REMOVING ─────────────────────────────────────────
fruits.add("grape")          # add one item
fruits.remove("banana")      # remove item (raises error if not there)
fruits.discard("mango")      # remove if present (NO error if not there)

# ─── MEMBERSHIP TEST β€” the main reason to use sets ─────────────
# This is O(1) β€” super fast even with millions of items
print("apple" in fruits)     # True
print("mango" in fruits)     # False

# Compare: checking in a LIST is O(n) β€” it checks every item one by one
# Checking in a SET is O(1) β€” instant, uses a hash lookup

# ─── SET OPERATIONS (like in maths) ─────────────────────────────
a = {1, 2, 3, 4}
b = {3, 4, 5, 6}

print(a | b)   # UNION: {1,2,3,4,5,6} β€” everything in either set
print(a & b)   # INTERSECTION: {3,4} β€” only what's in BOTH sets
print(a - b)   # DIFFERENCE: {1,2} β€” in 'a' but NOT in 'b'
print(a ^ b)   # SYMMETRIC DIFFERENCE: {1,2,5,6} β€” in one but not both

# ─── frozenset β€” an immutable set ───────────────────────────────
fs = frozenset({1, 2, 3})  # cannot add or remove items
# Can be used as a dict key (regular sets cannot)

Unpacking β€” Pulling Values Out of Collections

Unpacking lets you pull out multiple values from a list, tuple, or any iterable and assign them to separate variables all in one line. It makes your code much more readable.

python β€” unpacking and starred expressions
# Basic unpacking β€” number of variables must match number of items
point = (10, 20)
x, y = point           # x=10, y=20

rgb = (255, 128, 0)
red, green, blue = rgb # red=255, green=128, blue=0

# ─── STARRED EXPRESSIONS β€” grab the "rest" ──────────────────────
# The * before a variable name means "put everything else in here"
numbers = [1, 2, 3, 4, 5]
first, second, *rest = numbers
# first=1, second=2, rest=[3, 4, 5]

first, *middle, last = [1, 2, 3, 4, 5]
# first=1, middle=[2, 3, 4], last=5

# ─── SWAP WITHOUT A TEMP VARIABLE ───────────────────────────────
# In many languages, swapping requires a temporary variable:
# temp = a; a = b; b = temp

# In Python, you can do it in one line:
a, b = 10, 20
a, b = b, a    # now a=20, b=10  β€” elegant!

# ─── UNPACKING IN LOOPS ─────────────────────────────────────────
# Very common pattern β€” unpacking tuples while looping
pairs = [(1, "one"), (2, "two"), (3, "three")]
for number, word in pairs:    # unpack each tuple automatically
    print(f"{number} = {word}")

# ─── DICT UNPACKING WITH ** ─────────────────────────────────────
defaults = {"timeout": 30, "retries": 3}
# The ** operator "spreads" a dict into key=value pairs
config = {"host": "localhost", **defaults, "retries": 5}
# The later "retries": 5 overrides defaults' "retries": 3
# Result: {"host": "localhost", "timeout": 30, "retries": 5}

# ─── LIST UNPACKING WITH * ──────────────────────────────────────
def add(a, b, c):
    return a + b + c

nums = [1, 2, 3]
result = add(*nums)   # same as add(1, 2, 3) β€” * unpacks the list into args

Quick Comparison β€” Which Collection to Use?

TypeSyntaxOrdered?Mutable?Duplicates?Best For
list[1,2,3]YesYesYesOrdered data you'll add/remove items from
tuple(1,2,3)YesNoYesFixed data (coordinates, RGB, function returns)
dict{"a":1}Insertion orderYesNo (keys)Looking up values by a meaningful key
set{1,2,3}NoYesNoMembership testing, removing duplicates

πŸ”€Control Flow

Control flow lets your program make decisions and repeat actions. Without it, code just runs from top to bottom with no logic. With it, you can write "if this condition is true, do X; otherwise do Y" and "repeat this action 10 times".

if / elif / else β€” Making Decisions

The if statement is how your program makes decisions. It checks a condition. If the condition is True, it runs a block of code. If not, it moves on. You can chain multiple conditions with elif ("else if") and provide a default with else.

Think of a bouncer at a club: "If you are over 18, come in. Else if you have a VIP pass, come in. Otherwise, sorry, no entry." The bouncer checks each condition in order and acts on the first one that matches.

python β€” if/elif/else
score = 85

# Python checks these conditions from TOP to BOTTOM
# It runs the FIRST block whose condition is True, then STOPS
if score >= 90:
    grade = "A"
elif score >= 80:    # "elif" = "else if" β€” only checked if the above was False
    grade = "B"
elif score >= 70:
    grade = "C"
elif score >= 60:
    grade = "D"
else:                # "else" runs if ALL the above conditions were False
    grade = "F"

print(f"Grade: {grade}")  # "Grade: B"

# ─── TERNARY EXPRESSION β€” a one-line if/else ────────────────────
# Format: value_if_true if condition else value_if_false
label = "pass" if score >= 60 else "fail"

# ─── TRUTHINESS β€” what counts as True or False ──────────────────
# These are all "falsy" (treated as False in an if statement):
# False, None, 0, 0.0, "" (empty string), [], {}, () (empty collections)

# These are all "truthy" (treated as True):
# True, any non-zero number, any non-empty string, any non-empty collection

if not []:          # empty list is falsy, so "not []" is True
    print("empty list is falsy")

if not "":          # empty string is falsy
    print("empty string is falsy")

if not 0:           # zero is falsy
    print("zero is falsy")

if "hello":         # non-empty string is truthy
    print("non-empty string is truthy")

# ─── CHAINED COMPARISONS β€” a Python superpower ──────────────────
# Instead of: if score >= 0 and score <= 100:
# You can write:
if 0 <= score <= 100:           # much more readable!
    print("score is valid")

# ─── AND, OR, NOT β€” combining conditions ───────────────────────
age = 25
has_ticket = True

if age >= 18 and has_ticket:    # both must be True
    print("You can enter")

if age < 10 or age > 65:        # at least one must be True
    print("Discount applies")

if not has_ticket:              # NOT flips True to False and vice versa
    print("You need a ticket")

Loops β€” Repeating Actions

Loops let you repeat a block of code without writing it multiple times. Python has two types of loops: for loops (repeat once for each item in a collection) and while loops (repeat as long as a condition is true).

python β€” for, while, enumerate, zip
# ─── FOR LOOP β€” repeat for each item in a collection ────────────
fruits = ["apple", "banana", "cherry"]
for fruit in fruits:       # fruit takes each value in turn
    print(fruit)           # prints apple, then banana, then cherry

# ─── RANGE β€” loop a specific number of times ───────────────────
for i in range(5):         # i goes 0, 1, 2, 3, 4
    print(i)               # prints 0 to 4

# range(start, stop, step) β€” start included, stop NOT included
for i in range(1, 11):     # 1 to 10 (not 11)
    print(i)

for i in range(0, 20, 2):  # every 2nd number: 0, 2, 4, 6, ... 18
    print(i)

# ─── ENUMERATE β€” loop with automatic index numbering ────────────
# enumerate() gives you BOTH the index AND the value at each step
for index, fruit in enumerate(fruits):
    print(f"{index}: {fruit}")
# 0: apple
# 1: banana
# 2: cherry

# Start counting from 1 instead of 0
for index, fruit in enumerate(fruits, start=1):
    print(f"{index}. {fruit}")
# 1. apple   2. banana   3. cherry

# ─── ZIP β€” loop over two lists in parallel ──────────────────────
names  = ["Alice", "Bob", "Charlie"]
scores = [90, 85, 78]
for name, score in zip(names, scores):   # pairs them up: (Alice,90), (Bob,85)...
    print(f"{name} scored {score}")

# ─── WHILE LOOP β€” repeat while a condition is True ──────────────
count = 0
while count < 5:           # keep looping as long as count < 5
    print(count)
    count += 1             # IMPORTANT: must change count or loop runs forever!

# ─── BREAK β€” exit the loop early ────────────────────────────────
for number in range(100):
    if number == 10:
        break              # stop looping immediately when number reaches 10
    print(number)          # prints 0 to 9

# ─── CONTINUE β€” skip the rest of this iteration ─────────────────
for number in range(10):
    if number % 2 == 0:    # if number is even...
        continue           # ...skip the rest of this loop body and go to next
    print(number)          # prints only odd numbers: 1, 3, 5, 7, 9

# ─── FOR-ELSE β€” a Python-specific feature ───────────────────────
# The 'else' block runs ONLY if the loop completed WITHOUT a break
for number in [2, 4, 6, 8]:
    if number % 3 == 0:
        print(f"Found divisible by 3: {number}")
        break
else:
    print("No number divisible by 3 was found")  # this runs here

match/case β€” Pattern Matching (Python 3.10+)

The match statement is Python's version of "switch/case" from other languages, but much more powerful. It can match not just values but also the structure of data β€” like matching a dictionary that has specific keys, or a list of a specific length.

python β€” match/case (Python 3.10+)
# Match on exact values β€” like a cleaner if/elif chain
status_code = 404

match status_code:
    case 200:
        print("OK β€” Success")
    case 404:
        print("Not Found")      # this runs
    case 500:
        print("Server Error")
    case _:                     # _ is the "wildcard" β€” matches anything
        print("Unknown status")

# ─── MATCH ON DICTIONARY STRUCTURE ─────────────────────────────
command = {"action": "move", "direction": "north"}

match command:
    case {"action": "move", "direction": d}:
        # 'd' is automatically filled with the value from the dict
        print(f"Moving {d}")         # "Moving north"
    case {"action": "attack", "weapon": w}:
        print(f"Attacking with {w}")
    case _:
        print("Unknown command")

# ─── MATCH ON TYPES ─────────────────────────────────────────────
def describe(value):
    match value:
        case int(n) if n > 0:       # matches positive integers
            return f"positive int: {n}"
        case int(n):                # matches any other integer
            return f"int: {n}"
        case str(s):                # matches any string
            return f"string of length {len(s)}"
        case [x, y]:                # matches a list with exactly 2 items
            return f"two-element list: {x}, {y}"
        case None:
            return "nothing"
        case _:
            return "something else"

Exceptions β€” Handling Errors Gracefully

When something goes wrong in Python (dividing by zero, accessing a file that doesn't exist, etc.), Python raises an exception. If you don't handle it, your program crashes. The try/except block lets you catch exceptions and respond to them instead of crashing.

Think of it like a safety net under a trapeze artist. Most of the time, everything is fine. But if something goes wrong (the exception), the safety net catches them and the show can continue. Without the net (no try/except), one mistake ends the whole performance (program crash).

python β€” exceptions
# ─── BASIC TRY/EXCEPT ────────────────────────────────────────────
try:
    result = 10 / 0           # this raises ZeroDivisionError
except ZeroDivisionError:
    print("Cannot divide by zero!")   # this runs instead of crashing

# ─── CATCHING MULTIPLE EXCEPTION TYPES ──────────────────────────
try:
    number = int("not a number")   # raises ValueError
except ZeroDivisionError as e:
    print(f"Division error: {e}")
except ValueError as e:
    print(f"Value error: {e}")     # this runs
except (TypeError, AttributeError) as e:
    print(f"Type or attribute error: {e}")  # catch multiple in one line

# ─── ELSE AND FINALLY ────────────────────────────────────────────
try:
    result = 10 / 2
except ZeroDivisionError:
    print("division failed")
else:
    # 'else' runs ONLY if NO exception was raised
    print(f"Success! Result: {result}")   # this runs
finally:
    # 'finally' ALWAYS runs β€” whether or not an exception happened
    # Use it for cleanup (closing files, database connections, etc.)
    print("This always runs β€” cleanup here")

# ─── RAISING YOUR OWN EXCEPTIONS ────────────────────────────────
def set_age(age):
    if not isinstance(age, int):
        raise TypeError("Age must be an integer")    # raise stops execution
    if age < 0 or age > 150:
        raise ValueError("Age must be between 0 and 150")
    return age

# ─── CUSTOM EXCEPTION CLASSES ────────────────────────────────────
# Create your own exception types by inheriting from Exception
class ValidationError(ValueError):
    """Raised when user input fails validation."""
    def __init__(self, field, message):
        # Call the parent Exception's __init__ with a formatted message
        super().__init__(f"Field '{field}': {message}")
        self.field = field    # store the field name for later use

def validate_email(email):
    if "@" not in email:
        raise ValidationError("email", "must contain @")

try:
    validate_email("notanemail")
except ValidationError as e:
    print(e)           # "Field 'email': must contain @"
    print(e.field)     # "email" β€” we can access the field name

βš™οΈFunctions

What is a Function?

A function is a named, reusable block of code that does a specific job. You define it once, and then you can call it as many times as you want from anywhere in your program.

Functions help you avoid repeating yourself. Instead of writing the same 10 lines in 5 different places, you write them once inside a function and call the function 5 times.

A function is like a recipe. You write the recipe once (define the function). Whenever you want to cook that dish (call the function), you follow the same steps. You can cook it with different ingredients each time (different arguments) β€” but the method stays the same.

The parameters in the recipe are like "X cups of flour" β€” they're placeholders. The arguments are the actual amounts you use when you cook β€” "2 cups of flour today, 3 cups tomorrow".

python β€” function basics
# ─── DEFINING A FUNCTION ────────────────────────────────────────
# 'def' keyword tells Python "I'm defining a function"
# 'greet' is the function name
# 'name' is the PARAMETER (placeholder variable)
def greet(name):
    message = f"Hello, {name}!"    # build a message
    return message                 # 'return' sends a value back to the caller

# ─── CALLING A FUNCTION ─────────────────────────────────────────
result = greet("Alice")    # "Alice" is the ARGUMENT (actual value)
print(result)              # "Hello, Alice!"

# You can also use the return value directly without storing it
print(greet("Bob"))        # "Hello, Bob!"

# ─── DEFAULT ARGUMENTS ──────────────────────────────────────────
# You can give parameters a default value β€” used if caller doesn't provide one
def greet_with_title(name, title="Mr"):
    return f"Hello, {title} {name}!"

print(greet_with_title("Smith"))           # "Hello, Mr Smith!" β€” uses default
print(greet_with_title("Smith", "Dr"))     # "Hello, Dr Smith!" β€” overrides default

# ─── KEYWORD ARGUMENTS ───────────────────────────────────────────
# You can pass arguments by NAME, in any order
def create_profile(name, age, city):
    return f"{name}, {age}, from {city}"

create_profile("Alice", 30, "London")            # positional β€” order matters
create_profile(age=30, city="London", name="Alice")  # keyword β€” any order

# ─── *args β€” accept any number of positional arguments ──────────
def add_all(*numbers):
    # numbers becomes a TUPLE of all arguments passed
    total = 0
    for n in numbers:
        total += n
    return total

print(add_all(1, 2, 3))       # 6
print(add_all(1, 2, 3, 4, 5)) # 15

# ─── **kwargs β€” accept any number of keyword arguments ──────────
def show_info(**details):
    # details becomes a DICT of all keyword arguments passed
    for key, value in details.items():
        print(f"  {key}: {value}")

show_info(name="Alice", age=30, city="London")
# name: Alice
# age: 30
# city: London

# ─── RETURN MULTIPLE VALUES ─────────────────────────────────────
def min_and_max(numbers):
    return min(numbers), max(numbers)   # returns a TUPLE of two values

lowest, highest = min_and_max([3, 1, 4, 1, 5, 9])
print(f"Min: {lowest}, Max: {highest}")   # Min: 1, Max: 9

Mutable Default Arguments β€” a classic beginner trap!

If you use a list or dict as a default argument, it is created ONCE when the function is defined β€” not each time the function is called. This means all calls share the same object.

python β€” BUG (mutable default)
# BAD: the default list [] is created
# only ONCE and shared across all calls
def add_item(item, items=[]):
    items.append(item)
    return items

add_item("a")  # ["a"]
add_item("b")  # ["a", "b"] ← unexpected!
python β€” FIX (use None sentinel)
# GOOD: use None as default, create
# a new list inside the function each time
def add_item(item, items=None):
    if items is None:
        items = []     # fresh list each call
    items.append(item)
    return items

add_item("a")  # ["a"]
add_item("b")  # ["b"] ← correct!
python β€” lambdas and higher-order functions
# ─── LAMBDA β€” tiny anonymous (unnamed) functions ─────────────────
# Use when you need a simple function for a short time
# Format: lambda parameters: expression
double = lambda x: x * 2
print(double(5))   # 10

# Very common use: sorting by a custom key
people = [
    {"name": "Charlie", "age": 35},
    {"name": "Alice",   "age": 28},
    {"name": "Bob",     "age": 32}
]
# Sort by age β€” the lambda says "use the 'age' field as the sort key"
sorted_people = sorted(people, key=lambda p: p["age"])
# [Alice:28, Bob:32, Charlie:35]

# ─── map() β€” apply a function to every item ─────────────────────
nums = [1, 2, 3, 4, 5]
# map() applies the function to each item, returns a lazy iterator
doubled = list(map(lambda x: x * 2, nums))   # [2, 4, 6, 8, 10]

# ─── filter() β€” keep only items that pass a test ────────────────
evens = list(filter(lambda x: x % 2 == 0, nums))  # [2, 4]

# ─── CLOSURES β€” functions that remember their environment ────────
# A closure is a function that captures variables from its surrounding scope
def make_multiplier(factor):
    # 'factor' lives in the outer function's scope
    def multiply(number):
        return number * factor   # 'multiply' remembers 'factor'!
    return multiply              # return the inner function itself

double  = make_multiplier(2)    # factor=2 is "captured" in this function
triple  = make_multiplier(3)    # factor=3 is "captured" in this function
print(double(5))   # 10
print(triple(5))   # 15

# ─── functools.partial β€” pre-fill some arguments ────────────────
from functools import partial

def power(base, exponent):
    return base ** exponent

square = partial(power, exponent=2)   # exponent is now always 2
cube   = partial(power, exponent=3)   # exponent is now always 3
print(square(4))    # 16
print(cube(3))      # 27

πŸ“Modules, Files & Standard Library

What is a Module?

As your code grows, you can't put everything in one file. A module is simply a Python file. A package is a folder of Python files. You can import code from other modules β€” Python comes with hundreds of built-in modules in its "standard library", and you can install thousands more with pip.

python β€” imports and useful stdlib
# ─── IMPORTING MODULES ──────────────────────────────────────────
import os                          # import the whole 'os' module
from pathlib import Path           # import just the 'Path' class from pathlib
from typing import Optional, Union # import multiple names with a comma

import os.path as osp              # import and give it a shorter alias

# ─── USEFUL STDLIB MODULES ──────────────────────────────────────
from pathlib import Path
from datetime import datetime, timedelta
from collections import Counter, defaultdict, deque
from functools import lru_cache, cache, partial, wraps
import json, re, math, random, time

# ─── pathlib β€” working with file paths ──────────────────────────
# Old way: os.path.join("folder", "file.txt") β€” messy
# New way with pathlib:
base = Path(".")                    # current directory
config = base / "config.json"       # / operator joins paths!
config.exists()                     # True or False
config.read_text()                  # read the whole file as a string
config.write_text("hello")          # write text to the file
list(base.glob("**/*.py"))          # find all .py files recursively

# ─── datetime β€” working with dates and times ─────────────────────
now = datetime.now()
deadline = now + timedelta(days=7)  # 7 days from now
formatted = now.strftime("%Y-%m-%d %H:%M")   # "2025-03-15 14:30"
print(formatted)

# ─── json β€” working with JSON data ──────────────────────────────
import json
data = {"name": "Alice", "scores": [90, 85, 92]}
json_string = json.dumps(data, indent=2)   # Python dict β†’ JSON string
loaded = json.loads(json_string)            # JSON string β†’ Python dict

# ─── The __name__ guard ─────────────────────────────────────────
# When Python runs a file directly, __name__ is "__main__"
# When a file is imported, __name__ is the module name
# This lets you have code that only runs when the file is run directly:
if __name__ == "__main__":
    print("Running directly, not imported")

# ─── FILE I/O ────────────────────────────────────────────────────
# Always use 'with' β€” it automatically closes the file, even on errors
with open("data.txt", "w", encoding="utf-8") as f:
    f.write("Hello\n")           # write a line
    f.writelines(["Line 1\n", "Line 2\n"])  # write multiple lines

with open("data.txt", "r", encoding="utf-8") as f:
    content = f.read()           # read entire file as one string
    # OR: lines = f.readlines()  # read into a list, one item per line

# Lazy reading β€” processes one line at a time (memory-efficient for huge files)
for line in open("data.txt", encoding="utf-8"):
    print(line.strip())          # .strip() removes the newline at the end

# ─── VIRTUAL ENVIRONMENTS (how to manage packages) ──────────────
# python -m venv .venv              # create a virtual environment
# source .venv/bin/activate         # activate it (Linux/Mac)
# .venv\Scripts\activate            # activate it (Windows)
# pip install langchain fastapi     # install packages
# pip freeze > requirements.txt     # save your package list
# pip install -r requirements.txt   # install from saved list
python β€” context managers
from contextlib import contextmanager
import time

# Context managers run code before AND after a 'with' block
# The 'with' statement calls __enter__ before and __exit__ after
# This guarantees cleanup even if an error occurs

# ─── CREATING YOUR OWN CONTEXT MANAGER with @contextmanager ─────
@contextmanager
def timer(label=""):
    """Measures and prints the execution time of a block of code."""
    start = time.perf_counter()         # record start time
    try:
        yield                           # the 'with' block runs here
    finally:
        # 'finally' always runs, even if an error was raised in the block
        elapsed = time.perf_counter() - start
        print(f"{label} took {elapsed:.4f}s")

# Usage:
with timer("database query"):
    # ... do some slow work ...
    result = sum(range(1_000_000))
# Prints: "database query took 0.0423s"

# ─── CLASS-BASED CONTEXT MANAGER ─────────────────────────────────
class DatabaseConnection:
    def __init__(self, url):
        self.url = url
        self.conn = None

    def __enter__(self):
        # This runs when you enter the 'with' block
        print(f"Connecting to {self.url}...")
        self.conn = {"connected": True}  # simulate a real connection
        return self.conn                 # this becomes the 'as' variable

    def __exit__(self, exc_type, exc_val, exc_tb):
        # This runs when you EXIT the 'with' block (even on error)
        print("Closing connection...")
        self.conn = None
        return False   # False means "don't suppress any exceptions"

with DatabaseConnection("postgresql://localhost/mydb") as db:
    print(f"Working with: {db}")
# Automatically closes when the 'with' block ends

πŸ—οΈObject-Oriented Programming (OOP)

What is a Class?

A class is a blueprint or template for creating objects. Think of a house blueprint β€” the blueprint isn't a house, it's just the design. But you can use that blueprint to build many actual houses. Each house built from the same blueprint is an instance (also called an object).

In programming, a class defines: what data an object holds (called attributes), and what actions an object can perform (called methods).

Class = Blueprint. Object (Instance) = The actual thing built from the blueprint.

The Dog class is the blueprint. rex = Dog("Rex") creates a real Dog object named rex. buddy = Dog("Buddy") creates another real Dog object. They come from the same blueprint but are completely separate β€” rex and buddy are different dogs.

Defining a Class Step by Step

Let's build a Dog class from scratch and explain each piece:

python β€” class basics with detailed comments
class Dog:
    # ─── CLASS VARIABLE ─────────────────────────────────────────
    # This belongs to the CLASS itself, not any individual dog
    # ALL dogs share this same value
    species = "Canis lupus familiaris"

    # ─── __init__ β€” the "constructor" method ────────────────────
    # This runs AUTOMATICALLY every time you create a new Dog
    # It's how you set up the initial state of the object
    # 'self' = the object being created (Python passes it automatically)
    def __init__(self, name: str, age: int, breed: str):
        # ─── INSTANCE VARIABLES ───────────────────────────────
        # These belong to the INDIVIDUAL dog (each dog has their own)
        # 'self.name' means "this dog's name"
        self.name  = name       # store the name in the object
        self.age   = age        # store the age in the object
        self.breed = breed      # store the breed in the object
        self.tricks = []        # each dog starts with an empty list of tricks

    # ─── INSTANCE METHOD ────────────────────────────────────────
    # 'self' is always the first parameter β€” Python passes the object automatically
    # When you call rex.bark(), Python calls Dog.bark(rex) under the hood
    def bark(self):
        return f"{self.name} says: Woof!"

    def learn_trick(self, trick: str):
        self.tricks.append(trick)   # add a trick to THIS dog's list
        print(f"{self.name} learned '{trick}'!")

    def show_tricks(self):
        if not self.tricks:
            print(f"{self.name} knows no tricks yet.")
        else:
            tricks_str = ", ".join(self.tricks)
            print(f"{self.name} can: {tricks_str}")

    # ─── CLASS METHOD β€” operates on the class, not an instance ──
    # @classmethod means Python passes the CLASS (not an instance) as first arg
    # 'cls' is convention (short for class) β€” you could name it anything
    @classmethod
    def from_birth_year(cls, name: str, birth_year: int, breed: str) -> "Dog":
        """Create a Dog using birth year instead of current age."""
        current_year = 2025
        age = current_year - birth_year
        return cls(name, age, breed)   # cls() is like calling Dog()

    # ─── STATIC METHOD β€” a utility function related to the class ─
    # No 'self' or 'cls' β€” doesn't need the object or class
    @staticmethod
    def is_valid_name(name: str) -> bool:
        """Check if a name is valid (not empty and not just spaces)."""
        return bool(name and name.strip())

    # ─── __repr__ β€” how the object appears when you print it ────
    def __repr__(self) -> str:
        return f"Dog(name={self.name!r}, age={self.age}, breed={self.breed!r})"


# ─── CREATING OBJECTS (INSTANCES) ────────────────────────────────
rex   = Dog("Rex", 3, "German Shepherd")   # __init__ is called automatically
buddy = Dog("Buddy", 5, "Labrador")

# ─── USING OBJECTS ───────────────────────────────────────────────
print(rex.bark())          # "Rex says: Woof!"
print(buddy.bark())        # "Buddy says: Woof!"

rex.learn_trick("sit")
rex.learn_trick("roll over")
rex.show_tricks()          # "Rex can: sit, roll over"
buddy.show_tricks()        # "Buddy knows no tricks yet."

# rex and buddy each have their OWN tricks list
print(rex.tricks)          # ["sit", "roll over"]
print(buddy.tricks)        # []  β€” buddy's list is separate!

# ─── ACCESSING CLASS VARIABLES ───────────────────────────────────
print(Dog.species)         # "Canis lupus familiaris"  β€” through the class
print(rex.species)         # "Canis lupus familiaris"  β€” also works through instance

# ─── CLASSMETHOD AND STATICMETHOD ────────────────────────────────
luna = Dog.from_birth_year("Luna", 2021, "Poodle")   # creates a Dog with age 4
print(Dog.is_valid_name("Rex"))   # True
print(Dog.is_valid_name(""))      # False

print(repr(rex))  # Dog(name='Rex', age=3, breed='German Shepherd')

Inheritance β€” Building on Existing Classes

Inheritance lets one class extend another. The child class gets all the methods and attributes of the parent class, and can add new ones or override existing ones.

Imagine a Vehicle class. A Car is a Vehicle, a Truck is a Vehicle, a Motorcycle is a Vehicle. They all share common features (engine, wheels, can drive) but each has unique features too. Rather than copying code, you inherit β€” the Car class says "I'm like a Vehicle, plus I also have trunk space".

python β€” inheritance and super()
class Animal:
    """Base class for all animals."""
    def __init__(self, name: str):
        self.name = name

    def speak(self) -> str:
        """Override this in child classes."""
        return f"{self.name} makes a sound"

    def describe(self) -> str:
        return f"I am {self.name}"


# ─── CHILD CLASS β€” inherits from Animal ──────────────────────────
class Dog(Animal):
    def __init__(self, name: str, breed: str):
        # super().__init__ calls the PARENT class's __init__
        # This sets self.name (defined in Animal) for us
        # Without this, the Animal's __init__ wouldn't run
        super().__init__(name)
        self.breed = breed    # additional attribute specific to Dog

    def speak(self) -> str:
        # OVERRIDING the parent's speak() method
        # Dog has its own version of speak()
        return f"{self.name} says: Woof!"

    def fetch(self, item: str) -> str:
        return f"{self.name} fetches the {item}!"


class Cat(Animal):
    def speak(self) -> str:
        return f"{self.name} says: Meow!"


# ─── POLYMORPHISM β€” same interface, different behaviour ──────────
# All animals have a speak() method, but each acts differently
animals = [Dog("Rex", "Husky"), Cat("Whiskers"), Dog("Buddy", "Poodle")]
for animal in animals:
    print(animal.speak())   # Each calls its own version of speak()
# Rex says: Woof!
# Whiskers says: Meow!
# Buddy says: Woof!

# ─── isinstance() β€” check what type an object is ─────────────────
rex = Dog("Rex", "Husky")
print(isinstance(rex, Dog))      # True β€” rex IS a Dog
print(isinstance(rex, Animal))   # True β€” Dog inherits from Animal, so rex IS also an Animal
print(isinstance(rex, Cat))      # False

# ─── ABSTRACT BASE CLASS β€” force child classes to implement a method ──
from abc import ABC, abstractmethod

class Shape(ABC):
    """Abstract base class. Cannot be instantiated directly."""

    @abstractmethod
    def area(self) -> float:
        """Every Shape MUST implement this method."""
        ...   # ... means "no body β€” child must provide it"

    @abstractmethod
    def perimeter(self) -> float: ...

class Circle(Shape):
    def __init__(self, radius: float):
        self.radius = radius

    def area(self) -> float:
        return 3.14159 * self.radius ** 2

    def perimeter(self) -> float:
        return 2 * 3.14159 * self.radius

class Rectangle(Shape):
    def __init__(self, width: float, height: float):
        self.width = width
        self.height = height

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

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

# Shape()  # TypeError! Can't instantiate an abstract class
c = Circle(5)
r = Rectangle(4, 6)
print(c.area())       # 78.53975
print(r.area())       # 24

Dunder (Magic) Methods β€” Customizing Built-in Behaviour

Dunder methods (short for "double underscore") are special methods that Python calls automatically in response to certain operations. They let your custom classes work with Python's built-in syntax like +, len(), print(), etc.

python β€” dunder methods
class Vector:
    """A 2D mathematical vector."""
    def __init__(self, x: float, y: float):
        self.x = x
        self.y = y

    def __repr__(self) -> str:
        # Called when Python needs a "developer-friendly" representation
        # Used in the REPL and for debugging
        return f"Vector({self.x}, {self.y})"

    def __str__(self) -> str:
        # Called when you print() or str() an object
        # Should be a user-friendly string
        return f"({self.x}, {self.y})"

    def __add__(self, other: "Vector") -> "Vector":
        # Called when you use the + operator: v1 + v2
        return Vector(self.x + other.x, self.y + other.y)

    def __sub__(self, other: "Vector") -> "Vector":
        # Called when you use the - operator: v1 - v2
        return Vector(self.x - other.x, self.y - other.y)

    def __mul__(self, scalar: float) -> "Vector":
        # Called when you use *: v * 3  (Vector on the LEFT)
        return Vector(self.x * scalar, self.y * scalar)

    def __rmul__(self, scalar: float) -> "Vector":
        # Called when Vector is on the RIGHT: 3 * v
        return self.__mul__(scalar)

    def __len__(self) -> int:
        # Called when you use len(): len(v)
        return 2   # a 2D vector always has 2 components

    def __getitem__(self, index: int) -> float:
        # Called when you use indexing: v[0], v[1]
        return (self.x, self.y)[index]

    def __eq__(self, other: "Vector") -> bool:
        # Called when you use ==: v1 == v2
        return self.x == other.x and self.y == other.y

    def __abs__(self) -> float:
        # Called when you use abs(): abs(v)
        # Returns the length (magnitude) of the vector
        return (self.x**2 + self.y**2) ** 0.5

    def __bool__(self) -> bool:
        # Called in an if statement: if v:
        # A zero vector (0, 0) is falsy; any other vector is truthy
        return self.x != 0 or self.y != 0


v1 = Vector(3, 4)
v2 = Vector(1, 2)

print(v1)               # (3, 4) β€” calls __str__
print(repr(v1))         # Vector(3, 4) β€” calls __repr__
print(v1 + v2)          # (4, 6) β€” calls __add__
print(v1 - v2)          # (2, 2) β€” calls __sub__
print(v1 * 2)           # (6, 8) β€” calls __mul__
print(3 * v1)           # (9, 12) β€” calls __rmul__
print(abs(v1))          # 5.0 β€” calls __abs__ (Pythagorean theorem: √(3Β²+4Β²)=5)
print(len(v1))          # 2 β€” calls __len__
print(v1[0])            # 3 β€” calls __getitem__
print(v1 == Vector(3,4))# True β€” calls __eq__

Properties β€” Controlled Access to Attributes

A property looks like a regular attribute from the outside (you access it without parentheses like obj.value) but behind the scenes it runs a function. This lets you add validation or computation without changing the interface.

python β€” properties
class Temperature:
    """Temperature stored in Celsius, accessible in both scales."""
    def __init__(self, celsius: float = 0):
        # Store the value with a "private" name (leading underscore is convention)
        # The leading _ signals "internal use only" but doesn't enforce it
        self._celsius = celsius

    @property
    def celsius(self) -> float:
        """Getter β€” runs when you READ: temp.celsius"""
        return self._celsius

    @celsius.setter
    def celsius(self, value: float):
        """Setter β€” runs when you WRITE: temp.celsius = 25"""
        if value < -273.15:
            raise ValueError("Temperature cannot go below absolute zero!")
        self._celsius = value

    @property
    def fahrenheit(self) -> float:
        """A read-only computed property β€” converts Celsius to Fahrenheit."""
        # No setter means this property is READ-ONLY
        return self._celsius * 9/5 + 32

    @classmethod
    def from_fahrenheit(cls, f: float) -> "Temperature":
        """Alternative constructor β€” create from a Fahrenheit value."""
        return cls((f - 32) * 5/9)  # convert to Celsius and create instance


t = Temperature(100)
print(t.celsius)      # 100 β€” uses the getter
print(t.fahrenheit)   # 212.0 β€” computed automatically
t.celsius = 25        # uses the setter (runs the validation)
print(t.fahrenheit)   # 77.0

# t.celsius = -300    # ValueError! Below absolute zero

boiling = Temperature.from_fahrenheit(212)  # 100Β°C
print(boiling.celsius)   # 100.0

πŸ”¬Decorators & Generators

What is a Decorator?

A decorator is a function that takes another function as input, adds some extra behaviour around it, and returns a modified version. The @decorator syntax is just a shortcut β€” @timer above a function means "take this function, pass it to timer(), and replace it with what timer() returns".

Imagine you run a coffee shop and you make plain coffee. A decorator is like a machine that takes ANY cup of coffee and automatically wraps it in fancy packaging before handing it to the customer. The coffee itself doesn't change β€” it just gets extra wrapping.

Similarly, a decorator wraps your function with extra behaviour (like timing it, logging it, checking permissions) without changing the function itself.

python β€” decorators explained step by step
from functools import wraps  # preserves the wrapped function's metadata
import time

# ─── STEP 1: Understand what a decorator IS ─────────────────────
# A decorator is just a function that wraps another function
def timer(func):
    # 'func' is the function being decorated (e.g., slow_task)
    @wraps(func)   # makes the wrapper "look like" the original function
    def wrapper(*args, **kwargs):
        # CODE HERE runs BEFORE the original function
        start = time.perf_counter()

        # Call the original function with all its arguments
        result = func(*args, **kwargs)

        # CODE HERE runs AFTER the original function
        elapsed = time.perf_counter() - start
        print(f"{func.__name__} took {elapsed:.4f} seconds")

        return result  # pass back whatever the original function returned
    return wrapper     # return the wrapper function (NOT the result of calling it!)

# ─── STEP 2: Using the decorator with @ syntax ──────────────────
@timer
def slow_task():
    time.sleep(0.1)
    return "done"

# The @timer line is EXACTLY the same as writing:
# slow_task = timer(slow_task)

# Now when you call slow_task(), it actually calls wrapper()
result = slow_task()   # prints: "slow_task took 0.1002 seconds"

# ─── DECORATOR WITH ARGUMENTS ───────────────────────────────────
# To pass arguments to a decorator, you need THREE levels of functions:
# 1. Decorator factory (takes the config args)
# 2. Decorator (takes the function)
# 3. Wrapper (calls the function with retry logic)

def retry(max_attempts=3, delay=1.0):
    """Decorator that retries a function on failure."""
    def decorator(func):              # this is the actual decorator
        @wraps(func)
        def wrapper(*args, **kwargs):  # this wraps the function
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)  # try to call the function
                except Exception as e:
                    if attempt == max_attempts:
                        raise  # re-raise on final attempt
                    print(f"Attempt {attempt} failed: {e}. Retrying...")
                    time.sleep(delay)
        return wrapper
    return decorator

@retry(max_attempts=3, delay=0.5)   # first call retry() β†’ returns decorator
def fetch_data(url: str):            # then decorator applied to fetch_data
    # Simulates an API call that might fail
    import random
    if random.random() < 0.7:
        raise ConnectionError("Network error")
    return f"Data from {url}"

# ─── STACKING DECORATORS ────────────────────────────────────────
# Decorators are applied BOTTOM-UP (innermost first)
@timer            # applied second (outermost)
@retry(max_attempts=2)  # applied first (innermost)
def unreliable_api():
    pass
# This is equivalent to: unreliable_api = timer(retry(max_attempts=2)(unreliable_api))

# ─── BUILT-IN DECORATORS ────────────────────────────────────────
from functools import lru_cache, cached_property

@lru_cache(maxsize=128)   # caches results to avoid recomputation
def fibonacci(n: int) -> int:
    """Compute fibonacci without repeating work (memoization)."""
    if n < 2:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

print(fibonacci(50))    # fast! results are cached

class Circle:
    def __init__(self, radius: float):
        self.radius = radius

    @cached_property   # computed once and cached as an attribute
    def area(self) -> float:
        print("Computing area...")  # only prints once!
        return 3.14159 * self.radius ** 2

c = Circle(5)
print(c.area)   # "Computing area..." then 78.53975
print(c.area)   # 78.53975 (no recomputation β€” cached!)

Generators β€” Lazy Evaluation

A generator is a special kind of function that produces values one at a time, on demand, instead of computing everything upfront and returning a big list. This saves enormous amounts of memory when working with large datasets.

Imagine you need to serve 1,000 meals at a banquet. You could cook ALL 1,000 meals at once, pile them on a table, and then serve them (a normal function returning a list β€” uses tons of space). OR you could have a chef at the table who cooks each meal right when someone asks for it (a generator β€” uses almost no space). Same result, massively different resource usage.

python β€” generators
# ─── NORMAL FUNCTION vs GENERATOR ───────────────────────────────
def squares_list(n):
    """Regular function: computes ALL squares, stores them in memory."""
    result = []
    for x in range(n):
        result.append(x ** 2)
    return result  # returns a complete list β€” all in memory at once

def squares_generator(n):
    """Generator function: produces ONE square at a time."""
    for x in range(n):
        yield x ** 2   # 'yield' pauses the function and returns one value
                       # next time next() is called, execution resumes HERE

# The difference in memory usage is huge for large n:
squares_list(1_000_000)      # allocates ~8 MB for the list
squares_generator(1_000_000) # uses almost zero memory!

# ─── USING A GENERATOR ───────────────────────────────────────────
gen = squares_generator(5)
print(next(gen))   # 0  β€” asks for the first value
print(next(gen))   # 1  β€” asks for the second value
print(next(gen))   # 4  β€” asks for the third value
# ... and so on. After all values are exhausted, StopIteration is raised.

# You can also use a generator in a for loop (most common usage)
for square in squares_generator(10):
    print(square)   # 0, 1, 4, 9, 16, 25, 36, 49, 64, 81

# ─── FIBONACCI NUMBER GENERATOR ──────────────────────────────────
def fibonacci_gen(limit):
    """Generates Fibonacci numbers up to a limit."""
    a, b = 0, 1
    while a <= limit:
        yield a          # pause and return 'a'
        a, b = b, a + b  # then swap: new_a = old_b, new_b = old_a + old_b

print(list(fibonacci_gen(50)))   # [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]

# ─── INFINITE GENERATOR ──────────────────────────────────────────
def count_forever(start=0):
    """An infinite sequence β€” generates numbers forever."""
    n = start
    while True:    # this runs forever β€” that's OK for a generator!
        yield n
        n += 1

# Get just the first 5 numbers from an infinite sequence
from itertools import islice
first_five = list(islice(count_forever(10), 5))  # [10, 11, 12, 13, 14]

# ─── GENERATOR EXPRESSION β€” like a list comprehension, but lazy ──
# List comprehension β€” computes everything NOW:
squares_list_expr = [x**2 for x in range(1_000_000)]  # 8 MB

# Generator expression β€” computes on demand:
squares_gen_expr = (x**2 for x in range(1_000_000))   # near-zero memory
# Just change [] to () to get a generator

total = sum(squares_gen_expr)   # sum without storing all values!

# ─── yield from β€” delegating to another generator ────────────────
def flatten(nested):
    """Flattens a nested list structure."""
    for item in nested:
        if isinstance(item, list):
            yield from flatten(item)  # delegate to recursive call
        else:
            yield item

result = list(flatten([1, [2, [3, 4]], [5, 6], 7]))
print(result)  # [1, 2, 3, 4, 5, 6, 7]

Metaclasses & Descriptors

Metaclasses are "classes that create classes" β€” they control what happens when a class itself is defined. Descriptors are objects that control attribute access. These are advanced Python features, mostly used in frameworks.

python β€” metaclasses and descriptors
# ─── SINGLETON WITH METACLASS ────────────────────────────────────
# A Singleton class allows only ONE instance to ever exist
class SingletonMeta(type):
    """Metaclass that ensures only one instance exists."""
    _instances = {}   # class-level dict tracking instances

    def __call__(cls, *args, **kwargs):
        # This runs whenever you try to create a new instance: cls()
        if cls not in cls._instances:
            # First time β€” create and store the instance
            cls._instances[cls] = super().__call__(*args, **kwargs)
        # Always return the stored instance (same one every time)
        return cls._instances[cls]

class DatabasePool(metaclass=SingletonMeta):
    """Only one connection pool ever exists."""
    def __init__(self):
        self.connections = []
        print("Creating database pool...")  # only prints once!

pool1 = DatabasePool()   # creates new instance
pool2 = DatabasePool()   # returns the SAME instance
print(pool1 is pool2)    # True β€” they are literally the same object!

# ─── DESCRIPTOR β€” custom attribute access ────────────────────────
class PositiveNumber:
    """A descriptor that only accepts positive numbers."""
    def __set_name__(self, owner, name):
        self.name = f"_{name}"   # internal storage name with underscore

    def __get__(self, obj, objtype=None):
        if obj is None:
            return self   # when accessed on the class itself
        return getattr(obj, self.name, 0)   # return stored value or 0

    def __set__(self, obj, value):
        if not isinstance(value, (int, float)) or value <= 0:
            raise ValueError(f"{self.name} must be a positive number")
        setattr(obj, self.name, value)   # store the validated value

class Product:
    price = PositiveNumber()      # price must always be positive
    quantity = PositiveNumber()   # quantity must always be positive

    def __init__(self, name: str, price: float, quantity: int):
        self.name = name
        self.price = price          # goes through __set__ β€” validates!
        self.quantity = quantity

p = Product("Widget", 9.99, 100)
print(p.price)      # 9.99
# p.price = -5      # ValueError! Must be positive

⚑Async / Await

Why Do We Need Async?

Most programs spend a lot of time waiting β€” waiting for a database to respond, waiting for a web request to return, waiting for a file to load. In a regular (synchronous) program, while you wait, the whole program is frozen. Nothing else can happen.

Async programming lets your program do other things while waiting. When you start an operation that will take time (like a network request), instead of freezing and waiting, the program makes a note to "come back to this when it's ready" and goes off to do other work.

Imagine a restaurant with one waiter (your program's thread).

In a synchronous restaurant: When table 1 orders coffee (which takes 5 minutes to brew), the waiter stands at table 1 doing NOTHING for 5 minutes. Tables 2 and 3 sit there completely ignored. Terrible service.

In an asynchronous restaurant: When table 1 orders coffee, the waiter starts the coffee machine, then immediately goes to take table 2's order, then checks on table 3, then comes back when the coffee is ready. Same single waiter, massively better service.

The waiter is your event loop. Starting the coffee machine is the await β€” it starts the operation and lets the waiter go do other things.

python β€” asyncio fundamentals
import asyncio

# ─── async def β€” defines an asynchronous function (coroutine) ────
# You CANNOT call it like a regular function and get the result
# You must 'await' it from inside another async function
async def fetch_user(user_id: int) -> dict:
    """Simulate fetching a user from a database β€” takes 0.1 seconds."""
    await asyncio.sleep(0.1)   # 'await' means "pause here, let others run"
    return {"id": user_id, "name": f"user_{user_id}"}

# ─── SEQUENTIAL vs CONCURRENT ────────────────────────────────────
async def sequential():
    """Fetches users one at a time β€” total time: 0.3 seconds."""
    user1 = await fetch_user(1)   # wait 0.1s for user 1
    user2 = await fetch_user(2)   # THEN wait 0.1s for user 2
    user3 = await fetch_user(3)   # THEN wait 0.1s for user 3
    return [user1, user2, user3]

async def concurrent():
    """Fetches all users at the same time β€” total time: ~0.1 seconds!"""
    # asyncio.gather() starts ALL coroutines and waits for ALL to finish
    results = await asyncio.gather(
        fetch_user(1),   # starts immediately
        fetch_user(2),   # starts immediately
        fetch_user(3),   # starts immediately
    )
    # All three run at the SAME time β€” total wait is just 0.1s, not 0.3s!
    return results

# asyncio.run() β€” the entry point, creates and runs the event loop
asyncio.run(concurrent())

# ─── TASKS β€” fire and forget, run in background ──────────────────
async def background_job(name: str):
    await asyncio.sleep(2)
    print(f"Job '{name}' completed!")

async def main():
    # create_task() starts the coroutine IMMEDIATELY without waiting
    task1 = asyncio.create_task(background_job("email"))
    task2 = asyncio.create_task(background_job("report"))

    print("Tasks started, doing other work...")
    await asyncio.sleep(0.5)    # do some other work

    # Now wait for both tasks to finish
    await task1
    await task2

# ─── ASYNC CONTEXT MANAGER ───────────────────────────────────────
class AsyncDatabase:
    async def __aenter__(self):
        await asyncio.sleep(0.01)   # async connection setup
        print("Connected to database")
        return self

    async def __aexit__(self, *args):
        await asyncio.sleep(0.01)   # async cleanup
        print("Disconnected from database")

    async def query(self, sql: str) -> list:
        await asyncio.sleep(0.05)   # simulate async query
        return [{"id": 1}, {"id": 2}]

async def use_database():
    async with AsyncDatabase() as db:
        results = await db.query("SELECT * FROM users")
        return results

# ─── ASYNC GENERATOR ─────────────────────────────────────────────
async def stream_numbers(count: int):
    """Yields numbers one at a time with a delay between each."""
    for i in range(count):
        await asyncio.sleep(0.1)   # simulate streaming data
        yield i

async def consume_stream():
    async for number in stream_numbers(5):   # 'async for' for async generators
        print(number)   # 0, 1, 2, 3, 4 β€” arrives one at a time

# ─── TIMEOUT ─────────────────────────────────────────────────────
async def with_timeout():
    try:
        async with asyncio.timeout(2.0):   # Python 3.11+
            await asyncio.sleep(10)        # this will time out!
    except asyncio.TimeoutError:
        print("Operation timed out after 2 seconds")
python β€” asyncio with aiohttp (real HTTP requests)
import asyncio
import aiohttp   # pip install aiohttp β€” async HTTP library

async def fetch_url(session: aiohttp.ClientSession, url: str) -> str:
    """Fetch one URL asynchronously."""
    async with session.get(url) as response:
        response.raise_for_status()   # raises exception for 4xx/5xx errors
        return await response.text()  # await the response body

async def fetch_multiple(urls: list[str]) -> list[str]:
    """Fetch multiple URLs concurrently β€” much faster than one at a time."""
    # One shared session for all requests (efficient)
    async with aiohttp.ClientSession() as session:
        # Create one task per URL β€” all start at the same time
        tasks = [fetch_url(session, url) for url in urls]
        # gather() waits for ALL tasks and returns all results in order
        return await asyncio.gather(*tasks)

urls = [
    "https://httpbin.org/get",
    "https://httpbin.org/ip",
]
results = asyncio.run(fetch_multiple(urls))

When to use async? Use async when your code spends time waiting for external things: network requests (APIs, databases), file I/O on slow disks, waiting for user input. If your code is doing heavy CPU calculations (number crunching, image processing), async won't help β€” use multiprocessing for that instead.

🏷️Type Hints & Dataclasses

What are Type Hints?

Type hints are optional annotations that tell you (and your code editor) what type of data a variable or function parameter should hold. Python doesn't enforce them at runtime β€” they're documentation and tooling hints. Your editor uses them to catch bugs before you run your code.

python β€” type hints
from typing import Optional, Union, Any, TypeVar, Generic
from collections.abc import Callable, Sequence

# ─── BASIC ANNOTATIONS ───────────────────────────────────────────
name: str = "Alice"      # this variable should be a string
age:  int = 30           # this variable should be an integer
pi:   float = 3.14       # this variable should be a float

# Function annotations β€” parameter types and return type
def greet(name: str, times: int = 1) -> str:
    return (name + " ") * times
# The '-> str' says this function returns a string

# ─── OPTIONAL β€” can be None ──────────────────────────────────────
# Optional[str] means "either a string or None"
def find_user(user_id: int) -> Optional[str]:
    # Python 3.10+ alternative: str | None
    return None

# ─── UNION β€” one of several types ────────────────────────────────
# Python 3.10+: use X | Y instead of Union[X, Y]
def parse(value: str | int | None) -> int:
    if value is None:
        return 0
    return int(value)

# ─── COLLECTIONS ─────────────────────────────────────────────────
def process_items(items: list[int]) -> dict[str, int]:
    return {"sum": sum(items), "count": len(items)}

# ─── CALLABLE ────────────────────────────────────────────────────
# Callable[[arg_types], return_type]
def apply(fn: Callable[[int, int], int], a: int, b: int) -> int:
    return fn(a, b)

# ─── TYPEVAR β€” generics ──────────────────────────────────────────
T = TypeVar("T")   # T can be any type

def first_item(items: list[T]) -> T | None:
    """Works for list[int], list[str], list[Dog], etc."""
    return items[0] if items else None

# ─── GENERIC CLASS ───────────────────────────────────────────────
class Stack(Generic[T]):
    def __init__(self) -> None:
        self._items: list[T] = []

    def push(self, item: T) -> None:
        self._items.append(item)

    def pop(self) -> T:
        return self._items.pop()

int_stack = Stack[int]()    # a stack that holds integers
str_stack = Stack[str]()    # a stack that holds strings

# ─── TypedDict β€” a dict with a known structure ───────────────────
from typing import TypedDict

class UserDict(TypedDict):
    name: str
    age: int
    email: str | None

# ─── Protocol β€” duck typing with type safety ─────────────────────
from typing import Protocol

class Drawable(Protocol):
    def draw(self) -> None: ...   # any class with a draw() method qualifies

def render(shape: Drawable) -> None:
    shape.draw()   # works with ANY class that has draw(), no inheritance needed

Dataclasses β€” Automatic Boilerplate

Dataclasses automatically generate __init__, __repr__, and __eq__ methods based on your type annotations. This saves you from writing repetitive code.

python β€” dataclasses
from dataclasses import dataclass, field, asdict, astuple

# @dataclass automatically creates:
# - __init__(self, name, age, email, tags)
# - __repr__ that shows all fields
# - __eq__ that compares all fields
@dataclass
class User:
    name:  str
    age:   int
    email: str = ""                        # default value
    tags:  list[str] = field(default_factory=list)  # mutable default!
    # IMPORTANT: use field(default_factory=list) not tags: list[str] = []
    # Because [] as a default would be shared (the mutable default arg trap)

    def __post_init__(self):
        """Runs after __init__ β€” great for validation."""
        if self.age < 0:
            raise ValueError("Age cannot be negative")
        self.name = self.name.strip()   # clean up name

u = User("Alice", 30, "alice@example.com")
print(u)          # User(name='Alice', age=30, email='alice@example.com', tags=[])
print(asdict(u))  # {"name": "Alice", "age": 30, ...} β€” converts to dict

# ─── FROZEN DATACLASS β€” immutable ────────────────────────────────
@dataclass(frozen=True)
class Point:
    x: float
    y: float

p = Point(1.0, 2.0)
# p.x = 3.0   # FrozenInstanceError! Cannot modify a frozen dataclass
# Frozen dataclasses are hashable β€” can be used as dict keys or in sets!
d = {Point(0, 0): "origin", Point(1, 1): "one-one"}

# ─── ORDERED COMPARISON ──────────────────────────────────────────
@dataclass(order=True)
class Version:
    major: int
    minor: int
    patch: int

v1 = Version(1, 2, 3)
v2 = Version(1, 3, 0)
v3 = Version(2, 0, 0)
print(v1 < v2)    # True β€” compared field by field
print(sorted([v3, v1, v2]))  # [Version(1,2,3), Version(1,3,0), Version(2,0,0)]

Pydantic β€” Runtime Validation

Pydantic is like a dataclass but with runtime validation. It actually checks the values when an object is created and raises errors if they don't match the type annotations or constraints. FastAPI uses Pydantic extensively for request/response models.

python β€” pydantic v2
from pydantic import BaseModel, Field, validator, model_validator
from typing import Optional

class Address(BaseModel):
    street: str
    city:   str
    country: str = "US"   # default value

class User(BaseModel):
    # Field() lets you add validation constraints and documentation
    name:  str = Field(min_length=1, max_length=100, description="User's full name")
    age:   int = Field(ge=0, le=150, description="Age in years")  # ge=greater equal, le=less equal
    email: str
    address: Optional[Address] = None   # nested model β€” None if not provided
    tags: list[str] = []

    @validator("name")
    @classmethod
    def clean_name(cls, v: str) -> str:
        """Strip whitespace from name."""
        return v.strip()

    @model_validator(mode="after")
    def check_adult_email(self) -> "User":
        """Adults (18+) must have an email address."""
        if self.age >= 18 and not self.email:
            raise ValueError("Adults must have an email address")
        return self

# ─── CREATING INSTANCES ──────────────────────────────────────────
user = User(name="  Alice  ", age=30, email="alice@example.com")
print(user.name)     # "Alice" β€” whitespace stripped by validator

# ─── VALIDATION ERRORS ───────────────────────────────────────────
try:
    invalid = User(name="", age=30, email="a@b.com")  # name too short!
except Exception as e:
    print(e)   # validation error details

# ─── SERIALIZATION ───────────────────────────────────────────────
user.model_dump()          # β†’ Python dict
user.model_dump_json()     # β†’ JSON string
User.model_validate({"name": "Bob", "age": 25, "email": "b@b.com"})  # from dict

# ─── SETTINGS MANAGEMENT ─────────────────────────────────────────
from pydantic_settings import BaseSettings   # pip install pydantic-settings

class AppSettings(BaseSettings):
    """Read settings from environment variables or .env file."""
    database_url: str         # must be set in environment
    secret_key:   str         # must be set in environment
    debug:        bool = False   # optional, defaults to False
    max_connections: int = 10

    class Config:
        env_file = ".env"     # also read from .env file

settings = AppSettings()   # reads from environment automatically

πŸ› οΈComprehensions & Builtins

Python's standard library has powerful tools for working with sequences and collections. The itertools and functools modules provide functional programming tools that make common data-processing tasks concise and efficient.

python β€” itertools and functools
from itertools import (
    chain, product, combinations, permutations,
    groupby, islice, accumulate, cycle, repeat, count
)

# ─── chain β€” combine multiple iterables into one ────────────────
list(chain([1, 2], [3, 4], [5]))        # [1, 2, 3, 4, 5]
list(chain.from_iterable([[1,2],[3,4]])) # [1, 2, 3, 4] β€” from nested list

# ─── product β€” cartesian product (like nested for loops) ────────
list(product("AB", "12"))   # [('A','1'),('A','2'),('B','1'),('B','2')]
list(product([0,1], repeat=3))  # all 3-bit binary numbers

# ─── combinations β€” choose r items, order doesn't matter ────────
list(combinations([1, 2, 3, 4], 2))
# [(1,2),(1,3),(1,4),(2,3),(2,4),(3,4)]

# ─── permutations β€” choose r items, order DOES matter ───────────
list(permutations("ABC", 2))   # [('A','B'),('A','C'),('B','A'),...]

# ─── islice β€” take first n items from ANY iterator ───────────────
from itertools import count
first_10 = list(islice(count(0), 10))   # [0,1,2,...,9] from infinite count

# ─── groupby β€” group consecutive items with the same key ────────
# IMPORTANT: data must be SORTED by the key before groupby works!
data = [
    {"dept": "eng", "name": "Alice"},
    {"dept": "eng", "name": "Bob"},
    {"dept": "hr",  "name": "Carol"},
]
data.sort(key=lambda x: x["dept"])   # sort first!
for dept, members in groupby(data, key=lambda x: x["dept"]):
    names = [m["name"] for m in members]
    print(f"{dept}: {names}")

# ─── accumulate β€” running total/product ─────────────────────────
list(accumulate([1, 2, 3, 4, 5]))            # [1, 3, 6, 10, 15] running sum
list(accumulate([1, 2, 3, 4], lambda a, b: a * b))  # [1, 2, 6, 24] running product

# ─── FUNCTOOLS ───────────────────────────────────────────────────
from functools import reduce, partial, wraps, cache, lru_cache

# reduce β€” apply a function cumulatively to reduce a sequence to one value
from functools import reduce
total = reduce(lambda acc, x: acc + x, [1, 2, 3, 4, 5])  # 15
# Same as: ((((1+2)+3)+4)+5) = 15

# ─── USEFUL BUILT-IN FUNCTIONS ───────────────────────────────────
any([False, True, False])    # True β€” at least one is True
all([True, True, True])      # True β€” ALL are True
any([])                      # False β€” empty is falsy
all([])                      # True β€” vacuously true (no items to fail)

max([3, 1, 4], key=abs)      # 4 β€” max by absolute value
min("hello", key=ord)        # 'e' β€” min by ASCII code (e=101, h=104, l=108, o=111)

# zip with strict=True β€” raises error if lengths differ (Python 3.10+)
list(zip([1, 2, 3], [4, 5, 6]))            # [(1,4),(2,5),(3,6)]

# enumerate β€” add index to any iterable
list(enumerate(["a", "b", "c"], start=1))  # [(1,'a'),(2,'b'),(3,'c')]

# vars() and dir() β€” introspection
vars(obj)   # returns obj.__dict__ β€” the object's attributes as a dict
dir(obj)    # returns a list of all attributes and methods of obj

πŸ”—LangChain β€” Core Concepts & LCEL

What is a Large Language Model (LLM)?

Before diving into LangChain, you need to understand what an LLM is. Think of an LLM like a super-powered autocomplete. You've seen autocomplete on your phone β€” you type "I want to" and it suggests "eat pizza". An LLM is that idea taken to an extreme β€” trained on billions of text documents, it can complete, continue, summarize, translate, answer questions, and generate any kind of text.

GPT-4, Claude, and Gemini are examples of LLMs. You send them text (called a "prompt"), and they generate a response.

What is LangChain?

LangChain is a Python library that makes it easier to build applications powered by LLMs. Instead of writing raw API calls to OpenAI or Anthropic, LangChain gives you building blocks β€” prompts, models, output parsers, retrievers β€” that you can chain together into pipelines.

LangChain is like an assembly line for AI. Raw material (your question) goes into Station 1 (format into a proper prompt), moves to Station 2 (AI thinks about it), then Station 3 (parse the AI's response into a usable format), and out comes the finished product (a structured answer). Each station is reusable and replaceable.

LCEL (LangChain Expression Language) β€” chains are built using the pipe operator |. It works like Unix pipes. prompt | llm | parser means: send data through prompt first, then llm, then parser. Every component implements the Runnable interface with invoke(), stream(), and batch() methods.

Installation and Setup

Before using LangChain, you need to install it and set up your API keys. Store API keys in a .env file β€” never hardcode them in your code.

bash β€” install
pip install langchain langchain-openai langchain-community langchain-core
pip install langchain-anthropic   # for Claude models
pip install python-dotenv         # for loading .env file
python β€” initializing LLMs
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, SystemMessage
from dotenv import load_dotenv
import os

# load_dotenv() reads the .env file and puts variables into the environment
# Your .env file should contain: OPENAI_API_KEY=sk-...
load_dotenv()

# ─── OpenAI GPT ───────────────────────────────────────────────────
llm = ChatOpenAI(
    model="gpt-4o-mini",     # which model to use
    temperature=0.7,          # 0 = deterministic, 1+ = more creative/random
    max_tokens=1024,          # max response length
)

# ─── Anthropic Claude ─────────────────────────────────────────────
claude = ChatAnthropic(
    model="claude-opus-4-5",
    temperature=0,            # 0 = very consistent, deterministic
)

# ─── DIRECT INVOCATION ───────────────────────────────────────────
# Messages are objects, not plain strings
response = llm.invoke([
    SystemMessage(content="You are a helpful Python tutor."),
    HumanMessage(content="What is a list comprehension?"),
])
print(response.content)      # the AI's text response

# ─── STREAMING β€” get tokens as they're generated ─────────────────
# Instead of waiting for the full response, you get words one by one
for chunk in llm.stream("Explain async/await in simple terms"):
    print(chunk.content, end="", flush=True)   # print without newline, flush immediately

Chains β€” Piping Components Together

The power of LangChain comes from chaining components. The pipe operator | connects them: output of left becomes input of right. A typical chain is: prompt | llm | parser.

python β€” LCEL chains
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableLambda, RunnableParallel

llm    = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()   # converts AI response object β†’ plain string

# ─── BASIC CHAIN ──────────────────────────────────────────────────
# ChatPromptTemplate defines the structure with {placeholders}
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are an expert in {domain}. Give concise answers."),
    ("human",  "{question}"),
])

# The pipe | chains them together
# Data flows: input_dict β†’ prompt β†’ llm β†’ parser β†’ output_string
chain = prompt | llm | parser

result = chain.invoke({
    "domain":   "Python programming",
    "question": "What are list comprehensions?"
})
print(result)   # a plain string with the answer

# ─── BATCH PROCESSING ────────────────────────────────────────────
# Process multiple inputs in parallel
results = chain.batch([
    {"domain": "Python", "question": "What is a decorator?"},
    {"domain": "Python", "question": "What is a generator?"},
    {"domain": "SQL",    "question": "What is a JOIN?"},
])
# Returns a list of 3 results, processed concurrently

# ─── STREAMING ───────────────────────────────────────────────────
for chunk in chain.stream({"domain": "Python", "question": "Explain async"}):
    print(chunk, end="", flush=True)   # tokens arrive one at a time

# ─── PARALLEL EXECUTION ──────────────────────────────────────────
# Run two different chains at the same time on the same input
parallel_chain = RunnableParallel(
    summary  = ChatPromptTemplate.from_template("Summarize in one sentence: {text}") | llm | parser,
    keywords = ChatPromptTemplate.from_template("List 5 keywords from: {text}") | llm | parser,
)
output = parallel_chain.invoke({"text": "Python is a versatile programming language..."})
print(output["summary"])    # the summary
print(output["keywords"])   # the keywords

# ─── RUNNABLELAMBDA β€” wrap any function ─────────────────────────
def count_words(text: str) -> int:
    return len(text.split())

word_counter = RunnableLambda(count_words)
count_chain  = parser | word_counter   # first get string, then count words

# ─── BRANCHING β€” different paths based on content ────────────────
from langchain_core.runnables import RunnableBranch

classify_chain = (
    ChatPromptTemplate.from_template("Is this question about Python? Answer yes/no: {q}")
    | llm
    | parser
)

python_chain  = ChatPromptTemplate.from_template("Python expert: {q}") | llm | parser
general_chain = ChatPromptTemplate.from_template("General assistant: {q}") | llm | parser

branch = RunnableBranch(
    (lambda x: "python" in x["topic"].lower(), python_chain),  # condition, chain
    general_chain,  # default if no condition matches
)

Prompt Templates β€” Reusable Prompt Structures

Prompt templates define the structure of your prompts with placeholders. You fill in the placeholders at call time. This makes prompts reusable and easy to test.

python β€” prompt templates
from langchain_core.prompts import (
    ChatPromptTemplate,
    PromptTemplate,
    MessagesPlaceholder,
    FewShotPromptTemplate,
)
from langchain_core.messages import HumanMessage, AIMessage

# ─── BASIC CHAT PROMPT TEMPLATE ──────────────────────────────────
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful {role}. Be {style}."),
    ("human",  "{question}"),
])
# Fill in placeholders with a dict
filled = prompt.invoke({
    "role":     "Python tutor",
    "style":    "friendly and concise",
    "question": "What is a variable?",
})

# ─── MULTI-TURN CHAT WITH HISTORY ────────────────────────────────
# MessagesPlaceholder allows injecting a list of messages (conversation history)
chat_prompt = ChatPromptTemplate.from_messages([
    ("system",  "You are a helpful assistant."),
    MessagesPlaceholder(variable_name="history"),   # inject past messages here
    ("human",   "{input}"),
])
# When invoked, 'history' should be a list of HumanMessage/AIMessage objects
filled = chat_prompt.invoke({
    "history": [
        HumanMessage(content="Hi, my name is Alice."),
        AIMessage(content="Hello Alice! How can I help you?"),
    ],
    "input": "What is my name?"
})
# The AI will see the full conversation and answer "Your name is Alice."

# ─── FEW-SHOT TEMPLATE β€” teach by example ────────────────────────
# Few-shot prompting shows the AI examples of what you want
examples = [
    {"input": "happy",  "output": "sad"},
    {"input": "tall",   "output": "short"},
    {"input": "fast",   "output": "slow"},
]
example_template = PromptTemplate.from_template("Input: {input}\nOutput: {output}")

few_shot = FewShotPromptTemplate(
    examples=examples,
    example_prompt=example_template,
    prefix="Give the antonym (opposite) of each input word.",
    suffix="Input: {word}\nOutput:",
    input_variables=["word"],
)
print(few_shot.format(word="hot"))
# Output: ... examples ... Input: hot / Output:

# ─── PARTIAL PROMPTS β€” pre-fill some variables ───────────────────
base_prompt = ChatPromptTemplate.from_messages([
    ("system", "You are an expert in {domain}."),
    ("human",  "{question}"),
])
python_prompt = base_prompt.partial(domain="Python")  # domain is now fixed
result = python_prompt.invoke({"question": "What is a decorator?"})
# Only need to provide 'question' now

Output Parsers β€” Structuring AI Responses

By default, an LLM returns a string. Output parsers convert that string into a more useful format β€” a Python dict, a list, or a Pydantic object.

python β€” output parsers
from langchain_core.output_parsers import (
    StrOutputParser,          # AI response β†’ plain string
    JsonOutputParser,         # AI response β†’ Python dict
    PydanticOutputParser,     # AI response β†’ Pydantic object (validated!)
    CommaSeparatedListOutputParser,  # AI response β†’ Python list
)
from langchain_core.prompts import PromptTemplate
from pydantic import BaseModel, Field

llm = ChatOpenAI(model="gpt-4o-mini")

# ─── STRING PARSER β€” the simplest ────────────────────────────────
chain = ChatPromptTemplate.from_template("Answer: {q}") | llm | StrOutputParser()
result = chain.invoke({"q": "What color is the sky?"})
print(type(result))   # <class 'str'>

# ─── JSON PARSER ─────────────────────────────────────────────────
json_chain = (
    PromptTemplate.from_template(
        "Return a JSON object with 'name' and 'score' for this text: {text}"
    )
    | llm
    | JsonOutputParser()
)
result = json_chain.invoke({"text": "Alice got 95/100"})
print(result["name"])    # "Alice"
print(result["score"])   # 95

# ─── PYDANTIC PARSER β€” structured AND validated ───────────────────
class MovieReview(BaseModel):
    title:   str   = Field(description="The movie title")
    rating:  float = Field(description="Rating from 1.0 to 10.0")
    summary: str   = Field(description="One-sentence summary")
    liked:   bool  = Field(description="Whether the reviewer liked it")

parser = PydanticOutputParser(pydantic_object=MovieReview)

prompt = PromptTemplate(
    template="Review this movie and provide your response.\n\n{format_instructions}\n\nMovie: {movie}",
    input_variables=["movie"],
    partial_variables={
        "format_instructions": parser.get_format_instructions()
        # This inserts instructions telling the AI how to format its response
    },
)

chain = prompt | llm | parser
review: MovieReview = chain.invoke({"movie": "Inception (2010)"})
print(review.title)    # "Inception"
print(review.rating)   # 9.2 (or whatever the AI says)
print(review.liked)    # True
# It's now a proper Pydantic object, not just a string!

πŸ“šLangChain β€” RAG & Vector Stores

What is RAG?

LLMs are trained on data up to a certain date and don't know anything about YOUR private documents β€” your company's manuals, your codebase, your database. RAG (Retrieval-Augmented Generation) solves this by giving the AI a "cheat sheet" at question time.

Before asking the AI a question, RAG searches through YOUR documents for relevant pages, then pastes those pages into the question. The AI reads those pages and answers based on them. The AI doesn't need to "know" the answer β€” it just reads and synthesizes.

RAG is like an open-book exam. The LLM is the student. Without RAG, it's a closed-book exam β€” the student must answer from memory alone. With RAG, before answering, the student can quickly search their notes and textbooks for relevant information, then answer using those sources. Much more accurate!

πŸ“„
1. Load
Load documents (PDF, web, etc.)
β†’
βœ‚οΈ
2. Split
Break into small chunks
β†’
πŸ”’
3. Embed
Convert text to vectors
β†’
πŸ—„οΈ
4. Store
Save in vector database
β†’
πŸ€–
5. Retrieve & Generate
Find relevant chunks, ask LLM

What are Embeddings and Vector Stores?

An embedding converts text into a list of numbers (a vector) that captures the meaning. "king" and "queen" would have similar vectors because they have similar meanings. "cat" and "automobile" would have very different vectors.

A vector store (like FAISS or Chroma) stores all your document chunks as vectors and can instantly find the chunks closest in meaning to your query. This is the "search" step of RAG.

python β€” complete RAG pipeline
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_community.document_loaders import PyPDFLoader, WebBaseLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

# ─── STEP 1: LOAD DOCUMENTS ──────────────────────────────────────
# PyPDFLoader reads a PDF file and returns a list of Document objects
loader = PyPDFLoader("company_manual.pdf")
documents = loader.load()
# Each document has .page_content (the text) and .metadata (page number, etc.)

# ─── STEP 2: SPLIT INTO CHUNKS ───────────────────────────────────
# LLMs have a context window limit β€” you can't feed a 500-page PDF all at once
# So we split it into overlapping chunks
splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,       # each chunk is ~1000 characters
    chunk_overlap=200,     # chunks overlap by 200 chars (so context isn't lost at boundaries)
    add_start_index=True,  # track where each chunk came from in the original doc
)
chunks = splitter.split_documents(documents)
print(f"Split {len(documents)} pages into {len(chunks)} chunks")

# ─── STEP 3: CREATE EMBEDDINGS AND VECTOR STORE ──────────────────
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
# FAISS.from_documents() converts each chunk to a vector and stores them
vectorstore = FAISS.from_documents(chunks, embeddings)
vectorstore.save_local("faiss_index")   # save to disk for reuse

# ─── STEP 4: BUILD RETRIEVER ─────────────────────────────────────
# The retriever searches the vectorstore and returns relevant chunks
retriever = vectorstore.as_retriever(
    search_type="mmr",           # MMR = Maximum Marginal Relevance
    # MMR returns diverse results β€” avoids returning 5 nearly-identical chunks
    search_kwargs={"k": 5, "fetch_k": 20},
    # fetch_k=20: consider 20 candidates, return the 5 most relevant AND diverse
)

# ─── STEP 5: BUILD THE RAG CHAIN ─────────────────────────────────
rag_prompt = ChatPromptTemplate.from_messages([
    ("system", """You are a helpful assistant. Answer the question ONLY using the provided context.
If the answer is not in the context, say "I don't know β€” this is not covered in the documents."

Context:
{context}"""),
    ("human", "{question}"),
])

def format_docs(docs) -> str:
    """Join all retrieved document chunks into one context string."""
    return "\n\n".join(doc.page_content for doc in docs)

llm = ChatOpenAI(model="gpt-4o-mini")

rag_chain = (
    {
        # 'context': retrieve relevant chunks β†’ format them as a string
        "context": retriever | format_docs,
        # 'question': pass the question through unchanged
        "question": RunnablePassthrough()
    }
    | rag_prompt    # format into a chat prompt
    | llm           # send to the AI
    | StrOutputParser()  # extract the text response
)

# ─── USAGE ───────────────────────────────────────────────────────
answer = rag_chain.invoke("What is the warranty period for the product?")
print(answer)

# ─── LOADING FROM SAVED INDEX ────────────────────────────────────
# Next time, load from disk instead of rebuilding (expensive!)
saved_vs = FAISS.load_local("faiss_index", embeddings, allow_dangerous_deserialization=True)

πŸ€–LangChain β€” Agents & Tools

What are Agents?

A basic chain always follows the same steps in the same order. An agent is different β€” it lets the LLM decide what to do next. You give the LLM a set of tools (functions it can call), and the LLM chooses which tools to call and in what order to answer your question.

For example, if you ask "What's the weather in Paris and what's 125 Γ— 37?", the agent might call get_weather("Paris"), then calculate("125 * 37"), then synthesize the results into an answer.

An agent is like a smart assistant given a set of tools. You ask a question. The assistant thinks about what tools to use (a weather API? a calculator? a web search?), uses them in whatever order makes sense, and reports back. The assistant drives the process β€” you don't specify the steps.

python β€” tools and agent executor
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# ─── STEP 1: DEFINE TOOLS ────────────────────────────────────────
# The @tool decorator turns a function into an LLM-callable tool
# The docstring becomes the tool's description (the LLM reads it!)
@tool
def get_weather(city: str) -> str:
    """Get the current weather for a given city name."""
    # In a real app, this would call a weather API like OpenWeatherMap
    weather_data = {
        "Paris": "22Β°C, partly cloudy",
        "London": "15Β°C, rainy",
        "Tokyo": "28Β°C, sunny",
    }
    return weather_data.get(city, f"Weather data for {city} is unavailable.")

@tool
def search_web(query: str) -> str:
    """Search the web for current information about a topic."""
    # In a real app, this would call Google/Bing/Tavily API
    return f"Search results for '{query}': [Article 1 about {query}...]"

@tool
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression safely. Example: '125 * 37'"""
    try:
        # eval() with empty builtins β€” prevents code injection attacks
        result = eval(expression, {"__builtins__": {}}, {})
        return f"{expression} = {result}"
    except Exception as e:
        return f"Calculation error: {e}"

tools = [get_weather, search_web, calculate]

# ─── STEP 2: BIND TOOLS TO LLM ───────────────────────────────────
llm = ChatOpenAI(model="gpt-4o", temperature=0)  # temperature=0 for consistent tool use

# ─── STEP 3: CREATE THE AGENT ────────────────────────────────────
# The prompt must include an 'agent_scratchpad' for the LLM's reasoning steps
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant with access to weather, search, and calculator tools. Use them as needed."),
    ("human",  "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),  # required for tool use
])

agent = create_tool_calling_agent(llm, tools, prompt)

# AgentExecutor actually runs the agent loop:
# LLM β†’ tool call β†’ tool result β†’ LLM again β†’ ... β†’ final answer
executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,       # prints each step (great for debugging)
    max_iterations=5,   # safety limit β€” stop after 5 steps even if not done
    handle_parsing_errors=True,  # don't crash on format errors
)

# ─── STEP 4: USE THE AGENT ────────────────────────────────────────
result = executor.invoke({
    "input": "What is the weather in Paris? Also, what is 125 * 37?"
})
print(result["output"])
# Agent will:
# 1. Call get_weather("Paris")
# 2. Call calculate("125 * 37")
# 3. Combine results into a natural language answer

🧠LangChain β€” Memory & State

Why Does Memory Matter?

By default, every time you call an LLM it has no memory of previous conversations. If you say "My name is Alice" and then ask "What's my name?", it will say "I don't know" β€” because it forgot the first message.

To build a chatbot that remembers context, you need to maintain conversation history and inject it into every prompt.

python β€” conversation memory with session management
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

llm    = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()

# The prompt includes a MessagesPlaceholder where history will be injected
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant. Remember the conversation history."),
    MessagesPlaceholder(variable_name="history"),  # ← history goes here
    ("human",  "{input}"),
])

chain = prompt | llm | parser

# ─── SESSION STORE β€” one history per conversation ─────────────────
# In a real app, this would be Redis or a database
# Here we use a dict (in-memory, disappears on restart)
session_store: dict[str, ChatMessageHistory] = {}

def get_session_history(session_id: str) -> ChatMessageHistory:
    """Get or create a message history for a session."""
    if session_id not in session_store:
        session_store[session_id] = ChatMessageHistory()   # empty history
    return session_store[session_id]

# Wrap the chain with history management
chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,          # function to get/create session history
    input_messages_key="input",   # which key in the input dict is the user message
    history_messages_key="history", # which key to use for injecting history
)

# ─── MULTI-TURN CONVERSATION ─────────────────────────────────────
# Every call with the same session_id shares history
config = {"configurable": {"session_id": "user_alice"}}

r1 = chain_with_history.invoke({"input": "Hi! My name is Alice."}, config=config)
print(r1)  # "Hello Alice! Nice to meet you..."

r2 = chain_with_history.invoke({"input": "What is my name?"}, config=config)
print(r2)  # "Your name is Alice." β€” it remembers!

# Second user has their own separate history
config2 = {"configurable": {"session_id": "user_bob"}}
r3 = chain_with_history.invoke({"input": "What is my name?"}, config=config2)
print(r3)  # "I don't know your name yet..." β€” Bob's session is separate

πŸ•ΈοΈLangGraph β€” State Graphs & Nodes

What is LangGraph?

LangGraph lets you build AI workflows as directed graphs. Think of a flowchart where each box (node) does some work, and arrows (edges) connect them. Unlike a simple chain (which is linear), a graph can loop β€” a node can send control back to a previous node, enabling complex multi-step AI agents.

What is a State Machine?

Before explaining LangGraph's specifics, let's understand state machines. A state machine is a system that can be in one of several states, and it transitions between states based on events or conditions.

A traffic light is a state machine. It can be in state RED, YELLOW, or GREEN. It transitions from GREEN β†’ YELLOW β†’ RED β†’ GREEN. Each state has a behaviour (stop, slow down, go). LangGraph is a state machine for AI workflows β€” nodes are the states, edges are the transitions.

Think of LangGraph like a flowchart that can loop. Traditional chains are a straight line: Step 1 β†’ Step 2 β†’ Step 3 β†’ Done. LangGraph lets you go backwards: Step 3 β†’ "Is the answer good enough?" β†’ No β†’ Step 2 again β†’ Step 3 again β†’ Yes β†’ Done. This enables "keep trying until you get a good result" patterns.

Key Concepts

  • State β€” a shared "notebook" that all nodes can read from and write to. Defined as a TypedDict. It persists throughout the entire graph execution.
  • Nodes β€” Python functions that receive the current state, do some work (call an LLM, search the web, process data), and return updates to the state.
  • Edges β€” connections between nodes. Can be static (always go from A to B) or conditional (go to B or C depending on the state).
  • Reducers β€” functions that merge a node's returned update into the existing state. The default is "replace"; add_messages appends to a list.
bash β€” install
pip install langgraph langchain-openai
python β€” minimal chatbot with LangGraph
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages   # a reducer function
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

# ─── STEP 1: DEFINE THE STATE ────────────────────────────────────
# The state is a TypedDict β€” think of it as the "shared notebook"
# All nodes read from it and write updates to it
class State(TypedDict):
    # Annotated[list, add_messages] means:
    # - 'messages' is a list
    # - When a node returns new messages, add_messages APPENDS them (doesn't replace)
    messages: Annotated[list, add_messages]

# ─── STEP 2: CREATE THE GRAPH ────────────────────────────────────
# StateGraph(State) creates a graph that uses our State TypedDict
graph_builder = StateGraph(State)

llm = ChatOpenAI(model="gpt-4o-mini")

# ─── STEP 3: DEFINE NODES ────────────────────────────────────────
# Each node is a function: (state) -> partial_state_update
def chatbot_node(state: State) -> dict:
    """Send all messages to the LLM and get a response."""
    # state["messages"] is the conversation history
    response = llm.invoke(state["messages"])
    # Return only the CHANGED fields β€” the rest of the state is unchanged
    return {"messages": [response]}   # add_messages reducer appends this

# ─── STEP 4: ADD NODES TO THE GRAPH ──────────────────────────────
graph_builder.add_node("chatbot", chatbot_node)

# ─── STEP 5: ADD EDGES ───────────────────────────────────────────
# START and END are special built-in nodes
graph_builder.add_edge(START, "chatbot")   # graph starts at 'chatbot'
graph_builder.add_edge("chatbot", END)     # 'chatbot' always goes to END

# ─── STEP 6: COMPILE THE GRAPH ───────────────────────────────────
# compile() validates the graph and returns an executable app
app = graph_builder.compile()

# ─── STEP 7: RUN THE GRAPH ────────────────────────────────────────
result = app.invoke({
    "messages": [HumanMessage(content="What is Python used for?")]
})
# result is the final State dict
print(result["messages"][-1].content)   # the AI's last message

Designing State and Conditional Edges

Real workflows need more complex state and branching. Conditional edges look at the current state and decide which node to go to next β€” this is how you build loops and dynamic routing.

python β€” state reducers and conditional routing
from typing import TypedDict, Annotated
from dataclasses import dataclass, field
from langgraph.graph import StateGraph, START, END

# ─── EXAMPLE: RESEARCH AGENT WITH LOOP ───────────────────────────
@dataclass
class ResearchState:
    query:        str        = ""          # the user's question
    web_results:  list[str]  = field(default_factory=list)  # search results
    summary:      str        = ""          # final summary
    iterations:   int        = 0          # how many times we've searched
    is_complete:  bool       = False      # are we done?
    quality_ok:   bool       = False      # is the result good enough?

# ─── NODE FUNCTIONS ───────────────────────────────────────────────
def search_web(state: ResearchState) -> dict:
    """Search the web for information (simulated)."""
    print(f"Searching for: {state.query} (attempt {state.iterations + 1})")
    results = [
        f"Result {i}: Information about {state.query} β€” detail {i}"
        for i in range(1, 4)
    ]
    return {
        "web_results": results,
        "iterations": state.iterations + 1,
    }

def evaluate_results(state: ResearchState) -> dict:
    """Check if we have enough good results."""
    has_enough = len(state.web_results) >= 3
    return {"quality_ok": has_enough}

def summarize(state: ResearchState) -> dict:
    """Summarize the results into a final answer."""
    combined = " ".join(state.web_results)
    summary = f"Summary for '{state.query}': Based on {len(state.web_results)} sources: {combined[:100]}..."
    return {"summary": summary, "is_complete": True}

# ─── ROUTING FUNCTION ─────────────────────────────────────────────
# Returns the NAME of the next node to go to
def should_continue(state: ResearchState) -> str:
    """Decide whether to keep searching or to summarize."""
    if state.is_complete:
        return "done"              # we're done!
    if state.iterations >= 3:
        return "summarize"         # too many attempts, just summarize
    if not state.quality_ok:
        return "search"            # not enough results, search again
    return "summarize"             # results are good, summarize

# ─── BUILD THE GRAPH ──────────────────────────────────────────────
g = StateGraph(ResearchState)

g.add_node("search",   search_web)
g.add_node("evaluate", evaluate_results)
g.add_node("summarize", summarize)

g.add_edge(START, "search")          # always start with search
g.add_edge("search", "evaluate")     # after search, evaluate

# Conditional edge: based on should_continue(), go to "search", "summarize", or END
g.add_conditional_edges(
    "evaluate",          # from this node
    should_continue,     # call this function to decide
    {                    # map return values to node names
        "search":    "search",      # loop back to search
        "summarize": "summarize",   # proceed to summarize
        "done":      END,           # stop
    }
)
g.add_edge("summarize", END)

app = g.compile()
result = app.invoke(ResearchState(query="Python async programming"))
print(result.summary)

Checkpointing β€” Persistent Memory Across Conversations

Checkpointing saves the graph's state after each step, keyed by a thread_id. This lets you pause and resume conversations, build multi-turn chatbots, and implement human-in-the-loop approval flows.

python β€” checkpointing
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from typing import TypedDict, Annotated

class State(TypedDict):
    messages: Annotated[list, add_messages]

llm = ChatOpenAI(model="gpt-4o-mini")

def chatbot(state):
    return {"messages": [llm.invoke(state["messages"])]}

graph = StateGraph(State)
graph.add_node("chatbot", chatbot)
graph.add_edge(START, "chatbot")
graph.add_edge("chatbot", END)

# ─── ADD CHECKPOINTER ─────────────────────────────────────────────
memory = MemorySaver()             # in-memory checkpointer (use SqliteSaver for persistence)
app = graph.compile(checkpointer=memory)

# ─── MULTI-TURN CONVERSATION ─────────────────────────────────────
# thread_id identifies this conversation β€” all messages are stored per thread
config = {"configurable": {"thread_id": "conversation-with-alice"}}

# Turn 1
r1 = app.invoke({"messages": [HumanMessage("Hi, my name is Alice and I love Python.")]}, config=config)
print(r1["messages"][-1].content)   # "Hello Alice! Python is a great language..."

# Turn 2 β€” previous messages are automatically restored from the checkpoint!
r2 = app.invoke({"messages": [HumanMessage("What programming language did I mention?")]}, config=config)
print(r2["messages"][-1].content)   # "You mentioned Python." β€” it remembers!

# ─── HUMAN-IN-THE-LOOP ────────────────────────────────────────────
# interrupt_before pauses execution BEFORE the specified node
# Useful for letting a human review/approve before a dangerous action
app_with_review = graph.compile(
    checkpointer=memory,
    interrupt_before=["chatbot"],   # pause before 'chatbot' runs
)

# Start execution β€” stops at the interrupt point
result = app_with_review.invoke({"messages": [HumanMessage("Delete all data")]}, config=config)
# ... at this point, a human can review the state ...

# To APPROVE and continue:
result = app_with_review.invoke(None, config=config)  # None = no new input, just continue

# ─── PERSISTENT CHECKPOINTING ────────────────────────────────────
# For production, use SqliteSaver to persist across restarts
# from langgraph.checkpoint.sqlite import SqliteSaver
# with SqliteSaver.from_conn_string("checkpoints.db") as checkpointer:
#     app = graph.compile(checkpointer=checkpointer)

πŸ”€LangGraph β€” Multi-Agent Systems

Multi-Agent Systems β€” Specialists Working Together

Complex tasks benefit from specialized agents. Instead of one generalist agent that does everything, you have multiple specialist agents (a researcher, a coder, a writer) coordinated by a supervisor agent. The supervisor decides which specialist to call based on the current task.

Think of a hospital. When you come in, the receptionist (supervisor) assesses your problem and routes you to the right specialist β€” cardiologist, neurologist, orthopedist. Each specialist is an expert in their domain. The supervisor doesn't do the medical work; they coordinate. Multi-agent systems work the same way.

python β€” supervisor multi-agent pattern
from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent

llm = ChatOpenAI(model="gpt-4o", temperature=0)

# ─── SPECIALIST TOOLS ────────────────────────────────────────────
@tool
def analyze_code(code: str) -> str:
    """Analyze Python code for bugs, performance issues, and best practices."""
    # In production, this would use a real code analysis tool
    return f"Code analysis: {len(code)} chars. Looks mostly OK. Consider type hints."

@tool
def search_docs(query: str) -> str:
    """Search Python documentation for a specific topic."""
    return f"Documentation results for '{query}': [relevant docs found]"

@tool
def write_test(function_name: str) -> str:
    """Generate a unit test for a Python function."""
    return f"def test_{function_name}():\n    # Generated test for {function_name}\n    pass"

# ─── CREATE SPECIALIST AGENTS ────────────────────────────────────
# create_react_agent creates a ReAct (Reason + Act) agent from an LLM + tools
code_reviewer = create_react_agent(llm, [analyze_code])
doc_searcher  = create_react_agent(llm, [search_docs])
test_writer   = create_react_agent(llm, [write_test])

# ─── SUPERVISOR STATE ────────────────────────────────────────────
class SupervisorState(TypedDict):
    messages: Annotated[list, add_messages]
    next_agent: str     # which agent to call next
    task_complete: bool

# ─── SUPERVISOR LOGIC ────────────────────────────────────────────
def supervisor(state: SupervisorState) -> dict:
    """Read the last message and decide which agent should handle it."""
    last_message = state["messages"][-1].content.lower()

    if "review" in last_message or "bug" in last_message or "code" in last_message:
        return {"next_agent": "code_reviewer"}
    elif "documentation" in last_message or "how to" in last_message:
        return {"next_agent": "doc_searcher"}
    elif "test" in last_message or "unit test" in last_message:
        return {"next_agent": "test_writer"}
    else:
        return {"next_agent": "done"}   # no matching specialist β€” finish

# ─── ROUTING FUNCTION ────────────────────────────────────────────
def route(state: SupervisorState) -> Literal["code_reviewer", "doc_searcher", "test_writer", "__end__"]:
    """Map next_agent value to actual node names."""
    if state["next_agent"] == "done":
        return "__end__"
    return state["next_agent"]

# ─── AGENT WRAPPER FUNCTIONS ─────────────────────────────────────
def run_code_reviewer(state: SupervisorState) -> dict:
    result = code_reviewer.invoke({"messages": state["messages"]})
    return {"messages": result["messages"]}

def run_doc_searcher(state: SupervisorState) -> dict:
    result = doc_searcher.invoke({"messages": state["messages"]})
    return {"messages": result["messages"]}

def run_test_writer(state: SupervisorState) -> dict:
    result = test_writer.invoke({"messages": state["messages"]})
    return {"messages": result["messages"]}

# ─── BUILD THE SUPERVISOR GRAPH ──────────────────────────────────
g = StateGraph(SupervisorState)
g.add_node("supervisor",     supervisor)
g.add_node("code_reviewer",  run_code_reviewer)
g.add_node("doc_searcher",   run_doc_searcher)
g.add_node("test_writer",    run_test_writer)

g.add_edge(START, "supervisor")
g.add_conditional_edges("supervisor", route)  # route based on next_agent
g.add_edge("code_reviewer", END)
g.add_edge("doc_searcher",  END)
g.add_edge("test_writer",   END)

app = g.compile()
result = app.invoke({
    "messages": [HumanMessage("Review this code: def add(a,b): return a+b")],
    "next_agent": "",
    "task_complete": False,
})
print(result["messages"][-1].content)
python β€” ReAct agent with prebuilt helper
from langgraph.prebuilt import create_react_agent
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

llm = ChatOpenAI(model="gpt-4o", temperature=0)

@tool
def get_stock_price(ticker: str) -> str:
    """Get the current stock price for a ticker symbol like AAPL or GOOG."""
    mock_prices = {"AAPL": "$189.50", "GOOG": "$142.30", "MSFT": "$425.00"}
    return mock_prices.get(ticker.upper(), f"Price for {ticker} not found")

@tool
def calculate_portfolio_value(tickers: list[str], quantities: list[int]) -> str:
    """Calculate total portfolio value. Provide tickers and quantities as parallel lists."""
    return f"Calculated portfolio of {len(tickers)} stocks"

# create_react_agent builds a complete ReAct loop automatically
agent = create_react_agent(
    model=llm,
    tools=[get_stock_price, calculate_portfolio_value],
    state_modifier="You are a helpful financial assistant. Be precise with numbers.",
)

result = agent.invoke({
    "messages": [HumanMessage("What is the current price of Apple stock?")]
})
print(result["messages"][-1].content)

⚑FastAPI β€” Routing & Pydantic

What is an API?

An API (Application Programming Interface) is a way for programs to talk to each other. Think of a restaurant: you (the client) sit at a table and order food. The waiter (the API) takes your order to the kitchen (the server/database) and brings back your food (the response). You never go into the kitchen directly β€” you communicate through the waiter.

A web API works the same way over the internet using HTTP. Your browser or app sends a request to a URL, and the server sends back data (usually JSON).

HTTP Methods β€” The Verbs of the Web

HTTP has different "verbs" that describe what kind of operation you want to perform:

  • GET β€” Reading. "Give me this thing." GET /users/5 β†’ get user with ID 5
  • POST β€” Creating. "Make a new thing." POST /users with body β†’ create a new user
  • PUT β€” Updating (replace all). "Replace this thing entirely." PUT /users/5 β†’ replace user 5
  • PATCH β€” Updating (partial). "Change just this part." PATCH /users/5 β†’ update specific fields
  • DELETE β€” Deleting. "Remove this thing." DELETE /users/5 β†’ delete user 5

Think of a restaurant menu ordering system. GET is "show me the menu". POST is "place an order for a new dish". PUT is "completely redo my order". PATCH is "just change my drink". DELETE is "cancel my order".

What is FastAPI?

FastAPI is a modern Python framework for building web APIs. It's built on Starlette (for async HTTP handling) and Pydantic (for data validation). Key features:

  • Automatic input validation β€” if you say "age must be 0-150", FastAPI rejects invalid requests automatically
  • Automatic documentation β€” visit /docs and you get a live, interactive API explorer for free
  • Very fast β€” one of the fastest Python frameworks, on par with Node.js and Go for I/O-bound work
  • Built for async β€” designed to handle thousands of concurrent requests
FastAPI = ASGI async web framework built on Starlette + Pydantic. Automatic OpenAPI/Swagger docs at /docs, ReDoc at /redoc. Type hints drive validation and documentation automatically.

Building Your First FastAPI App

Let's build a complete CRUD (Create, Read, Update, Delete) API for users. CRUD covers the four fundamental operations any database-backed API needs.

bash β€” install and run
pip install fastapi uvicorn[standard] pydantic
# uvicorn is the ASGI server that actually runs your FastAPI app
uvicorn main:app --reload --port 8000
# main = the Python file name (main.py)
# app  = the FastAPI instance variable name inside that file
# --reload = restart automatically when code changes (for development only!)
python β€” main.py (complete CRUD API)
from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, Field
from typing import Optional
import uuid   # for generating unique IDs

# Create the FastAPI app instance
app = FastAPI(
    title="User API",
    description="A simple CRUD API for managing users",
    version="1.0.0",
)

# ─── IN-MEMORY DATABASE ─────────────────────────────────────────
# In a real app, this would be PostgreSQL/MongoDB
# Here we use a simple dict: user_id β†’ user_data
users_db: dict[str, dict] = {}

# ─── PYDANTIC MODELS (DATA SCHEMAS) ─────────────────────────────
# UserCreate defines what the CLIENT sends when creating a user
class UserCreate(BaseModel):
    name:  str = Field(min_length=1, max_length=100, description="User's full name")
    email: str = Field(description="User's email address")
    age:   int = Field(ge=0, le=150, description="Age between 0 and 150")
    # ge=greater_than_or_equal, le=less_than_or_equal

# UserResponse defines what the SERVER sends back
class UserResponse(BaseModel):
    id:    str   # the server generates this, client doesn't send it
    name:  str
    email: str
    age:   int

# ─── API ROUTES ────────────────────────────────────────────────────
@app.get("/")
def root():
    """Health check endpoint."""
    return {"status": "ok", "service": "User API", "version": "1.0.0"}

# POST /users β€” Create a new user
@app.post(
    "/users",
    response_model=UserResponse,              # FastAPI validates and formats the response
    status_code=status.HTTP_201_CREATED,      # 201 = Created (not 200 OK)
)
def create_user(data: UserCreate):
    # FastAPI automatically parsed the request body into a UserCreate object
    # and validated all the fields before this function even ran!
    user_id = str(uuid.uuid4())    # generate a random unique ID
    user = {"id": user_id, **data.model_dump()}   # combine id with all user data
    users_db[user_id] = user       # save to our "database"
    return user

# GET /users β€” List all users
@app.get("/users", response_model=list[UserResponse])
def list_users():
    return list(users_db.values())   # return all users as a list

# GET /users/{user_id} β€” Get one user by ID
@app.get("/users/{user_id}", response_model=UserResponse)
def get_user(user_id: str):
    # {user_id} in the path becomes a function parameter automatically
    if user_id not in users_db:
        # HTTPException sends an error response with the given status code
        raise HTTPException(
            status_code=404,
            detail=f"User with id '{user_id}' not found"
        )
    return users_db[user_id]

# PUT /users/{user_id} β€” Replace a user entirely
@app.put("/users/{user_id}", response_model=UserResponse)
def update_user(user_id: str, data: UserCreate):
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="User not found")
    users_db[user_id] = {"id": user_id, **data.model_dump()}
    return users_db[user_id]

# DELETE /users/{user_id} β€” Delete a user
@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_user(user_id: str):
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="User not found")
    del users_db[user_id]
    # 204 No Content β€” don't return anything (just confirms deletion)

Once running, visit http://localhost:8000/docs to see an automatically generated interactive API explorer. You can test every endpoint directly from your browser without writing any additional code!

Pydantic Models β€” The Core of FastAPI

Pydantic is like a strict form. You define the expected shape of the data, and Pydantic automatically validates every incoming request. If the data doesn't match, FastAPI returns a 422 error with a clear message explaining exactly what's wrong β€” before your code even runs.

Think of a bank loan application form. The form specifies "Name must be text, Income must be a positive number, Employment date must be in the past". If you try to submit a form with "Income: banana", it gets rejected immediately at the door. That's what Pydantic does for your API β€” rejects bad data before it can cause problems in your database.

python β€” advanced Pydantic models
from fastapi import FastAPI
from pydantic import BaseModel, Field, validator
from typing import Optional
from enum import Enum

app = FastAPI()

# ─── ENUMS β€” restrict to specific values ─────────────────────────
class Priority(str, Enum):
    low    = "low"
    medium = "medium"
    high   = "high"
    # priority can ONLY be one of these three values

class Status(str, Enum):
    pending    = "pending"
    in_progress = "in_progress"
    done       = "done"

# ─── REQUEST MODEL β€” what client sends ───────────────────────────
class TaskCreate(BaseModel):
    title:          str      = Field(min_length=1, max_length=200, example="Fix login bug")
    description:    Optional[str] = Field(None, max_length=2000)
    priority:       Priority = Priority.medium   # default to medium
    assignee_email: Optional[str] = None

    @validator("title")   # this runs before the object is created
    @classmethod
    def title_must_not_be_blank(cls, value):
        stripped = value.strip()
        if not stripped:
            raise ValueError("Title cannot be blank or just spaces")
        return stripped   # return cleaned value

# ─── RESPONSE MODEL β€” what server returns ────────────────────────
class TaskResponse(TaskCreate):   # inherit all fields from TaskCreate
    id:         int
    status:     Status  = Status.pending
    created_at: str

    class Config:
        from_attributes = True   # allows creating from SQLAlchemy ORM objects

# ─── NESTED MODELS ───────────────────────────────────────────────
class Address(BaseModel):
    street:   str
    city:     str
    zip_code: str = Field(pattern=r"^\d{5}$")   # regex: exactly 5 digits

class CompanyCreate(BaseModel):
    name:         str
    headquarters: Address          # nested Pydantic model
    branches:     list[Address] = []   # list of nested models

@app.post("/companies")
def create_company(data: CompanyCreate):
    # FastAPI automatically validates all nested fields too!
    return data.model_dump()

# ─── HIDING SENSITIVE FIELDS ─────────────────────────────────────
class UserWithPassword(BaseModel):
    id:              int
    name:            str
    email:           str
    hashed_password: str   # INTERNAL β€” should NEVER be in API response!

class UserPublic(BaseModel):
    id:    int
    name:  str
    email: str
    # hashed_password is intentionally missing

@app.get("/me", response_model=UserPublic)  # response_model filters the output!
def get_me():
    # Even though we return hashed_password, the response_model removes it
    user = UserWithPassword(id=1, name="Alice", email="a@x.com", hashed_password="$2b$12$secret")
    return user  # hashed_password will NOT appear in the response βœ“

Parameters β€” Path, Query, Headers, Cookies

FastAPI reads parameters from different parts of the HTTP request. The location is determined automatically by where you declare the parameter.

python β€” all parameter types
from fastapi import FastAPI, Query, Path, Header, Cookie
from typing import Optional

app = FastAPI()

# ─── PATH PARAMETERS ─────────────────────────────────────────────
# Declared in the URL pattern AND as function parameters
@app.get("/items/{item_id}")
def get_item(
    item_id: int = Path(
        ge=1,                              # must be >= 1
        le=9999,                           # must be <= 9999
        description="Item's primary key",
    ),
):
    return {"item_id": item_id}
# GET /items/42 β†’ item_id=42
# GET /items/0  β†’ 422 Validation Error (must be >= 1)
# GET /items/abc β†’ 422 Validation Error (must be an integer)

# ─── QUERY PARAMETERS ────────────────────────────────────────────
# Everything NOT in the path pattern becomes a query parameter
@app.get("/search")
def search(
    q:        str           = Query(min_length=2, description="Search term"),
    page:     int           = Query(default=1, ge=1),
    per_page: int           = Query(default=20, ge=1, le=100),
    active:   Optional[bool] = Query(default=None),  # None means "show all"
    tags:     list[str]     = Query(default=[]),      # multi-value: ?tags=a&tags=b
):
    return {"q": q, "page": page, "per_page": per_page, "active": active, "tags": tags}
# GET /search?q=python&page=2&tags=web&tags=api
# β†’ {"q": "python", "page": 2, "per_page": 20, "active": null, "tags": ["web","api"]}

# ─── HEADERS AND COOKIES ──────────────────────────────────────────
@app.get("/secure")
def secure_route(
    authorization:  str           = Header(description="Bearer token"),
    x_request_id:   Optional[str] = Header(default=None),  # custom header
    session_id:     Optional[str] = Cookie(default=None),   # from cookie
):
    return {
        "auth":       authorization,
        "request_id": x_request_id,
        "session":    session_id,
    }
# Note: Header() automatically converts header names
# X-Request-ID becomes x_request_id in Python (underscores, lowercase)

Error Handling β€” Custom Exception Responses

FastAPI has built-in error handling but you can customize the error format to match your API's conventions. Custom exceptions make your error responses consistent and machine-readable.

python β€” custom exceptions and handlers
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
import logging

app = FastAPI()

# ─── CUSTOM EXCEPTION CLASS ──────────────────────────────────────
class AppError(Exception):
    """Base class for all application errors."""
    def __init__(self, code: str, message: str, status_code: int = 400):
        self.code        = code         # machine-readable error code (e.g. "USER_NOT_FOUND")
        self.message     = message      # human-readable message
        self.status_code = status_code  # HTTP status code
        super().__init__(message)

class NotFoundError(AppError):
    def __init__(self, resource: str, resource_id):
        super().__init__(
            code="NOT_FOUND",
            message=f"{resource} with id '{resource_id}' not found",
            status_code=404
        )

class ValidationAppError(AppError):
    def __init__(self, field: str, message: str):
        super().__init__(
            code="VALIDATION_ERROR",
            message=f"Invalid value for '{field}': {message}",
            status_code=422
        )

# ─── REGISTER EXCEPTION HANDLERS ─────────────────────────────────
@app.exception_handler(AppError)
async def app_error_handler(request: Request, exc: AppError):
    """Convert AppError into a consistent JSON response."""
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error":   exc.code,
            "message": exc.message,
        },
    )

@app.exception_handler(404)
async def not_found_handler(request: Request, exc: HTTPException):
    """Custom 404 response that includes the path."""
    return JSONResponse(
        status_code=404,
        content={
            "error":   "NOT_FOUND",
            "message": "The requested resource does not exist",
            "path":    str(request.url.path),
        },
    )

# ─── USING CUSTOM EXCEPTIONS ─────────────────────────────────────
@app.get("/users/{user_id}")
def get_user(user_id: int):
    if user_id not in [1, 2, 3]:  # mock check
        raise NotFoundError("User", user_id)
    return {"id": user_id, "name": f"User {user_id}"}

πŸ”ŒFastAPI β€” Dependency Injection

What is Dependency Injection?

Dependency injection is a pattern where instead of creating dependencies (database connections, authentication objects, settings) inside a function, you declare them as function parameters and let the framework provide them.

In FastAPI, you use Depends() to declare that a function needs something, and FastAPI automatically creates and provides it before calling your function.

Dependency injection is like a hotel breakfast buffet. Instead of each guest going to the kitchen to cook their own eggs (creating the dependency themselves), the hotel sets up everything and you just pick up what you need (the framework provides it). The setup code is written once, and every endpoint that needs it gets it automatically.

python β€” Depends() for DB and auth
from fastapi import FastAPI, Depends, HTTPException, Header
from typing import Annotated
import jwt   # pip install PyJWT

app = FastAPI()

# ─── DATABASE DEPENDENCY ─────────────────────────────────────────
def get_db():
    """Provide a database connection for each request."""
    # In a real app: db = SessionLocal() (SQLAlchemy session)
    db = {"connected": True, "data": {}}   # fake DB for demo
    try:
        yield db          # 'yield' gives the connection to the endpoint
    finally:
        pass              # db.close() β€” always closes, even on errors

# Annotated makes the type hint and the Depends() work together cleanly
DBDep = Annotated[dict, Depends(get_db)]

@app.get("/items")
def list_items(db: DBDep):
    # FastAPI calls get_db() automatically and passes the result as 'db'
    return {"db_connected": db["connected"], "items": []}

# ─── AUTHENTICATION DEPENDENCY ──────────────────────────────────
class CurrentUser:
    def __init__(self, user_id: int, email: str, is_admin: bool = False):
        self.user_id  = user_id
        self.email    = email
        self.is_admin = is_admin

SECRET_KEY = "your-secret-key-change-in-production"

async def get_current_user(
    authorization: str = Header(description="Bearer token")
) -> CurrentUser:
    """Extract and validate the JWT token from the Authorization header."""
    if not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Token must start with 'Bearer '")
    token = authorization[7:]   # remove "Bearer " prefix
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        return CurrentUser(
            user_id  = payload["sub"],
            email    = payload["email"],
            is_admin = payload.get("is_admin", False),
        )
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token has expired")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")

AuthDep = Annotated[CurrentUser, Depends(get_current_user)]

# ─── COMPOSING DEPENDENCIES ─────────────────────────────────────
def require_admin(user: AuthDep) -> CurrentUser:
    """Dependency that also checks for admin rights."""
    if not user.is_admin:
        raise HTTPException(status_code=403, detail="Admin access required")
    return user

@app.get("/profile")
def get_profile(user: AuthDep):
    return {"user_id": user.user_id, "email": user.email}

@app.delete("/admin/reset-all")
def admin_reset(user: Annotated[CurrentUser, Depends(require_admin)]):
    return {"message": f"Admin {user.email} reset everything"}

# ─── CLASS-BASED DEPENDENCY β€” for parametrized dependencies ──────
class Pagination:
    """Reusable pagination dependency."""
    def __init__(self, page: int = 1, per_page: int = 20):
        if page < 1:
            raise HTTPException(400, "page must be >= 1")
        self.offset = (page - 1) * per_page   # how many records to skip
        self.limit  = per_page                # how many to return

PaginationDep = Annotated[Pagination, Depends()]

@app.get("/products")
def list_products(pag: PaginationDep, db: DBDep):
    return {
        "offset":  pag.offset,
        "limit":   pag.limit,
        "products": [],
    }

πŸ›‘οΈFastAPI β€” Middleware & Auth

What is Middleware?

Middleware is code that runs for every single request, before it reaches your endpoint, and for every response, before it's sent back to the client. It's great for logging, CORS headers, request timing, authentication, and compression.

Middleware is like a security checkpoint at an airport. Every passenger (request) must go through the checkpoint before reaching their gate (endpoint). The checkpoint can scan bags (validate headers), stamp passports (add response headers), or redirect suspicious people (authentication failures). The individual gates don't need to know about the checkpoint β€” it runs automatically for everyone.

python β€” middleware
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
import time, logging, uuid

app = FastAPI()

# ─── CORS MIDDLEWARE ──────────────────────────────────────────────
# CORS (Cross-Origin Resource Sharing) controls which domains can call your API
# Without CORS, browsers block requests from other domains (security feature)
app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "http://localhost:3000",       # local React dev server
        "https://app.example.com",     # production frontend
    ],
    allow_credentials=True,            # allow cookies/auth headers
    allow_methods=["*"],               # allow all HTTP methods
    allow_headers=["*"],               # allow all headers
)

# ─── GZIP COMPRESSION ─────────────────────────────────────────────
# Compress responses larger than 1000 bytes β€” reduces bandwidth
app.add_middleware(GZipMiddleware, minimum_size=1000)

# ─── CUSTOM MIDDLEWARE β€” request logging and timing ───────────────
@app.middleware("http")
async def logging_middleware(request: Request, call_next):
    # Generate a unique ID for this request (useful for tracing in logs)
    request_id = str(uuid.uuid4())[:8]
    start = time.perf_counter()

    # This runs BEFORE the endpoint handler
    logging.info(f"[{request_id}] {request.method} {request.url.path} started")

    # Call the next handler (eventually reaches your endpoint)
    response = await call_next(request)

    # This runs AFTER the endpoint handler
    elapsed_ms = (time.perf_counter() - start) * 1000
    response.headers["X-Request-ID"]    = request_id
    response.headers["X-Process-Time"]  = f"{elapsed_ms:.2f}ms"
    logging.info(f"[{request_id}] β†’ {response.status_code} in {elapsed_ms:.1f}ms")

    return response
python β€” JWT authentication with OAuth2
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel
import jwt
from datetime import datetime, timedelta, timezone
from passlib.context import CryptContext   # pip install passlib bcrypt

SECRET_KEY  = "change-this-to-a-random-secret-in-production"
ALGORITHM   = "HS256"
TOKEN_EXPIRE_MINUTES = 30

app = FastAPI()
pwd_context  = CryptContext(schemes=["bcrypt"])   # bcrypt for password hashing
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")

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

# Simulated user database β€” in production, use a real database
USERS_DB = {
    "alice": {
        "password": pwd_context.hash("secret123"),  # NEVER store plain passwords!
        "email":    "alice@example.com",
        "is_admin": False,
    }
}

def create_access_token(data: dict, expires_in: timedelta = None) -> str:
    """Create a signed JWT token with an expiry time."""
    payload = {
        **data,
        "exp": datetime.now(timezone.utc) + (expires_in or timedelta(minutes=15))
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

@app.post("/auth/token", response_model=Token)
async def login(form: OAuth2PasswordRequestForm = Depends()):
    """Login endpoint β€” exchange username+password for a JWT token."""
    user = USERS_DB.get(form.username)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid username or password",
                           headers={"WWW-Authenticate": "Bearer"})
    if not pwd_context.verify(form.password, user["password"]):
        raise HTTPException(status_code=401, detail="Invalid username or password",
                           headers={"WWW-Authenticate": "Bearer"})
    # Credentials are valid β€” create and return a token
    token = create_access_token(
        data={"sub": form.username, "email": user["email"]},
        expires_in=timedelta(minutes=TOKEN_EXPIRE_MINUTES),
    )
    return Token(access_token=token)

async def get_current_user(token: str = Depends(oauth2_scheme)) -> str:
    """Decode and validate a JWT token, returning the username."""
    try:
        payload  = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        if not username:
            raise HTTPException(status_code=401, detail="Invalid token payload")
        return username
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token has expired β€” please log in again")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")

@app.get("/me")
async def get_me(username: str = Depends(get_current_user)):
    """Protected endpoint β€” requires a valid JWT token."""
    user = USERS_DB.get(username)
    return {"username": username, "email": user["email"]}

πŸš€FastAPI β€” Async & Background Tasks

FastAPI is built for async from the ground up. Async endpoints can handle many concurrent requests without blocking. Background tasks run after the response is sent, so the user doesn't wait for slow operations like sending emails.

python β€” async endpoints, lifespan, and background tasks
from fastapi import FastAPI, BackgroundTasks
from contextlib import asynccontextmanager
import asyncio, httpx

# ─── LIFESPAN β€” startup and shutdown events ──────────────────────
# Run code when the server starts and when it stops
# Replaces the old @app.on_event("startup") approach
@asynccontextmanager
async def lifespan(app: FastAPI):
    # ─── STARTUP CODE ────────────────────────────────────────────
    print("Server starting up...")
    # Create shared resources here β€” they'll be available for ALL requests
    app.state.http_client = httpx.AsyncClient()  # shared HTTP client (efficient)
    app.state.db_pool     = None  # db_pool = await create_pool(DB_URL)
    print("Ready to accept requests")

    yield   # ← application runs here (between startup and shutdown)

    # ─── SHUTDOWN CODE ───────────────────────────────────────────
    print("Server shutting down, cleaning up...")
    await app.state.http_client.aclose()   # close the HTTP client
    # await app.state.db_pool.close()      # close database pool

app = FastAPI(lifespan=lifespan)

# ─── ASYNC ENDPOINT β€” non-blocking I/O ───────────────────────────
@app.get("/external-data")
async def fetch_external(url: str):
    """Fetch data from an external API without blocking other requests."""
    # httpx.AsyncClient is the async version of the 'requests' library
    async with httpx.AsyncClient() as client:
        resp = await client.get(url, timeout=10.0)
        resp.raise_for_status()
        return resp.json()

# ─── CONCURRENT REQUESTS ─────────────────────────────────────────
@app.get("/dashboard")
async def dashboard():
    """Fetch multiple data sources at the same time."""
    async with httpx.AsyncClient() as client:
        # Create all three requests as tasks β€” they START at the same time
        users_task    = asyncio.create_task(client.get("https://api.example.com/users"))
        products_task = asyncio.create_task(client.get("https://api.example.com/products"))
        stats_task    = asyncio.create_task(client.get("https://api.example.com/stats"))

        # Wait for ALL three to finish β€” total time β‰ˆ slowest request, not sum of all
        users, products, stats = await asyncio.gather(users_task, products_task, stats_task)

    return {
        "users":    users.json(),
        "products": products.json(),
        "stats":    stats.json(),
    }

# ─── BACKGROUND TASKS β€” fire and forget ──────────────────────────
def send_welcome_email(to_email: str):
    """Slow operation β€” sends an email. Runs in background."""
    import time
    time.sleep(2)   # simulate slow email service
    print(f"Welcome email sent to {to_email}")

def log_registration(user_id: int, email: str):
    """Log the registration event."""
    print(f"User {user_id} registered with {email}")

@app.post("/register")
async def register(email: str, background_tasks: BackgroundTasks):
    """Register a user and return immediately β€” emails are sent in background."""
    user_id = 42  # would come from database

    # These tasks run AFTER the response is sent β€” user doesn't wait!
    background_tasks.add_task(send_welcome_email, email)
    background_tasks.add_task(log_registration, user_id, email)

    return {"message": "Registration successful! Welcome email will arrive shortly.", "user_id": user_id}

# ─── STREAMING RESPONSES (SSE) ────────────────────────────────────
from fastapi.responses import StreamingResponse
import json

async def generate_stream(prompt: str):
    """Stream response tokens one at a time."""
    words = f"Here is a response to: {prompt}".split()
    for word in words:
        yield f"data: {json.dumps({'token': word})}\n\n"  # SSE format
        await asyncio.sleep(0.1)  # simulate token generation delay
    yield "data: [DONE]\n\n"

@app.get("/stream")
async def stream_response(prompt: str):
    """Stream LLM-style token-by-token response."""
    return StreamingResponse(
        generate_stream(prompt),
        media_type="text/event-stream",
        headers={
            "Cache-Control":  "no-cache",
            "Connection":     "keep-alive",
            "X-Accel-Buffering": "no",   # prevents nginx from buffering
        },
    )

🎯Interview: Python Q&A

== tests value equality β€” it calls the __eq__ method and asks "do these two objects contain the same value?" is tests identity β€” it asks "are these two variables pointing to the exact same object in memory?" (equivalent to id(a) == id(b)).

You should almost always use == for comparisons. The main exceptions are: always use is to compare with None, True, and False. Never use is to compare strings or numbers β€” CPython caches small integers (-5 to 256) and interned strings, so is may return True for those but the behaviour is an implementation detail, not guaranteed.

python
a = [1, 2, 3]
b = [1, 2, 3]
print(a == b)   # True β€” same VALUES
print(a is b)   # False β€” DIFFERENT objects in memory

c = a           # c points to the SAME object as a
print(c is a)   # True β€” literally the same object

# Always use 'is' for None checks
def find(key):
    return None

result = find("x")
if result is None:      # βœ“ correct
    print("not found")
if result == None:      # βœ“ works but not Pythonic (and can be overridden by __eq__)
    pass

# CPython integer caching β€” an implementation detail, NOT reliable
x = 256; y = 256; print(x is y)  # True (cached)
x = 257; y = 257; print(x is y)  # False (not cached)

The Global Interpreter Lock (GIL) is a mutex in CPython (the standard Python implementation) that allows only one thread to execute Python bytecode at a time, even on multi-core CPUs. This is a safety measure that simplifies CPython's memory management by preventing race conditions on the reference counter.

The practical effect: CPU-bound tasks (heavy computation, number crunching) do NOT benefit from threading in CPython β€” the GIL prevents true parallelism. However, I/O-bound tasks (network requests, file operations, database queries) DO benefit from threading because the GIL is released while waiting for I/O.

For CPU-bound parallelism, use multiprocessing β€” each process gets its own Python interpreter and its own GIL, achieving true parallelism. For I/O-bound tasks, asyncio is usually the best choice as it has lower overhead than threads. Note: Python 3.13+ has an experimental free-threaded mode that removes the GIL.

python β€” when to use threads vs processes vs asyncio
# I/O-bound β†’ asyncio (best) or threading (also works)
import asyncio, aiohttp
async def fetch_all(urls):
    async with aiohttp.ClientSession() as s:
        return await asyncio.gather(*[s.get(u) for u in urls])

# CPU-bound β†’ multiprocessing (bypasses GIL, true parallelism)
from multiprocessing import Pool
def heavy_computation(n):
    return sum(i**2 for i in range(n))
with Pool(processes=4) as pool:
    results = pool.map(heavy_computation, [10**6] * 4)

# Python 3.13+ experimental no-GIL:
# python -X gil=0 your_script.py

Python uses reference counting as its primary memory management mechanism. Every object has an internal counter tracking how many references point to it. When that counter reaches zero, the memory is freed immediately β€” no waiting for a GC cycle.

However, reference counting fails with circular references β€” where object A references B and B references A. Both stay at refcount 1 forever even if nothing else points to them. Python's cyclic garbage collector runs periodically to detect and collect these cycles. It uses a generational approach: most objects die young (Generation 0), older survivors are promoted to Generations 1 and 2 which are collected less frequently.

python β€” memory management
import sys, gc
a = [1, 2, 3]
print(sys.getrefcount(a))  # 2 (a + getrefcount's own arg)
b = a
print(sys.getrefcount(a))  # 3 (a, b, getrefcount's arg)
del b
print(sys.getrefcount(a))  # 2 again

# Circular reference β€” needs the cyclic GC
x = []
x.append(x)   # x references itself!
del x          # refcount drops to 1 (still has self-reference), not freed
gc.collect()   # cyclic GC collects it

gc.disable()  # disables cyclic GC (risky β€” only for perf-critical code)
gc.collect()  # manual trigger

A generator is a special function that uses yield instead of return. When called, it returns a generator object without executing any code. When you call next() on the generator, it executes until the next yield, returns that value, and suspends β€” preserving all local variables in place. The next next() call resumes execution from where it left off.

Key differences: generators produce values lazily (one at a time on demand), use almost zero memory regardless of how many values they'll produce, can represent infinite sequences, and can only be iterated once. Regular functions compute everything and return all at once.

python
def squares_list(n):     # regular: allocates list in memory
    return [x**2 for x in range(n)]

def squares_gen(n):      # generator: computes one at a time
    for x in range(n):
        yield x**2   # suspends here, resumes on next next()

gen = squares_gen(1_000_000)
next(gen)   # 0  β€” O(1) memory for any n
next(gen)   # 1
# vs: squares_list(1_000_000) allocates ~8 MB right away

# Generator expression
gen_expr = (x**2 for x in range(1_000_000))  # lazy β€” no computation yet
print(sum(gen_expr))  # sums without storing all values!

A shallow copy creates a new outer container but the inner objects are still shared. If the original contains mutable inner objects (lists within lists), modifying the inner objects in the copy will affect the original too. A deep copy recursively copies everything β€” the copy is completely independent at every level of nesting.

python
import copy
original = [[1, 2], [3, 4]]

# Shallow copy β€” new outer list, but inner lists are SHARED
shallow = original.copy()   # or list(original) or original[:]
shallow[0][0] = 99           # modifies the SHARED inner list!
print(original[0][0])        # 99 ← original is also changed!

original = [[1, 2], [3, 4]]  # reset
deep = copy.deepcopy(original)  # fully independent copy
deep[0][0] = 99               # modifies deep's own copy
print(original[0][0])          # 1 ← original is unchanged βœ“

# When to use each:
# shallow: flat structures (list of strings, list of ints)
# deep:    nested/complex mutable structures
# neither: immutable objects (strings, tuples of immutables)

A decorator is a callable that takes a function as input and returns a new (usually enhanced) function. The @decorator syntax is syntactic sugar β€” @timer above a function definition is exactly equivalent to writing my_func = timer(my_func) after the definition. Decorators are used for cross-cutting concerns: logging, timing, authentication, caching, rate limiting, retry logic.

Always use @functools.wraps(func) inside your decorator β€” this copies the original function's __name__, __doc__, and other metadata to the wrapper, so debugging tools still see the original function name.

python
from functools import wraps

def require_auth(func):
    """Decorator that checks authentication before calling the function."""
    @wraps(func)   # preserves __name__, __doc__, etc.
    def wrapper(*args, **kwargs):
        request = kwargs.get("request")
        if not request or not request.get("authenticated"):
            raise PermissionError("Not authenticated")
        return func(*args, **kwargs)   # call the original function
    return wrapper   # return the wrapper (NOT the result of calling func!)

@require_auth   # same as: get_profile = require_auth(get_profile)
def get_profile(request):
    return {"name": "Alice"}

# With arguments β€” needs three levels
def retry(max_attempts=3):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for i in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception:
                    if i == max_attempts - 1:
                        raise
        return wrapper
    return decorator

@retry(max_attempts=3)  # retry(3) β†’ returns decorator β†’ applied to api_call
def api_call(): pass

Default argument values are evaluated exactly once when the function is defined β€” not each time the function is called. This means if you use a mutable object (list, dict, set) as a default, all calls that don't override that argument share the same object. Mutations in one call persist to the next call. The fix is to use None as the default and create a fresh mutable object inside the function body.

python β€” bug and fix
# BUG β€” the [] is created ONCE at function definition time
def add_item(item, items=[]):
    items.append(item)
    return items

print(add_item("a"))  # ["a"]
print(add_item("b"))  # ["a", "b"] ← the same list!
print(add_item("c"))  # ["a", "b", "c"] ← keeps growing!

# FIX β€” use None as sentinel, create fresh list each time
def add_item(item, items=None):
    if items is None:   # create a new list for each call that doesn't provide one
        items = []
    items.append(item)
    return items

print(add_item("a"))  # ["a"]
print(add_item("b"))  # ["b"] ← fresh list βœ“

The Method Resolution Order (MRO) is the order in which Python searches for a method or attribute through a class hierarchy when you have multiple inheritance. Python uses the C3 linearization algorithm which guarantees two properties: (1) subclasses always come before their parent classes, (2) parents appear in the order they are listed in the class definition. You can see the MRO with ClassName.__mro__.

The super() function follows the MRO β€” it calls the next class in the MRO chain, not necessarily the direct parent. This is important for cooperative multiple inheritance with mixins.

python
class A:
    def method(self): return "A"
class B(A):
    def method(self): return "B"
class C(A):
    def method(self): return "C"
class D(B, C):    # inherits from both B and C
    pass

print(D.__mro__)  # (<class 'D'>, <class 'B'>, <class 'C'>, <class 'A'>, <class 'object'>)
print(D().method())  # "B" β€” found in B first (before C)

# super() follows MRO β€” each class calls the NEXT in the chain
class LogMixin:
    def __init__(self):
        print("LogMixin.__init__")
        super().__init__()   # calls next in MRO β€” which might be Base!

class Base:
    def __init__(self):
        print("Base.__init__")

class Child(LogMixin, Base):  # MRO: Child β†’ LogMixin β†’ Base β†’ object
    def __init__(self):
        print("Child.__init__")
        super().__init__()    # calls LogMixin.__init__

Child()
# Child.__init__
# LogMixin.__init__
# Base.__init__

A context manager is an object that defines __enter__ (setup code) and __exit__ (cleanup code) methods. When used with with, Python calls __enter__ when entering the block and __exit__ when leaving β€” even if an exception occurs. This guarantees cleanup regardless of errors.

Use context managers for: file handles (auto-close), database connections and transactions (auto-commit/rollback), locks (auto-release), temporary state changes (auto-restore), and timers.

python
from contextlib import contextmanager

@contextmanager
def db_transaction(conn):
    """Automatically commits on success, rolls back on failure."""
    try:
        yield conn          # the 'with' block runs here
        conn.commit()       # only runs if no exception was raised
    except Exception:
        conn.rollback()     # undo changes on error
        raise               # re-raise so caller knows it failed

with db_transaction(connection) as conn:
    conn.execute("INSERT INTO orders VALUES (1, 'book')")
    conn.execute("UPDATE inventory SET qty = qty - 1")
# If either execute() fails, both are rolled back automatically

Async/await is built on top of Python's generators and coroutines. async def defines a coroutine function. When called, it returns a coroutine object β€” no code runs yet. The event loop drives execution by calling send() on coroutines to advance them.

When a coroutine hits an await, it suspends execution and yields control back to the event loop. The event loop registers the I/O operation with the OS (using select/epoll) and goes to run other coroutines. When the OS signals the I/O is complete, the event loop resumes the suspended coroutine. This is called cooperative multitasking β€” coroutines voluntarily give up control at await points.

python
import asyncio

# Sequential: 2 seconds total (waits for each in turn)
async def sequential():
    await asyncio.sleep(1)
    await asyncio.sleep(1)

# Concurrent: 1 second total (both run at the same time)
async def concurrent():
    await asyncio.gather(asyncio.sleep(1), asyncio.sleep(1))

import time
start = time.time()
asyncio.run(sequential())
print(f"Sequential: {time.time()-start:.1f}s")   # ~2.0s

start = time.time()
asyncio.run(concurrent())
print(f"Concurrent: {time.time()-start:.1f}s")   # ~1.0s

🎯Interview: LangChain & LangGraph

LangChain Expression Language (LCEL) is a declarative way to compose chains using the pipe operator |. Every component β€” prompts, LLMs, parsers, retrievers β€” implements the same Runnable interface, which means every chain automatically gets: synchronous invocation (invoke()), async invocation (ainvoke()), streaming (stream()), async streaming (astream()), and batch processing (batch()).

The old approach (LLMChain, ConversationChain) required different code paths for sync vs async, and streaming was not automatic. With LCEL you write one chain and it works everywhere.

python
chain = prompt | llm | parser

chain.invoke({"question": "What is Python?"})           # sync
await chain.ainvoke({"question": "What is Python?"})    # async
for chunk in chain.stream({"question": "..."}):         # streaming
    print(chunk, end="")
chain.batch([{"question": "q1"}, {"question": "q2"}])   # parallel batch

Retrieval-Augmented Generation (RAG) is a technique that grounds LLM responses in external documents rather than relying solely on what the model learned during training. This prevents hallucination on domain-specific or recent information. The pipeline has two phases: indexing (done once) and retrieval + generation (done for each query).

Indexing: load documents β†’ split into chunks β†’ convert each chunk to a vector embedding β†’ store in a vector database. Retrieval + generation: embed the user's query β†’ find the k most similar document chunks β†’ inject those chunks into the prompt as context β†’ ask the LLM to answer using only that context.

python β€” RAG steps
# INDEXING (done once, save to disk)
loader    = PyPDFLoader("document.pdf")
chunks    = RecursiveCharacterTextSplitter(chunk_size=1000).split_documents(loader.load())
vs        = FAISS.from_documents(chunks, OpenAIEmbeddings())
retriever = vs.as_retriever(search_type="mmr", search_kwargs={"k": 5})

# RETRIEVAL + GENERATION (done per query)
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | rag_prompt | llm | StrOutputParser()
)
answer = rag_chain.invoke("What is the return policy?")

LCEL is ideal for linear pipelines where data flows in one direction without loops or branching based on state. LangGraph is the right choice when your workflow needs: cycles/loops (retry until satisfied, agent loops), branching based on LLM output, persistent state across multiple invocations, human-in-the-loop approval steps, or multiple coordinated agents. LangGraph is essentially a state machine for AI workflows.

python
# LCEL β€” linear: A β†’ B β†’ C  (no loops possible)
chain = prompt | llm | parser

# LangGraph β€” cyclic: agent decides to call tools, loop until done
def route(state):
    if state["messages"][-1].tool_calls:
        return "tools"   # loop: go back to call tools
    return END           # done: no more tool calls needed

graph.add_conditional_edges("agent", route, {"tools": "tools", END: END})

The state in LangGraph is a TypedDict (or dataclass) that serves as the shared memory of the entire graph execution. Every node receives the full current state and returns a dict of partial updates. A reducer is a function that defines how to merge a node's returned update into the existing state value for a specific field.

The default reducer is "last-write-wins" β€” the returned value simply replaces the existing value. The add_messages reducer is a special built-in that appends new messages to the existing messages list rather than replacing the whole list. You specify the reducer with Annotated[type, reducer_func].

python
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
from operator import add

class State(TypedDict):
    messages: Annotated[list, add_messages]  # APPENDS new messages
    count:    int                             # REPLACES (last-write-wins)
    tags:     Annotated[list, add]            # APPENDS items (using operator.add)

def my_node(state: State) -> dict:
    # Return ONLY what changed β€” rest of state is untouched
    return {"count": state["count"] + 1}  # only updates 'count'

Checkpointers persist the complete graph state after each step to a storage backend (memory, SQLite, Redis, PostgreSQL). The state is keyed by thread_id in the config. On the next invocation with the same thread_id, the graph automatically loads the saved state before executing. This means all previous messages in the conversation are available without the caller re-sending them.

python
from langgraph.checkpoint.memory import MemorySaver
memory = MemorySaver()
app = graph.compile(checkpointer=memory)
config = {"configurable": {"thread_id": "chat_abc"}}

# Turn 1 β€” state saved with thread_id="chat_abc"
app.invoke({"messages": [HumanMessage("My name is Alice")]}, config=config)

# Turn 2 β€” state loaded automatically from checkpoint
app.invoke({"messages": [HumanMessage("What's my name?")]}, config=config)
# LLM sees full history and responds: "Your name is Alice!"

Tool calling (also called function calling) is a feature where the LLM, instead of generating a plain text response, outputs a structured JSON object specifying a function name and arguments to call. Modern LLMs (GPT-4, Claude) are specifically fine-tuned to output well-formed tool calls. You bind tools to the LLM with llm.bind_tools(tools). When the LLM decides to use a tool, LangChain (or your code) executes the tool and feeds the result back to the LLM for the next step.

python
from langchain_core.tools import tool

@tool
def multiply(a: int, b: int) -> int:
    """Multiply two integers together and return the product."""
    return a * b

llm_with_tools = llm.bind_tools([multiply])
response = llm_with_tools.invoke("What is 42 times 7?")

if response.tool_calls:
    tool_call = response.tool_calls[0]
    # {"name": "multiply", "args": {"a": 42, "b": 7}}
    result = multiply.invoke(tool_call["args"])  # 294

Similarity search returns the k chunks with the highest cosine similarity to the query vector. This can return near-duplicate chunks β€” for example, if your document mentions "return policy" five times on different pages, all five might be returned, leaving no room for other relevant sections.

MMR (Maximum Marginal Relevance) balances relevance with diversity. It first fetches a larger set (e.g., 20 candidates), then iteratively selects the document that is both relevant to the query AND maximally different from already-selected documents. The lambda_mult parameter controls the trade-off: 0.0 = maximum diversity, 1.0 = pure relevance (same as similarity search).

python
mmr_retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={
        "k":           5,     # final documents to return
        "fetch_k":     20,    # candidates to consider
        "lambda_mult": 0.5,   # 0.5 = balanced between relevance and diversity
    }
)

🎯Interview: FastAPI

FastAPI's performance comes from three sources: (1) Starlette β€” an ASGI framework that supports native async/await, allowing a single worker process to handle thousands of concurrent requests without threads; (2) Pydantic v2 β€” the validation layer is written in Rust, making it extremely fast for data parsing and validation; (3) the async-first design means I/O-bound operations (database queries, external API calls) don't block the event loop.

Flask uses WSGI (synchronous) and has no built-in validation. Django also uses WSGI by default and its validation (forms/serializers) is slower than Pydantic. FastAPI also generates OpenAPI documentation automatically from your code, which Flask and Django require third-party libraries to do.

FeatureFastAPIFlaskDjango
Async nativeYes (ASGI)Partial (via Quart)Partial
Auto validationPydantic (Rust)ManualForms/DRF
Auto docsBuilt-in OpenAPIExternal lib neededDRF browsable
PerformanceVery highMediumMedium
Learning curveLow-MediumLowHigh

FastAPI resolves Depends() at request time by calling the dependency function (or class constructor) and injecting the result as a function argument. Dependencies can yield for resource lifecycle management β€” code before the yield runs before the endpoint, code after the yield runs in a finally block after the endpoint (cleanup guaranteed even on errors). Dependencies are composable β€” a dependency can itself have dependencies.

python
def get_db():
    db = SessionLocal()   # open connection
    try:
        yield db           # provide to endpoint
    finally:
        db.close()         # ALWAYS closes β€” even if endpoint throws

def get_current_user(
    db: Session = Depends(get_db),          # composes get_db
    token: str  = Depends(oauth2_scheme),   # and oauth2_scheme
) -> User:
    return db.query(User).filter_by(token=token).first()

@app.get("/profile")
def profile(user: User = Depends(get_current_user)):
    return user   # FastAPI called get_db, then get_current_user, then profile

Use async def when your endpoint uses async I/O β€” await on database queries, await on external HTTP calls (aiohttp, httpx), or any other awaitable operation. This allows FastAPI to handle other requests while waiting for I/O.

Use plain def when your endpoint uses blocking/synchronous libraries (like the standard requests library, or synchronous SQLAlchemy). FastAPI automatically runs sync endpoints in a thread pool, so they don't block the event loop either.

The critical mistake to avoid: using synchronous blocking code inside an async def endpoint. This blocks the event loop and prevents ALL other requests from being served β€” worse than a regular sync endpoint!

python
@app.get("/async-ok")    # βœ“ async I/O β€” doesn't block event loop
async def async_endpoint():
    async with httpx.AsyncClient() as client:
        r = await client.get("https://api.example.com")
        return r.json()

@app.get("/sync-ok")     # βœ“ sync in thread pool β€” also doesn't block event loop
def sync_endpoint():
    return requests.get("https://api.example.com").json()

@app.get("/bad")         # βœ— BLOCKING I/O in async context β€” blocks EVERYTHING!
async def bad_endpoint():
    return requests.get("https://api.example.com").json()  # blocks event loop!
python β€” SQLAlchemy + FastAPI
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session, declarative_base
from fastapi import Depends

DATABASE_URL = "postgresql://user:pass@localhost/mydb"
engine       = create_engine(DATABASE_URL, pool_size=10, max_overflow=20)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base         = declarative_base()

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

@app.get("/users")
def list_users(db: Session = Depends(get_db)):
    return db.query(User).all()

# Async SQLAlchemy 2.0 + asyncpg
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
async_engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/mydb")
AsyncSession  = async_sessionmaker(async_engine, expire_on_commit=False)

async def get_async_db():
    async with AsyncSession() as session:
        yield session

@app.get("/users-async")
async def list_users_async(db: AsyncSession = Depends(get_async_db)):
    result = await db.execute(select(User))
    return result.scalars().all()
project structure
app/
β”œβ”€β”€ main.py              # FastAPI instance, lifespan, include routers
β”œβ”€β”€ config.py            # Settings via pydantic-settings (reads .env)
β”œβ”€β”€ database.py          # engine, SessionLocal, Base
β”œβ”€β”€ models/              # SQLAlchemy ORM models (database tables)
β”‚   β”œβ”€β”€ __init__.py
β”‚   β”œβ”€β”€ user.py
β”‚   └── product.py
β”œβ”€β”€ schemas/             # Pydantic models (request/response shapes)
β”‚   β”œβ”€β”€ user.py          # UserCreate, UserResponse, UserUpdate
β”‚   └── product.py
β”œβ”€β”€ routers/             # APIRouter β€” one file per resource/domain
β”‚   β”œβ”€β”€ users.py         # /users endpoints
β”‚   └── products.py      # /products endpoints
β”œβ”€β”€ dependencies.py      # Shared Depends() functions (get_db, get_current_user)
β”œβ”€β”€ services/            # Business logic (no HTTP concerns)
β”‚   └── user_service.py
└── utils/               # Helpers (hashing, email, etc.)

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

app = FastAPI(lifespan=lifespan)
app.include_router(users.router,    prefix="/users",    tags=["users"])
app.include_router(products.router, prefix="/products", tags=["products"])

# routers/users.py
from fastapi import APIRouter
router = APIRouter()

@router.get("/")
def list_users(): ...

@router.post("/")
def create_user(): ...
python β€” testing with TestClient and pytest
from fastapi.testclient import TestClient
from main import app
import pytest

client = TestClient(app)

def test_create_user():
    response = client.post("/users", json={
        "name": "Alice", "email": "alice@x.com", "age": 30
    })
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == "Alice"
    assert "id" in data

def test_user_not_found():
    response = client.get("/users/nonexistent-id-999")
    assert response.status_code == 404
    assert response.json()["detail"] == "User not found"

# Override dependencies for testing (inject test DB instead of real DB)
def override_db():
    yield {"connected": True}   # in-memory test "database"

app.dependency_overrides[get_db] = override_db  # swap the dependency

# Async testing
import pytest_asyncio
from httpx import AsyncClient

@pytest.mark.asyncio
async def test_async_route():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        r = await ac.get("/async-route")
    assert r.status_code == 200

response_model serves three purposes: (1) Security filtering β€” it strips any fields not defined in the response model. If your database model has a hashed_password field, setting response_model=UserPublic (which doesn't include it) ensures the password never leaves the server. (2) Documentation β€” FastAPI uses the response model to generate accurate OpenAPI docs showing exactly what the endpoint returns. (3) Validation β€” FastAPI validates the return value matches the model, catching bugs where endpoints return wrong data shapes.

python
class UserInternal(BaseModel):
    id: int; name: str; hashed_password: str

class UserPublic(BaseModel):
    id: int; name: str   # no hashed_password!

@app.get("/me", response_model=UserPublic)  # ← this filters the response
def get_me():
    user = UserInternal(id=1, name="Alice", hashed_password="$2b$12$...")
    return user   # FastAPI automatically removes hashed_password βœ“

# Useful response_model options:
@app.get("/me", response_model=UserPublic,
    response_model_exclude_unset=True,   # omit fields that weren't set
    response_model_include={"id","name"},  # whitelist specific fields
    response_model_exclude={"secret"},     # blacklist specific fields
)
Copied!