π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.
π¦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:
TrueorFalse - None β represents "nothing" or "no value yet". Like an empty box with no contents.
# 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.
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.
# 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 {}.
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.
# 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.
# 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.
# 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.
# 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.
# 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?
| Type | Syntax | Ordered? | Mutable? | Duplicates? | Best For |
|---|---|---|---|---|---|
| list | [1,2,3] | Yes | Yes | Yes | Ordered data you'll add/remove items from |
| tuple | (1,2,3) | Yes | No | Yes | Fixed data (coordinates, RGB, function returns) |
| dict | {"a":1} | Insertion order | Yes | No (keys) | Looking up values by a meaningful key |
| set | {1,2,3} | No | Yes | No | Membership 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.
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).
# βββ 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.
# 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).
# βββ 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".
# βββ 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.
# 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!# 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!# βββ 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.
# βββ 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
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:
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".
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.
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.
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.
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.
# βββ 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.
# βββ 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.
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")
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.
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.
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.
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.
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.
|. 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.
pip install langchain langchain-openai langchain-community langchain-core
pip install langchain-anthropic # for Claude models
pip install python-dotenv # for loading .env file
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.
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.
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.
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!
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.
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.
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.
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_messagesappends to a list.
pip install langgraph langchain-openai
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.
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.
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.
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)
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 /userswith 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
/docsand 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
/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.
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!)
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.
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.
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.
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.
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.
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
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.
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.
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.
# 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.pyPython 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.
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 triggerA 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.
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.
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.
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(): passDefault 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.
# 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.
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.
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 automaticallyAsync/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.
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.
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 batchRetrieval-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.
# 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.
# 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].
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.
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.
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"]) # 294Similarity 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).
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.
| Feature | FastAPI | Flask | Django |
|---|---|---|---|
| Async native | Yes (ASGI) | Partial (via Quart) | Partial |
| Auto validation | Pydantic (Rust) | Manual | Forms/DRF |
| Auto docs | Built-in OpenAPI | External lib needed | DRF browsable |
| Performance | Very high | Medium | Medium |
| Learning curve | Low-Medium | Low | High |
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.
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 profileUse 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!
@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!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()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(): ...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 == 200response_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.
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
)