import this
# Beautiful is better than ugly.
# Explicit is better than implicit.
# Simple is better than complex.
# Readability counts.

python --version
# or
python3 --version
# Expected: Python 3.12.0

pip --version

print("Hello, World!")
# Output: Hello, World!

# Single-line comment
"""
Multi-line comment (docstring)
"""
print("Comments make code readable!")  # Inline comment

>>> print("Hello from REPL!")
Hello from REPL!
>>> 2 + 2
4
>>> name = "Python"
>>> print(f"I love {name}!")
I love Python!
>>> exit()

# Save as hello.py then run:
python hello.py
# or
python3 hello.py

# Python 2 (OLD - Don't use!)
print "Hello"       # statement, not function
# 5 / 2 = 2         # integer division

# Python 3 (CURRENT)
print("Hello")      # function
# 5 / 2 = 2.5       # true division
input("Enter: ")    # input() replaces raw_input()

name       = "Alice"
age        = 25
height     = 5.6
is_student = True
nothing    = None

# Multiple assignment
x = y = z = 0
a, b, c = 1, 2, 3   # Unpacking

# ✅ Valid
my_name = "Bob"
age2    = 30
_private = "hidden"
PI      = 3.14159   # Convention for constants

# ❌ Invalid
# 2name = "Bob"     # Cannot start with number
# my-name = "Bob"   # Hyphens not allowed
# class = "Python"  # Reserved keyword

import keyword
print(keyword.kwlist)  # See all reserved words

age      = 25          # int
big_num  = 1_000_000  # underscores for readability
pi       = 3.14159    # float
temp     = -23.5      # negative float
sci      = 1.5e10     # scientific notation

print(type(age))             # <class 'int'>
print(type(pi))              # <class 'float'>
print(isinstance(age, int))  # True

name    = "Alice"
city    = 'London'
message = """This is
a multi-line string"""

# f-strings (Python 3.6+) — preferred way
age      = 25
greeting = f"Hello, I am {name} and I am {age} years old."
print(greeting)

print(len(name))       # 5
print(name.upper())    # ALICE
print(name[0])         # A  (indexing)
print(name[1:3])       # li (slicing)

is_active = True
is_admin  = False

# Falsy values: False, 0, 0.0, "", [], {}, None
print(bool(0))      # False
print(bool(""))     # False
print(bool(42))     # True
print(bool("hi"))   # True

result = None
if result is None:  # Use 'is', not '=='
    print("No result yet")

num_str   = "42"
num_int   = int(num_str)     # "42" → 42
num_float = float(num_str)   # "42" → 42.0
back_str  = str(num_int)     # 42  → "42"

x = 3.9
print(int(x))    # 3 (truncates, doesn't round!)
print(round(x))  # 4

print(int(True))   # 1
print(int(False))  # 0

a, b = 17, 5

print(a + b)   # 22       Addition
print(a - b)   # 12       Subtraction
print(a * b)   # 85       Multiplication
print(a / b)   # 3.4      True division (float)
print(a // b)  # 3        Floor division (int)
print(a % b)   # 2        Modulus (remainder)
print(a ** b)  # 1419857  Exponentiation

# PEMDAS / BODMAS
print(2 + 3 * 4)    # 14 (not 20!)
print((2 + 3) * 4)  # 20

x, y = 10, 20

print(x == y)   # False  Equal to
print(x != y)   # True   Not equal
print(x < y)    # True   Less than
print(x > y)    # False  Greater than
print(x <= 10)  # True   Less than or equal
print(x >= 10)  # True   Greater than or equal

# Chained comparisons (Pythonic!)
age = 25
print(18 <= age <= 65)  # True

# Logical operators
has_id    = True
is_member = False
print(age >= 18 and has_id)    # True
print(age >= 18 or is_member)  # True
print(not is_member)           # True

x = 10
x += 5    # x = 15
x -= 3    # x = 12
x *= 2    # x = 24
x /= 4    # x = 6.0
x //= 2   # x = 3.0
x **= 3   # x = 27.0
x %= 5    # x = 2.0

# Walrus operator := (Python 3.8+)
numbers = [1, 2, 3, 4, 5]
if (n := len(numbers)) > 3:
    print(f"List has {n} elements")

# input() ALWAYS returns a STRING
name = input("What is your name? ")
print(f"Hello, {name}!")

# Convert to number
age = int(input("How old are you? "))
print(f"In 10 years you'll be {age + 10}")

# Safe input with error handling
try:
    number = int(input("Enter a number: "))
    print(f"Double: {number * 2}")
except ValueError:
    print("That's not a valid number!")

age = 18

if age >= 18:
    print("You are an adult.")

# Nothing happens if condition is False
# Indentation is MANDATORY in Python!

age = 15

if age >= 18:
    print("You can vote.")
else:
    print("You cannot vote yet.")

# Output: You cannot vote yet.

score = 75

if score >= 90:
    grade = "A"
elif score >= 80:
    grade = "B"
elif score >= 70:
    grade = "C"
elif score >= 60:
    grade = "D"
else:
    grade = "F"

print(f"Your grade is: {grade}")
# Output: Your grade is: C

# value_if_true if condition else value_if_false
age = 20
status = "adult" if age >= 18 else "minor"
print(status)  # adult

# Nested ternary (use sparingly!)
x = 5
result = "positive" if x > 0 else "negative" if x < 0 else "zero"
print(result)  # positive

age = 25
has_license = True

if age >= 18:
    if has_license:
        print("You can drive.")
    else:
        print("You need a license to drive.")
else:
    print("You are too young to drive.")

# Better way using 'and':
if age >= 18 and has_license:
    print("You can drive.")

# Iterating over a list
fruits = ["apple", "banana", "cherry"]
for fruit in fruits:
    print(fruit)
# apple
# banana
# cherry

# Iterating over a string
for char in "Python":
    print(char, end=" ")
# P y t h o n

# range(stop)
for i in range(5):
    print(i, end=" ")
# 0 1 2 3 4

# range(start, stop)
for i in range(1, 6):
    print(i, end=" ")
# 1 2 3 4 5

# range(start, stop, step)
for i in range(0, 20, 5):
    print(i, end=" ")
# 0 5 10 15

# Counting backwards
for i in range(10, 0, -1):
    print(i, end=" ")
# 10 9 8 7 6 5 4 3 2 1

fruits = ["apple", "banana", "cherry"]

for index, fruit in enumerate(fruits):
    print(f"{index}: {fruit}")
# 0: apple
# 1: banana
# 2: cherry

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

names  = ["Alice", "Bob", "Charlie"]
scores = [95, 87, 92]

for name, score in zip(names, scores):
    print(f"{name}: {score}")
# Alice: 95
# Bob: 87
# Charlie: 92

count = 1

while count <= 5:
    print(f"Count: {count}")
    count += 1   # IMPORTANT: update the condition variable!

# Count: 1
# Count: 2
# Count: 3
# Count: 4
# Count: 5

secret = "python"

while True:
    guess = input("Guess the password: ").lower()
    if guess == secret:
        print("Access granted!")
        break
    else:
        print("Wrong! Try again.")

# break exits the loop immediately
for i in range(10):
    if i == 5:
        break
    print(i, end=" ")
# 0 1 2 3 4

# Find first even number
numbers = [1, 3, 7, 4, 9, 2]
for num in numbers:
    if num % 2 == 0:
        print(f"First even: {num}")
        break
# First even: 4

# continue skips the rest of the current iteration
for i in range(10):
    if i % 2 == 0:
        continue    # skip even numbers
    print(i, end=" ")
# 1 3 5 7 9

# Skip empty strings
names = ["Alice", "", "Bob", "", "Charlie"]
for name in names:
    if not name:
        continue
    print(f"Hello, {name}!")

# pass is a placeholder — does nothing
for i in range(5):
    if i == 3:
        pass    # TODO: handle this case later
    print(i, end=" ")
# 0 1 2 3 4

# Useful for empty function/class bodies
def my_future_function():
    pass   # implement later

# else runs only if loop completed WITHOUT break
numbers = [1, 3, 5, 7, 9]

for num in numbers:
    if num % 2 == 0:
        print(f"Found even: {num}")
        break
else:
    print("No even numbers found!")
# No even numbers found!

# while/else
count = 0
while count < 3:
    count += 1
else:
    print(f"Loop finished. count = {count}")
# Loop finished. count = 3

# Multiplication table
for i in range(1, 4):
    for j in range(1, 4):
        print(f"{i} x {j} = {i*j}")
    print("---")
# 1 x 1 = 1
# 1 x 2 = 2
# 1 x 3 = 3
# ---
# 2 x 1 = 2  ... etc.

# Pattern printing
for i in range(1, 6):
    print("*" * i)
# *
# **
# ***
# ****
# *****

# Define a function
def greet():
    print("Hello, World!")

# Call the function
greet()
# Output: Hello, World!

# Functions can be called multiple times
greet()
greet()

# Function with a docstring (documentation)
def greet():
    """This function prints a greeting message."""
    print("Hello, World!")

print(greet.__doc__)
# This function prints a greeting message.

# Single parameter
def greet(name):
    print(f"Hello, {name}!")

greet("Alice")   # Hello, Alice!
greet("Bob")     # Hello, Bob!

# Multiple parameters
def add(a, b):
    print(f"{a} + {b} = {a + b}")

add(3, 5)    # 3 + 5 = 8
add(10, 20)  # 10 + 20 = 30

# Single return value
def square(n):
    return n ** 2

result = square(5)
print(result)   # 25

# Multiple return values (returns a tuple)
def min_max(numbers):
    return min(numbers), max(numbers)

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

# Early return
def divide(a, b):
    if b == 0:
        return None   # Early exit
    return a / b

print(divide(10, 2))   # 5.0
print(divide(10, 0))   # None

def greet(name, greeting="Hello"):
    print(f"{greeting}, {name}!")

greet("Alice")              # Hello, Alice!
greet("Bob", "Good morning") # Good morning, Bob!
greet("Charlie", "Hi")      # Hi, Charlie!

# Practical example
def create_user(name, role="user", active=True):
    return {"name": name, "role": role, "active": active}

print(create_user("Alice"))
# {'name': 'Alice', 'role': 'user', 'active': True}

print(create_user("Bob", role="admin"))
# {'name': 'Bob', 'role': 'admin', 'active': True}

def describe_pet(name, animal, age):
    print(f"{name} is a {age}-year-old {animal}.")

# Positional arguments (order matters)
describe_pet("Rex", "dog", 3)

# Keyword arguments (order doesn't matter)
describe_pet(age=3, name="Rex", animal="dog")
describe_pet(name="Whiskers", animal="cat", age=5)

# Mix: positional first, then keyword
describe_pet("Rex", animal="dog", age=3)

def add_all(*args):
    """Add any number of numbers."""
    return sum(args)

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

def greet_all(*names):
    for name in names:
        print(f"Hello, {name}!")

greet_all("Alice", "Bob", "Charlie")

def print_info(**kwargs):
    """Print any key-value pairs."""
    for key, value in kwargs.items():
        print(f"  {key}: {value}")

print_info(name="Alice", age=25, city="London")
# name: Alice
# age: 25
# city: London

# Combining all argument types
def full_example(required, *args, default="hi", **kwargs):
    print(f"Required: {required}")
    print(f"Args: {args}")
    print(f"Default: {default}")
    print(f"Kwargs: {kwargs}")

full_example("must", 1, 2, 3, default="hello", x=10, y=20)

x = 10   # Global variable

def my_func():
    y = 20   # Local variable
    print(x)  # Can READ global variable
    print(y)  # Can access local variable

my_func()
print(x)   # 10 — accessible
# print(y) # NameError — y is local to my_func!

# Modifying global variable inside function
counter = 0

def increment():
    global counter    # declare intent to modify global
    counter += 1

increment()
increment()
print(counter)  # 2

def outer():
    count = 0

    def inner():
        nonlocal count   # modify enclosing scope variable
        count += 1
        print(f"Inner count: {count}")

    inner()
    inner()
    print(f"Outer count: {count}")

outer()
# Inner count: 1
# Inner count: 2
# Outer count: 2

# Syntax: lambda arguments: expression
square = lambda x: x ** 2
print(square(5))   # 25

add = lambda a, b: a + b
print(add(3, 4))   # 7

# Commonly used with sorted(), map(), filter()
names = ["Charlie", "Alice", "Bob"]
names.sort(key=lambda x: x.lower())
print(names)  # ['Alice', 'Bob', 'Charlie']

numbers = [1, 2, 3, 4, 5, 6]
evens = list(filter(lambda x: x % 2 == 0, numbers))
print(evens)  # [2, 4, 6]

doubled = list(map(lambda x: x * 2, numbers))
print(doubled)  # [2, 4, 6, 8, 10, 12]

def factorial(n):
    """Calculate n! recursively."""
    if n == 0 or n == 1:   # Base case
        return 1
    return n * factorial(n - 1)  # Recursive case

print(factorial(5))   # 120  (5×4×3×2×1)
print(factorial(10))  # 3628800

# Fibonacci sequence
def fibonacci(n):
    if n <= 1:
        return n
    return fibonacci(n - 1) + fibonacci(n - 2)

for i in range(8):
    print(fibonacci(i), end=" ")
# 0 1 1 2 3 5 8 13

def make_multiplier(factor):
    """Returns a function that multiplies by factor."""
    def multiplier(number):
        return number * factor   # 'factor' is remembered!
    return multiplier

double = make_multiplier(2)
triple = make_multiplier(3)

print(double(5))   # 10
print(triple(5))   # 15
print(double(10))  # 20

# Counter using closure
def make_counter():
    count = 0
    def counter():
        nonlocal count
        count += 1
        return count
    return counter

my_counter = make_counter()
print(my_counter())  # 1
print(my_counter())  # 2
print(my_counter())  # 3

# Empty list
empty = []
empty = list()

# List of integers
numbers = [1, 2, 3, 4, 5]

# List of strings
fruits = ["apple", "banana", "cherry"]

# Mixed types
mixed = [1, "hello", 3.14, True, None]

# Nested list
matrix = [[1, 2, 3], [4, 5, 6], [7, 8, 9]]

# List from range
squares = list(range(1, 6))
print(squares)  # [1, 2, 3, 4, 5]

fruits = ["apple", "banana", "cherry", "date", "elderberry"]

# Positive indexing (0-based)
print(fruits[0])   # apple
print(fruits[2])   # cherry

# Negative indexing (from end)
print(fruits[-1])  # elderberry
print(fruits[-2])  # date

# Slicing [start:stop:step]
print(fruits[1:3])    # ['banana', 'cherry']
print(fruits[:3])     # ['apple', 'banana', 'cherry']
print(fruits[2:])     # ['cherry', 'date', 'elderberry']
print(fruits[::2])    # ['apple', 'cherry', 'elderberry']
print(fruits[::-1])   # reversed list

fruits = ["apple", "banana", "cherry"]

# Change an element
fruits[1] = "blueberry"
print(fruits)  # ['apple', 'blueberry', 'cherry']

# Add elements
fruits.append("date")          # add to end
fruits.insert(1, "avocado")    # insert at index
fruits.extend(["elderberry", "fig"])  # add multiple

# Remove elements
fruits.remove("avocado")       # remove by value (first match)
popped = fruits.pop()          # remove & return last item
popped2 = fruits.pop(0)        # remove & return at index
del fruits[0]                  # delete at index

# Clear all
fruits.clear()                 # empty the list

numbers = [3, 1, 4, 1, 5, 9, 2, 6, 5]

print(len(numbers))          # 9  — length
print(numbers.count(1))      # 2  — count occurrences
print(numbers.index(5))      # 4  — first index of value
print(sum(numbers))          # 36 — sum
print(min(numbers))          # 1  — minimum
print(max(numbers))          # 9  — maximum

numbers.sort()               # sort in-place (ascending)
print(numbers)               # [1, 1, 2, 3, 4, 5, 5, 6, 9]

numbers.sort(reverse=True)   # sort descending
numbers.reverse()            # reverse in-place

sorted_copy = sorted(numbers)  # returns NEW sorted list
copy = numbers.copy()          # shallow copy

# Check membership
print(5 in numbers)   # True
print(99 in numbers)  # False

fruits = ["apple", "banana", "cherry"]

# Simple iteration
for fruit in fruits:
    print(fruit)

# With index using enumerate
for i, fruit in enumerate(fruits):
    print(f"{i}: {fruit}")

# List comprehension (covered in Ch 13)
upper = [f.upper() for f in fruits]
print(upper)  # ['APPLE', 'BANANA', 'CHERRY']

# Nested list access
matrix = [[1, 2, 3], [4, 5, 6], [7, 8, 9]]
for row in matrix:
    for val in row:
        print(val, end=" ")
    print()

# Creating tuples
empty   = ()
single  = (42,)          # ← comma required for single item!
coords  = (10.5, 20.3)
colors  = ("red", "green", "blue")
mixed   = (1, "hello", True, 3.14)

# Tuple packing (no parentheses needed)
point = 3, 4
print(point)        # (3, 4)
print(type(point))  # 

# Tuple unpacking
x, y = point
print(x, y)         # 3 4

# Extended unpacking
first, *rest = (1, 2, 3, 4, 5)
print(first)  # 1
print(rest)   # [2, 3, 4, 5]

colors = ("red", "green", "blue", "red")

# Indexing and slicing (same as lists)
print(colors[0])     # red
print(colors[-1])    # red
print(colors[1:3])   # ('green', 'blue')

# Tuple methods (only 2!)
print(colors.count("red"))   # 2
print(colors.index("green")) # 1

# Length and membership
print(len(colors))           # 4
print("blue" in colors)      # True

# Concatenation and repetition
t1 = (1, 2, 3)
t2 = (4, 5, 6)
print(t1 + t2)    # (1, 2, 3, 4, 5, 6)
print(t1 * 2)     # (1, 2, 3, 1, 2, 3)

# Convert between list and tuple
lst = list(colors)   # tuple → list
tpl = tuple(lst)     # list → tuple

# Empty dictionary
empty = {}
empty = dict()

# Basic dictionary
person = {
    "name":  "Alice",
    "age":   25,
    "city":  "London"
}

# Using dict() constructor
car = dict(make="Toyota", model="Camry", year=2023)

# Nested dictionary
students = {
    "alice": {"grade": "A", "score": 95},
    "bob":   {"grade": "B", "score": 82}
}

# Access nested
print(students["alice"]["score"])  # 95

person = {"name": "Alice", "age": 25, "city": "London"}

# Access by key
print(person["name"])         # Alice

# .get() — safe access (no KeyError)
print(person.get("age"))      # 25
print(person.get("email"))    # None
print(person.get("email", "N/A"))  # N/A (default)

# Add or update
person["email"] = "[email protected]"  # add new key
person["age"]   = 26                   # update existing

# Delete
del person["city"]             # delete key
removed = person.pop("email")  # remove & return value
person.popitem()               # remove last inserted pair

# Check key existence
print("name" in person)        # True
print("phone" in person)       # False

person = {"name": "Alice", "age": 25, "city": "London"}

# Views — dynamic, reflect changes
print(person.keys())    # dict_keys(['name', 'age', 'city'])
print(person.values())  # dict_values(['Alice', 25, 'London'])
print(person.items())   # dict_items([('name','Alice'), ...])

# Iterate
for key in person:
    print(key, ":", person[key])

for key, value in person.items():
    print(f"  {key} → {value}")

# Merge dictionaries
defaults = {"theme": "light", "lang": "en"}
settings = {"lang": "fr", "font": "Arial"}

# Python 3.9+ merge operator
merged = defaults | settings
print(merged)
# {'theme': 'light', 'lang': 'fr', 'font': 'Arial'}

# update() — modifies in-place
defaults.update(settings)

# setdefault — set only if key missing
person.setdefault("email", "[email protected]")

# {key: value for item in iterable}
squares = {x: x**2 for x in range(1, 6)}
print(squares)
# {1: 1, 2: 4, 3: 9, 4: 16, 5: 25}

# With condition
even_squares = {x: x**2 for x in range(1, 11) if x % 2 == 0}
print(even_squares)
# {2: 4, 4: 16, 6: 36, 8: 64, 10: 100}

# Invert a dictionary
original = {"a": 1, "b": 2, "c": 3}
inverted = {v: k for k, v in original.items()}
print(inverted)  # {1: 'a', 2: 'b', 3: 'c'}

# Creating sets
empty   = set()          # NOT {} — that's a dict!
numbers = {1, 2, 3, 4, 5}
fruits  = {"apple", "banana", "cherry"}

# Duplicates are automatically removed
dupes = {1, 2, 2, 3, 3, 3, 4}
print(dupes)  # {1, 2, 3, 4}

# Create from list (remove duplicates)
names = ["Alice", "Bob", "Alice", "Charlie", "Bob"]
unique = set(names)
print(unique)  # {'Alice', 'Bob', 'Charlie'}

# Add and remove
fruits.add("date")
fruits.discard("banana")   # no error if missing
fruits.remove("cherry")    # raises KeyError if missing
popped = fruits.pop()      # remove random element

# Membership (very fast!)
print("apple" in fruits)   # True

a = {1, 2, 3, 4, 5}
b = {4, 5, 6, 7, 8}

# Union — all elements from both
print(a | b)           # {1, 2, 3, 4, 5, 6, 7, 8}
print(a.union(b))

# Intersection — only common elements
print(a & b)           # {4, 5}
print(a.intersection(b))

# Difference — in a but not b
print(a - b)           # {1, 2, 3}
print(a.difference(b))

# Symmetric difference — in either but not both
print(a ^ b)           # {1, 2, 3, 6, 7, 8}
print(a.symmetric_difference(b))

# Subset and superset
small = {1, 2}
print(small.issubset(a))    # True  — small ⊆ a
print(a.issuperset(small))  # True  — a ⊇ small
print(a.isdisjoint({9,10})) # True  — no common elements

# LIST — ordered, changeable collection
shopping_cart = ["milk", "eggs", "bread"]
shopping_cart.append("butter")

# TUPLE — fixed record / coordinates
location    = (25.2048, 55.2708)   # Dubai GPS coords
rgb_red     = (255, 0, 0)
db_record   = ("Alice", 25, "admin")

# DICTIONARY — structured data / lookup table
user = {"id": 1, "name": "Alice", "role": "admin"}
word_count = {"hello": 5, "world": 3, "python": 10}

# SET — unique items / fast membership test
visited_pages = {"/home", "/about", "/contact"}
valid_roles   = {"admin", "user", "moderator"}

if user["role"] in valid_roles:
    print("Valid role!")

# Single and double quotes — interchangeable
name    = 'Alice'
message = "Hello, World!"

# Use one type to include the other
quote   = "It's a great day!"
dialog  = 'He said "Python is awesome!"'

# Triple quotes — multiline strings
poem = """Roses are red,
Violets are blue,
Python is awesome,
And so are you!"""

# Raw strings — backslashes are literal
path    = r"C:\Users\Alice\Documents"
pattern = r"\d+\.\d+"   # regex pattern

# String from other types
num_str = str(42)        # "42"
pi_str  = str(3.14159)   # "3.14159"

print(type(name))        # <class 'str'>

text = "Python"
#       P  y  t  h  o  n
# idx:  0  1  2  3  4  5
# neg: -6 -5 -4 -3 -2 -1

print(text[0])     # P
print(text[-1])    # n
print(text[1:4])   # yth
print(text[:3])    # Pyt
print(text[3:])    # hon
print(text[::2])   # Pto   (every 2nd char)
print(text[::-1])  # nohtyP (reversed!)

# Strings are immutable — cannot assign to index
# text[0] = "J"  ← TypeError!

# Concatenation
first = "Hello"
last  = "World"
full  = first + ", " + last + "!"
print(full)   # Hello, World!

# Repetition
line  = "-" * 30
print(line)   # ------------------------------

# Membership
print("ell" in "Hello")    # True
print("xyz" in "Hello")    # False
print("xyz" not in "Hello") # True

# Length
print(len("Python"))       # 6

# Iteration
for char in "Hi!":
    print(char)   # H  i  !

# Common escape characters
print("Hello\nWorld")    # newline
print("Hello\tWorld")    # tab
print("She said \"Hi\"") # double quote
print('It\'s fine')      # single quote
print("Back\\slash")     # backslash
print("\u2764")          # Unicode heart ❤

# Raw string ignores escape sequences
print(r"C:\new\folder")  # C:\new\folder (literal \n \f)

# Multiline with \
long_str = "This is a very long string that " \
           "continues on the next line."
print(long_str)

name  = "Alice"
age   = 25
score = 95.678

# Basic f-string
print(f"Name: {name}, Age: {age}")
# Name: Alice, Age: 25

# Expressions inside f-strings
print(f"Next year: {age + 1}")
print(f"Upper: {name.upper()}")
print(f"Square: {age ** 2}")

# Number formatting
print(f"Score: {score:.2f}")       # 95.68  (2 decimal places)
print(f"Score: {score:08.2f}")     # 00095.68 (padded)
print(f"Big num: {1000000:,}")     # 1,000,000 (comma separator)
print(f"Percent: {0.876:.1%}")     # 87.6%
print(f"Hex: {255:#x}")            # 0xff

# Alignment
print(f"{'left':<10}|")    # left      |
print(f"{'right':>10}|")   # right|
print(f"{'center':^10}|")  #   center  |

# Debug with = (Python 3.8+)
x = 42
print(f"{x = }")   # x = 42

# Positional
print("Hello, {}! You are {} years old.".format("Alice", 25))

# Named placeholders
print("Hello, {name}! You are {age}.".format(name="Alice", age=25))

# Indexed
print("{0} and {1}, {0} again".format("Alice", "Bob"))
# Alice and Bob, Alice again

# Number formatting
print("Pi is {:.4f}".format(3.14159))  # Pi is 3.1416
print("{:,}".format(1000000))          # 1,000,000

# Old-style formatting (still seen in legacy code)
name  = "Alice"
score = 95.5

print("Hello, %s!" % name)
print("Score: %.1f" % score)
print("Name: %s, Age: %d" % ("Alice", 25))

# Format specifiers:
# %s = string   %d = integer
# %f = float    %x = hex

text = "  Hello, World!  "

# Case conversion
print(text.upper())       # "  HELLO, WORLD!  "
print(text.lower())       # "  hello, world!  "
print(text.title())       # "  Hello, World!  "
print(text.capitalize())  # "  hello, world!  " (only 1st char)
print(text.swapcase())    # "  hELLO, wORLD!  "

# Case checking
print("HELLO".isupper())  # True
print("hello".islower())  # True
print("Hello World".istitle())  # True

# Whitespace stripping
print(text.strip())       # "Hello, World!"
print(text.lstrip())      # "Hello, World!  "
print(text.rstrip())      # "  Hello, World!"

# Strip specific characters
print("###hello###".strip("#"))   # hello
print("...hello...".strip("."))   # hello

text = "Python is great. Python is fun!"

# find() — returns index, -1 if not found
print(text.find("Python"))      # 0
print(text.find("Python", 5))   # 17 (search from index 5)
print(text.find("Java"))        # -1

# index() — like find() but raises ValueError if not found
print(text.index("great"))      # 10
# text.index("Java")  ← ValueError!

# rfind() / rindex() — search from right
print(text.rfind("Python"))     # 17

# count() — count occurrences
print(text.count("Python"))     # 2
print(text.count("is"))         # 2

# startswith / endswith
print(text.startswith("Python"))  # True
print(text.endswith("fun!"))      # True
print(text.startswith(("Hi", "Python")))  # True (tuple check)

text = "Python is great. Python is fun!"

# replace(old, new)
print(text.replace("Python", "Java"))
# Java is great. Java is fun!

# replace with count limit
print(text.replace("Python", "Java", 1))
# Java is great. Python is fun!

# Chaining replacements
clean = "  Hello,   World!  "
clean = clean.strip().replace("  ", " ")
print(clean)  # Hello, World!

# translate() — character-by-character replacement
table = str.maketrans("aeiou", "AEIOU")
print("hello world".translate(table))
# hEllO wOrld

# split() — string → list
sentence = "Python is awesome"
words = sentence.split()          # split on whitespace
print(words)  # ['Python', 'is', 'awesome']

csv_line = "Alice,25,London,Engineer"
fields = csv_line.split(",")
print(fields)  # ['Alice', '25', 'London', 'Engineer']

# split with maxsplit
print("a:b:c:d".split(":", 2))   # ['a', 'b', 'c:d']

# splitlines() — split on line breaks
text = "line1\nline2\nline3"
print(text.splitlines())  # ['line1', 'line2', 'line3']

# join() — list → string (opposite of split)
words  = ["Python", "is", "awesome"]
result = " ".join(words)
print(result)   # Python is awesome

# Join with different separators
print(", ".join(["Alice", "Bob", "Charlie"]))
# Alice, Bob, Charlie

print("-".join(["2024", "01", "15"]))
# 2024-01-15

# Efficient string building
parts = [str(i) for i in range(5)]
print("".join(parts))   # 01234

# Content checking — all return True or False
print("hello".isalpha())     # True  — only letters
print("hello123".isalpha())  # False
print("12345".isdigit())     # True  — only digits
print("12.5".isdigit())      # False (dot not a digit)
print("abc123".isalnum())    # True  — letters or digits
print("   ".isspace())       # True  — only whitespace
print("Hello World".istitle()) # True

# Practical: validate user input
def validate_username(username):
    if not username:
        return "Username cannot be empty"
    if not username[0].isalpha():
        return "Must start with a letter"
    if not username.isalnum():
        return "Only letters and numbers allowed"
    if len(username) < 3:
        return "Too short (min 3 chars)"
    return "Valid!"

print(validate_username("Alice123"))  # Valid!
print(validate_username("123abc"))    # Must start with a letter

# ljust, rjust, center
name = "Alice"
print(name.ljust(10))         # "Alice     "
print(name.rjust(10))         # "     Alice"
print(name.center(10))        # "  Alice   "
print(name.center(10, "*"))   # **Alice***

# zfill — pad with zeros
print("42".zfill(6))          # 000042
print("3.14".zfill(8))        # 00003.14

# Building a table
headers = ["Name", "Age", "City"]
rows = [
    ("Alice",   25, "London"),
    ("Bob",     30, "Paris"),
    ("Charlie", 22, "Tokyo")
]

print(f"{'Name':<10} {'Age':>5} {'City':<10}")
print("-" * 27)
for name, age, city in rows:
    print(f"{name:<10} {age:>5} {city:<10}")

# Reverse a string
text = "Python"
print(text[::-1])   # nohtyP

# Check palindrome
def is_palindrome(s):
    s = s.lower().replace(" ", "")
    return s == s[::-1]

print(is_palindrome("racecar"))    # True
print(is_palindrome("A man a plan a canal Panama"))  # True
print(is_palindrome("hello"))      # False

# Count vowels
def count_vowels(text):
    return sum(1 for c in text.lower() if c in "aeiou")

print(count_vowels("Hello World"))  # 3

# Title case with exceptions
def smart_title(text):
    small = {"a", "an", "the", "and", "or", "but", "in", "on"}
    words = text.lower().split()
    return " ".join(
        w if (w in small and i != 0) else w.capitalize()
        for i, w in enumerate(words)
    )

print(smart_title("the lord of the rings"))
# The Lord of the Rings

# Since strings are immutable, convert to list to "modify"
text = "Hello"
chars = list(text)         # ['H', 'e', 'l', 'l', 'o']
chars[0] = "J"
result = "".join(chars)    # "Jello"
print(result)

# Remove duplicates while preserving order
def remove_duplicate_chars(s):
    seen = set()
    return "".join(c for c in s if not (c in seen or seen.add(c)))

print(remove_duplicate_chars("programming"))  # progamin

# Caesar cipher
def caesar(text, shift):
    result = []
    for char in text:
        if char.isalpha():
            base = ord('A') if char.isupper() else ord('a')
            result.append(chr((ord(char) - base + shift) % 26 + base))
        else:
            result.append(char)
    return "".join(result)

print(caesar("Hello, World!", 3))  # Khoor, Zruog!

# Syntax: open(filename, mode, encoding)
# Modes:
#   "r"  — read (default)       file must exist
#   "w"  — write                creates/overwrites file
#   "a"  — append               creates/adds to end
#   "x"  — exclusive create     fails if file exists
#   "r+" — read and write
#   "b"  — binary mode (add to above: "rb", "wb")

# ✅ ALWAYS use 'with' — auto-closes the file
with open("hello.txt", "w", encoding="utf-8") as f:
    f.write("Hello, World!\n")
# File is automatically closed here

# ❌ Manual open/close — error-prone
f = open("hello.txt", "r")
content = f.read()
f.close()   # easy to forget!

# Assume "data.txt" contains:
# Line 1: Hello
# Line 2: World
# Line 3: Python

# read() — entire file as one string
with open("data.txt", "r", encoding="utf-8") as f:
    content = f.read()
print(content)

# readline() — one line at a time
with open("data.txt", "r", encoding="utf-8") as f:
    first_line  = f.readline()   # "Hello\n"
    second_line = f.readline()   # "World\n"
print(first_line.strip())   # Hello

# readlines() — all lines as a list
with open("data.txt", "r", encoding="utf-8") as f:
    lines = f.readlines()
print(lines)  # ['Hello\n', 'World\n', 'Python\n']

# ✅ Best practice — iterate line by line (memory efficient)
with open("data.txt", "r", encoding="utf-8") as f:
    for line in f:
        print(line.strip())

def read_file(filename):
    """Safely read a file, return None if not found."""
    try:
        with open(filename, "r", encoding="utf-8") as f:
            return f.read()
    except FileNotFoundError:
        print(f"Error: '{filename}' not found.")
        return None
    except PermissionError:
        print(f"Error: No permission to read '{filename}'.")
        return None

content = read_file("notes.txt")
if content:
    print(content)

# write() — creates or OVERWRITES the file
with open("output.txt", "w", encoding="utf-8") as f:
    f.write("First line\n")
    f.write("Second line\n")
    f.write("Third line\n")

# writelines() — write a list of strings
lines = ["Apple\n", "Banana\n", "Cherry\n"]
with open("fruits.txt", "w", encoding="utf-8") as f:
    f.writelines(lines)

# append() — adds to end WITHOUT overwriting
with open("output.txt", "a", encoding="utf-8") as f:
    f.write("Fourth line (appended)\n")

# Write multiple lines with join
data = ["Alice", "Bob", "Charlie"]
with open("names.txt", "w", encoding="utf-8") as f:
    f.write("\n".join(data))

# Exclusive create — fails if file already exists
try:
    with open("new_file.txt", "x", encoding="utf-8") as f:
        f.write("Brand new file!")
except FileExistsError:
    print("File already exists!")

with open("data.txt", "r", encoding="utf-8") as f:
    # tell() — current position in bytes
    print(f.tell())         # 0 (start)

    first = f.read(5)       # read 5 characters
    print(first)            # Hello
    print(f.tell())         # 5

    # seek(position) — move to position
    f.seek(0)               # back to start
    content = f.read()
    print(content)

    # seek(0, 2) — move to end of file
    f.seek(0, 2)
    print(f"File size: {f.tell()} bytes")

import csv

# Write CSV with csv.writer
students = [
    ["Name",    "Age", "Grade"],
    ["Alice",   25,    "A"],
    ["Bob",     22,    "B"],
    ["Charlie", 28,    "A+"],
]

with open("students.csv", "w", newline="", encoding="utf-8") as f:
    writer = csv.writer(f)
    writer.writerows(students)

# Write CSV with csv.DictWriter (recommended)
fieldnames = ["name", "age", "city"]
people = [
    {"name": "Alice",   "age": 25, "city": "London"},
    {"name": "Bob",     "age": 30, "city": "Paris"},
    {"name": "Charlie", "age": 22, "city": "Tokyo"},
]

with open("people.csv", "w", newline="", encoding="utf-8") as f:
    writer = csv.DictWriter(f, fieldnames=fieldnames)
    writer.writeheader()      # writes the header row
    writer.writerows(people)

import csv

# Read with csv.reader
with open("students.csv", "r", encoding="utf-8") as f:
    reader = csv.reader(f)
    header = next(reader)    # skip header row
    print("Header:", header)
    for row in reader:
        print(row)

# Read with csv.DictReader (recommended — named fields)
with open("people.csv", "r", encoding="utf-8") as f:
    reader = csv.DictReader(f)
    for row in reader:
        print(f"{row['name']} lives in {row['city']}")

# Read all rows into a list of dicts
with open("people.csv", "r", encoding="utf-8") as f:
    people = list(csv.DictReader(f))

# Filter: people over 25
seniors = [p for p in people if int(p["age"]) > 25]
print(seniors)

import json

data = {
    "name":    "Alice",
    "age":     25,
    "skills":  ["Python", "SQL", "Git"],
    "address": {
        "city":    "London",
        "country": "UK"
    },
    "active": True
}

# json.dumps() — Python object → JSON string
json_str = json.dumps(data)
print(json_str)

# Pretty-print with indent
pretty = json.dumps(data, indent=4, sort_keys=True)
print(pretty)

# json.dump() — write directly to file
with open("user.json", "w", encoding="utf-8") as f:
    json.dump(data, f, indent=4)

print("JSON file saved!")

import json

# json.loads() — JSON string → Python object
json_str = '{"name": "Alice", "age": 25, "active": true}'
data = json.loads(json_str)
print(data["name"])    # Alice
print(type(data))      # <class 'dict'>

# json.load() — read directly from file
with open("user.json", "r", encoding="utf-8") as f:
    user = json.load(f)

print(user["name"])           # Alice
print(user["skills"])         # ['Python', 'SQL', 'Git']
print(user["address"]["city"]) # London

# JSON type mapping:
# JSON true/false  → Python True/False
# JSON null        → Python None
# JSON array []    → Python list
# JSON object {}   → Python dict

import json
import os

CONFIG_FILE = "config.json"

def load_config():
    """Load config, return defaults if file missing."""
    defaults = {
        "theme":    "light",
        "language": "en",
        "font_size": 14,
        "autosave":  True
    }
    if not os.path.exists(CONFIG_FILE):
        return defaults
    with open(CONFIG_FILE, "r", encoding="utf-8") as f:
        return json.load(f)

def save_config(config):
    """Save config to JSON file."""
    with open(CONFIG_FILE, "w", encoding="utf-8") as f:
        json.dump(config, f, indent=4)

# Load, modify, save
config = load_config()
config["theme"] = "dark"
config["font_size"] = 16
save_config(config)
print("Config saved:", config)

import os

# Current directory
print(os.getcwd())              # /home/user/project

# Change directory
# os.chdir("/home/user")

# List files in directory
files = os.listdir(".")
print(files)

# Check existence
print(os.path.exists("data.txt"))   # True/False
print(os.path.isfile("data.txt"))   # True if file
print(os.path.isdir("myFolder"))    # True if directory

# File info
print(os.path.getsize("data.txt"))  # size in bytes
print(os.path.basename("/home/user/data.txt"))  # data.txt
print(os.path.dirname("/home/user/data.txt"))   # /home/user
print(os.path.splitext("data.txt"))             # ('data', '.txt')

# Join paths safely (handles OS differences)
path = os.path.join("home", "user", "data.txt")
print(path)   # home/user/data.txt (Linux) or home\user\data.txt (Windows)

# Create / remove directories
os.makedirs("output/reports", exist_ok=True)  # create nested dirs
# os.rmdir("empty_folder")    # remove empty dir
# os.remove("file.txt")       # delete a file
# os.rename("old.txt", "new.txt")  # rename

from pathlib import Path

# Create Path objects
p = Path(".")                  # current directory
home = Path.home()             # home directory
file = Path("data/output.txt")

# Path operations with / operator
config = home / "projects" / "app" / "config.json"
print(config)

# File info
print(file.name)        # output.txt
print(file.stem)        # output
print(file.suffix)      # .txt
print(file.parent)      # data

# Check and create
print(file.exists())
print(file.is_file())
file.parent.mkdir(parents=True, exist_ok=True)

# Read and write (no open() needed!)
Path("hello.txt").write_text("Hello, World!", encoding="utf-8")
content = Path("hello.txt").read_text(encoding="utf-8")
print(content)

# List files
for f in Path(".").iterdir():
    print(f)

# Glob — find files by pattern
for txt_file in Path(".").glob("*.txt"):
    print(txt_file)

# Recursive glob
for py_file in Path(".").rglob("*.py"):
    print(py_file)

# Write binary data
data = bytes([72, 101, 108, 108, 111])  # "Hello" in ASCII
with open("binary.bin", "wb") as f:
    f.write(data)

# Read binary data
with open("binary.bin", "rb") as f:
    content = f.read()
print(content)          # b'Hello'
print(list(content))    # [72, 101, 108, 108, 111]

# Copy a file (binary-safe)
def copy_file(src, dst):
    with open(src, "rb") as f_in:
        with open(dst, "wb") as f_out:
            f_out.write(f_in.read())
    print(f"Copied '{src}' to '{dst}'")

copy_file("photo.jpg", "photo_backup.jpg")

# SYNTAX ERROR — detected before running
# if x = 5:       ← SyntaxError: invalid syntax
# print "hello"   ← SyntaxError (Python 2 style)

# EXCEPTIONS — occur at runtime
# These are valid syntax but fail when executed:

# ZeroDivisionError
result = 10 / 0

# NameError
print(undefined_variable)

# TypeError
result = "hello" + 5

# ValueError
number = int("abc")

# IndexError
lst = [1, 2, 3]
print(lst[10])

# KeyError
d = {"a": 1}
print(d["z"])

# AttributeError
"hello".nonexistent_method()

# FileNotFoundError
open("missing.txt")

# Basic structure
try:
    result = 10 / 0
except ZeroDivisionError:
    print("Cannot divide by zero!")

# Catching the exception object
try:
    number = int("abc")
except ValueError as e:
    print(f"ValueError caught: {e}")
# ValueError caught: invalid literal for int() with base 10: 'abc'

# Bare except — catches ALL exceptions (not recommended)
try:
    risky_code = 1 / 0
except:
    print("Something went wrong")   # ← too broad, hides bugs!

# Better: catch Exception base class
try:
    risky_code = 1 / 0
except Exception as e:
    print(f"Error: {type(e).__name__}: {e}")

def safe_divide(a, b):
    try:
        return a / b
    except ZeroDivisionError:
        print("Error: Cannot divide by zero!")
        return None
    except TypeError:
        print("Error: Both arguments must be numbers!")
        return None

print(safe_divide(10, 2))    # 5.0
print(safe_divide(10, 0))    # Error: Cannot divide by zero!
print(safe_divide(10, "x"))  # Error: Both arguments must be numbers!

# Catch multiple in one line (tuple)
try:
    value = int(input("Enter a number: "))
    result = 100 / value
except (ValueError, ZeroDivisionError) as e:
    print(f"Input error: {e}")

try:
    number = int(input("Enter a number: "))
    result = 100 / number
except ValueError:
    print("That's not a valid number!")
except ZeroDivisionError:
    print("Cannot divide by zero!")
else:
    # Runs ONLY if no exception occurred
    print(f"Result: {result}")
finally:
    # ALWAYS runs — exception or not
    print("Calculation attempt complete.")

# Real-world: file handling with finally
f = None
try:
    f = open("data.txt", "r")
    content = f.read()
except FileNotFoundError:
    print("File not found!")
finally:
    if f:
        f.close()   # always close the file
        print("File closed.")

def set_age(age):
    if not isinstance(age, int):
        raise TypeError(f"Age must be an integer, got {type(age).__name__}")
    if age < 0:
        raise ValueError(f"Age cannot be negative: {age}")
    if age > 150:
        raise ValueError(f"Age {age} is unrealistically high")
    return age

try:
    set_age(-5)
except ValueError as e:
    print(f"ValueError: {e}")

try:
    set_age("twenty")
except TypeError as e:
    print(f"TypeError: {e}")

# Re-raise an exception
def process_data(data):
    try:
        return int(data)
    except ValueError:
        print("Logging: invalid data received")
        raise   # re-raises the original exception

def load_config(filename):
    try:
        with open(filename) as f:
            import json
            return json.load(f)
    except FileNotFoundError as e:
        raise RuntimeError(f"Config file missing: {filename}") from e
    except json.JSONDecodeError as e:
        raise RuntimeError(f"Invalid JSON in config: {filename}") from e

try:
    config = load_config("settings.json")
except RuntimeError as e:
    print(f"Config error: {e}")
    print(f"Caused by: {e.__cause__}")

# Basic custom exception
class AppError(Exception):
    """Base exception for this application."""
    pass

class ValidationError(AppError):
    """Raised when input validation fails."""
    pass

class DatabaseError(AppError):
    """Raised when a database operation fails."""
    pass

# Custom exception with extra data
class InsufficientFundsError(Exception):
    """Raised when a bank account has insufficient funds."""
    def __init__(self, balance, amount):
        self.balance = balance
        self.amount  = amount
        self.deficit = amount - balance
        super().__init__(
            f"Insufficient funds: need ${amount:.2f}, "
            f"have ${balance:.2f} (short ${self.deficit:.2f})"
        )

# Using custom exceptions
class BankAccount:
    def __init__(self, balance=0):
        self.balance = balance

    def withdraw(self, amount):
        if amount <= 0:
            raise ValidationError("Withdrawal amount must be positive")
        if amount > self.balance:
            raise InsufficientFundsError(self.balance, amount)
        self.balance -= amount
        return self.balance

account = BankAccount(100)
try:
    account.withdraw(150)
except InsufficientFundsError as e:
    print(e)
    print(f"Deficit: ${e.deficit:.2f}")

# BaseException
#   ├── SystemExit
#   ├── KeyboardInterrupt
#   └── Exception
#         ├── ArithmeticError
#         │     ├── ZeroDivisionError
#         │     └── OverflowError
#         ├── LookupError
#         │     ├── IndexError
#         │     └── KeyError
#         ├── TypeError
#         ├── ValueError
#         │     └── UnicodeError
#         ├── OSError
#         │     ├── FileNotFoundError
#         │     ├── PermissionError
#         │     └── IsADirectoryError
#         ├── AttributeError
#         ├── NameError
#         ├── RuntimeError
#         └── StopIteration

# ✅ Catch specific BEFORE general
try:
    result = int("abc") / 0
except ValueError:
    print("Value error")
except ZeroDivisionError:
    print("Division by zero")
except Exception as e:
    print(f"Unexpected: {e}")   # catch-all fallback

import time

def retry(func, max_attempts=3, delay=1):
    """Retry a function up to max_attempts times."""
    for attempt in range(1, max_attempts + 1):
        try:
            return func()
        except Exception as e:
            print(f"Attempt {attempt} failed: {e}")
            if attempt < max_attempts:
                time.sleep(delay)
            else:
                raise RuntimeError(f"All {max_attempts} attempts failed") from e

# Usage
import random
def unreliable_operation():
    if random.random() < 0.7:   # 70% chance of failure
        raise ConnectionError("Network timeout")
    return "Success!"

try:
    result = retry(unreliable_operation, max_attempts=3)
    print(result)
except RuntimeError as e:
    print(f"Operation failed: {e}")

def safe_int(value, default=0):
    """Convert to int safely, return default on failure."""
    try:
        return int(value)
    except (ValueError, TypeError):
        return default

def safe_float(value, default=0.0):
    try:
        return float(value)
    except (ValueError, TypeError):
        return default

def safe_get(dictionary, key, default=None):
    """Safely get a value from a dict."""
    try:
        return dictionary[key]
    except (KeyError, TypeError):
        return default

print(safe_int("42"))       # 42
print(safe_int("abc"))      # 0
print(safe_int("abc", -1))  # -1
print(safe_float("3.14"))   # 3.14
print(safe_get({"a": 1}, "b", "N/A"))  # N/A

def get_positive_integer(prompt):
    """Keep asking until user enters a valid positive integer."""
    while True:
        try:
            value = int(input(prompt))
            if value <= 0:
                raise ValueError("Must be a positive number")
            return value
        except ValueError as e:
            print(f"Invalid input: {e}. Please try again.")

def get_choice(prompt, choices):
    """Keep asking until user picks a valid choice."""
    choices_str = "/".join(choices)
    while True:
        answer = input(f"{prompt} ({choices_str}): ").strip().lower()
        if answer in choices:
            return answer
        print(f"Please enter one of: {choices_str}")

# age = get_positive_integer("Enter your age: ")
# choice = get_choice("Continue?", ["yes", "no"])

import warnings

def old_function(x):
    warnings.warn(
        "old_function() is deprecated, use new_function() instead.",
        DeprecationWarning,
        stacklevel=2
    )
    return x * 2

def new_function(x):
    return x * 2

# Show deprecation warnings
warnings.simplefilter("always")
result = old_function(5)

# Warning types:
# DeprecationWarning  — for deprecated features
# UserWarning         — general user warnings
# RuntimeWarning      — runtime issues
# ResourceWarning     — resource usage issues

# 1. Import the whole module
import math
print(math.pi)          # 3.141592653589793
print(math.sqrt(16))    # 4.0

# 2. Import specific names
from math import pi, sqrt, floor
print(pi)               # 3.141592653589793
print(sqrt(25))         # 5.0

# 3. Import with alias (rename)
import numpy as np           # common convention
import pandas as pd
import matplotlib.pyplot as plt

import math as m
print(m.factorial(5))   # 120

# 4. Import everything (avoid in production!)
from math import *
print(sin(pi / 2))      # 1.0  ← pollutes namespace!

# 5. Import from sub-module
from os.path import join, exists
from collections import defaultdict, Counter

# File: mathutils.py
"""Utility functions for mathematical operations."""

PI = 3.14159265358979

def circle_area(radius):
    """Return the area of a circle."""
    return PI * radius ** 2

def circle_circumference(radius):
    """Return the circumference of a circle."""
    return 2 * PI * radius

def is_prime(n):
    """Return True if n is a prime number."""
    if n < 2:
        return False
    for i in range(2, int(n ** 0.5) + 1):
        if n % i == 0:
            return False
    return True

def factorial(n):
    """Return n! recursively."""
    return 1 if n <= 1 else n * factorial(n - 1)

# File: main.py  (same directory as mathutils.py)
import mathutils

print(mathutils.circle_area(5))          # 78.539...
print(mathutils.is_prime(17))            # True
print(mathutils.factorial(6))            # 720

# Or import specific names
from mathutils import circle_area, is_prime

print(circle_area(3))   # 28.274...
print(is_prime(29))     # True

# Check what's inside a module
print(dir(mathutils))
print(mathutils.__doc__)   # module docstring

# File: mathutils.py

def circle_area(radius):
    return 3.14159 * radius ** 2

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

# This block ONLY runs when file is executed directly
# It does NOT run when the file is imported
if __name__ == "__main__":
    print("Running mathutils.py directly...")
    print(f"Area of circle r=5: {circle_area(5):.2f}")
    print(f"Is 17 prime? {is_prime(17)}")

# When imported: __name__ == "mathutils"
# When run directly: __name__ == "__main__"

# Project structure:
# myapp/
# ├── __init__.py          ← makes it a package
# ├── main.py
# ├── utils/
# │   ├── __init__.py
# │   ├── mathutils.py
# │   └── stringutils.py
# └── models/
#     ├── __init__.py
#     └── user.py

# File: myapp/utils/__init__.py
# (can be empty, or expose public API)
from .mathutils  import circle_area, is_prime
from .stringutils import slugify

# File: myapp/main.py — importing from package
from myapp.utils import circle_area
from myapp.utils.mathutils import factorial
from myapp.models.user import User

# Relative imports (inside the package)
from .utils import circle_area    # same package
from ..models import User         # parent package

import os

# Environment variables
home    = os.environ.get("HOME", "/home/user")
path    = os.environ.get("PATH")
os.environ["MY_VAR"] = "hello"

# Current process
print(os.getpid())        # process ID
print(os.getcwd())        # current working directory
os.chdir("/tmp")          # change directory

# File and directory operations
os.makedirs("a/b/c", exist_ok=True)
os.rename("old.txt", "new.txt")
os.remove("file.txt")
os.rmdir("empty_dir")

# Path utilities
print(os.path.join("home", "user", "file.txt"))
print(os.path.exists("data.txt"))
print(os.path.getsize("data.txt"))   # bytes
print(os.path.abspath("data.txt"))   # absolute path

# Walk directory tree
for root, dirs, files in os.walk("."):
    for file in files:
        full_path = os.path.join(root, file)
        print(full_path)

# Run shell command
exit_code = os.system("echo Hello from shell")
print(f"Exit code: {exit_code}")

import sys

# Python version info
print(sys.version)          # 3.11.2 (main, ...)
print(sys.version_info)     # sys.version_info(major=3, minor=11, ...)
print(sys.version_info.major)  # 3

# Platform
print(sys.platform)         # 'linux', 'win32', 'darwin'

# Command-line arguments
# python script.py arg1 arg2
print(sys.argv)             # ['script.py', 'arg1', 'arg2']
print(sys.argv[0])          # script name
print(sys.argv[1:])         # all arguments

# Module search path
print(sys.path)             # list of directories Python searches

# Add custom path
sys.path.insert(0, "/my/custom/modules")

# Standard streams
sys.stdout.write("Hello\n")   # same as print()
sys.stderr.write("Error!\n")  # write to stderr

# Exit the program
# sys.exit(0)    # 0 = success
# sys.exit(1)    # non-zero = error

# Memory size of an object
print(sys.getsizeof([1, 2, 3]))   # bytes

import math

# Constants
print(math.pi)      # 3.141592653589793
print(math.e)       # 2.718281828459045
print(math.tau)     # 6.283185307179586  (2π)
print(math.inf)     # infinity
print(math.nan)     # Not a Number

# Rounding
print(math.floor(3.7))   # 3  (round down)
print(math.ceil(3.2))    # 4  (round up)
print(math.trunc(3.9))   # 3  (truncate decimal)

# Powers and logarithms
print(math.sqrt(144))    # 12.0
print(math.pow(2, 10))   # 1024.0
print(math.log(math.e))  # 1.0  (natural log)
print(math.log(100, 10)) # 2.0  (log base 10)
print(math.log2(8))      # 3.0
print(math.log10(1000))  # 3.0

# Trigonometry (radians)
print(math.sin(math.pi / 2))   # 1.0
print(math.cos(0))             # 1.0
print(math.degrees(math.pi))   # 180.0
print(math.radians(180))       # 3.14159...

# Other
print(math.factorial(10))  # 3628800
print(math.gcd(48, 18))    # 6
print(math.isnan(math.nan))   # True
print(math.isinf(math.inf))   # True

import random

# Seed for reproducibility
random.seed(42)

# Random floats
print(random.random())          # 0.0 to 1.0
print(random.uniform(1.5, 9.5)) # float between 1.5 and 9.5

# Random integers
print(random.randint(1, 10))    # 1 to 10 inclusive
print(random.randrange(0, 100, 5))  # 0,5,10,...,95

# Random choices from sequences
colors = ["red", "green", "blue", "yellow"]
print(random.choice(colors))    # one random item
print(random.choices(colors, k=3))  # 3 with replacement
print(random.sample(colors, k=3))   # 3 without replacement

# Shuffle a list in-place
deck = list(range(1, 14))
random.shuffle(deck)
print(deck)

# Weighted choices
outcomes = ["win", "lose", "draw"]
weights  = [0.3, 0.5, 0.2]
result   = random.choices(outcomes, weights=weights, k=1)[0]
print(result)

# Cryptographically secure random (for passwords/tokens)
import secrets
print(secrets.token_hex(16))    # 32-char hex token
print(secrets.randbelow(100))   # secure random int

from datetime import datetime, date, time, timedelta

# Current date and time
now   = datetime.now()
today = date.today()
print(now)     # 2024-01-15 14:30:25.123456
print(today)   # 2024-01-15

# Create specific dates
birthday  = date(1990, 6, 15)
meeting   = datetime(2024, 3, 20, 14, 30, 0)

# Access components
print(now.year, now.month, now.day)
print(now.hour, now.minute, now.second)
print(now.weekday())    # 0=Monday, 6=Sunday
print(now.strftime("%A"))  # "Monday"

# Formatting (strftime)
print(now.strftime("%Y-%m-%d"))          # 2024-01-15
print(now.strftime("%d/%m/%Y %H:%M"))    # 15/01/2024 14:30
print(now.strftime("%B %d, %Y"))         # January 15, 2024

# Parsing strings (strptime)
dt = datetime.strptime("2024-03-15", "%Y-%m-%d")
print(dt)

# Arithmetic with timedelta
one_week   = timedelta(weeks=1)
next_week  = today + one_week
yesterday  = today - timedelta(days=1)

# Difference between dates
delta = date(2025, 1, 1) - today
print(f"Days until 2025: {delta.days}")

# Age calculator
def calculate_age(birthdate):
    today = date.today()
    age   = today.year - birthdate.year
    if (today.month, today.day) < (birthdate.month, birthdate.day):
        age -= 1
    return age

print(calculate_age(date(1990, 6, 15)))

from collections import Counter, defaultdict, OrderedDict, deque, namedtuple

# Counter — count occurrences
words   = ["apple", "banana", "apple", "cherry", "banana", "apple"]
counter = Counter(words)
print(counter)                    # Counter({'apple': 3, 'banana': 2, ...})
print(counter.most_common(2))     # [('apple', 3), ('banana', 2)]
print(counter["apple"])           # 3

# Counter on strings
letter_count = Counter("mississippi")
print(letter_count)   # Counter({'s': 4, 'i': 4, 'p': 2, 'm': 1})

# defaultdict — auto-creates missing keys
groups = defaultdict(list)
for name, dept in [("Alice","Eng"),("Bob","HR"),("Carol","Eng")]:
    groups[dept].append(name)
print(dict(groups))   # {'Eng': ['Alice', 'Carol'], 'HR': ['Bob']}

word_count = defaultdict(int)
for word in ["hi", "hi", "hello"]:
    word_count[word] += 1   # no KeyError!

# deque — fast append/pop from both ends
dq = deque([1, 2, 3])
dq.appendleft(0)    # [0, 1, 2, 3]
dq.append(4)        # [0, 1, 2, 3, 4]
dq.popleft()        # removes 0
dq.rotate(1)        # rotate right by 1

# namedtuple — tuple with named fields
Point = namedtuple("Point", ["x", "y"])
p = Point(3, 4)
print(p.x, p.y)     # 3 4
print(p[0], p[1])   # 3 4  (still a tuple)

Person = namedtuple("Person", ["name", "age", "city"])
alice  = Person("Alice", 25, "London")
print(alice.name)   # Alice

import itertools

# chain — combine multiple iterables
result = list(itertools.chain([1,2], [3,4], [5,6]))
print(result)   # [1, 2, 3, 4, 5, 6]

# combinations — all unique combos (no repeat)
cards = list(itertools.combinations(["A","B","C","D"], 2))
print(cards)    # [('A','B'), ('A','C'), ('A','D'), ('B','C'), ...]

# permutations — all ordered arrangements
perms = list(itertools.permutations([1, 2, 3]))
print(len(perms))  # 6

# product — cartesian product
grid = list(itertools.product([0,1], repeat=3))
print(grid)   # [(0,0,0),(0,0,1),(0,1,0),...,(1,1,1)]

# groupby — group consecutive elements
data = [("Alice","Eng"),("Bob","Eng"),("Carol","HR"),("Dave","HR")]
for dept, members in itertools.groupby(data, key=lambda x: x[1]):
    print(dept, list(members))

# islice — slice an iterator
result = list(itertools.islice(range(100), 5, 15, 2))
print(result)   # [5, 7, 9, 11, 13]

# count, cycle, repeat — infinite iterators
counter = itertools.count(start=10, step=5)
print([next(counter) for _ in range(4)])  # [10, 15, 20, 25]

cycler  = itertools.cycle(["A", "B", "C"])
print([next(cycler) for _ in range(7)])   # ['A','B','C','A','B','C','A']

# Run these commands in your terminal (not Python!)

# Install a package
pip install requests
pip install pandas numpy matplotlib

# Install specific version
pip install requests==2.31.0
pip install "requests>=2.28,<3.0"

# Upgrade a package
pip install --upgrade requests

# Uninstall
pip uninstall requests

# List installed packages
pip list
pip show requests       # details about one package

# Freeze — save all dependencies to file
pip freeze > requirements.txt

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

# Virtual environments (isolate project dependencies)
python -m venv venv          # create virtual env
source venv/bin/activate     # activate (Linux/Mac)
venv\Scripts\activate        # activate (Windows)
deactivate                   # deactivate

# Popular packages:
# requests      — HTTP requests
# pandas        — data analysis
# numpy         — numerical computing
# matplotlib    — data visualization
# flask         — web framework
# django        — full-stack web framework
# sqlalchemy    — database ORM
# pytest        — testing framework
# pillow        — image processing
# beautifulsoup4 — web scraping

import sys

# Python searches for modules in this order:
# 1. Built-in modules (math, os, sys, etc.)
# 2. Current directory
# 3. PYTHONPATH environment variable directories
# 4. Standard library directories
# 5. Site-packages (pip-installed packages)

# View the search path
for path in sys.path:
    print(path)

# Add a custom directory at runtime
sys.path.insert(0, "/path/to/my/modules")

# Check if a module is built-in
import sys
print("math" in sys.builtin_module_names)   # True
print("requests" in sys.builtin_module_names) # False

# Reload a module (useful during development)
import importlib
import mymodule
importlib.reload(mymodule)   # re-reads the file

# Define a class
class Dog:
    """A simple Dog class."""

    # Class attribute — shared by ALL instances
    species = "Canis familiaris"

    # Constructor — called when creating an instance
    def __init__(self, name, breed, age):
        # Instance attributes — unique to each object
        self.name  = name
        self.breed = breed
        self.age   = age

    # Instance method
    def bark(self):
        return f"{self.name} says: Woof!"

    def describe(self):
        return f"{self.name} is a {self.age}-year-old {self.breed}."

# Create instances (objects)
rex   = Dog("Rex",   "German Shepherd", 3)
buddy = Dog("Buddy", "Golden Retriever", 5)

# Access attributes
print(rex.name)       # Rex
print(buddy.breed)    # Golden Retriever
print(rex.species)    # Canis familiaris (class attribute)

# Call methods
print(rex.bark())     # Rex says: Woof!
print(buddy.describe()) # Buddy is a 5-year-old Golden Retriever.

class Person:
    """Represents a person."""

    def __init__(self, name, age, email=None):
        # Validate in __init__
        if not isinstance(name, str) or not name.strip():
            raise ValueError("Name must be a non-empty string")
        if not isinstance(age, int) or age < 0:
            raise ValueError("Age must be a non-negative integer")

        self.name  = name.strip().title()
        self.age   = age
        self.email = email
        self._id   = id(self)   # unique identifier

    def greet(self):
        return f"Hi, I'm {self.name} and I'm {self.age} years old."

    def have_birthday(self):
        self.age += 1
        return f"Happy Birthday {self.name}! Now {self.age}."

# Create objects
alice = Person("alice", 25, "[email protected]")
bob   = Person("Bob",   30)

print(alice.greet())
print(bob.greet())
print(alice.have_birthday())

# Each object is independent
print(alice.age)   # 26 (after birthday)
print(bob.age)     # 30 (unchanged)

class Employee:
    # Class attribute — shared by ALL instances
    company    = "TechCorp"
    employee_count = 0

    def __init__(self, name, salary):
        # Instance attributes — unique per object
        self.name   = name
        self.salary = salary
        Employee.employee_count += 1   # update class attribute

    def get_info(self):
        return f"{self.name} at {Employee.company}: ${self.salary:,}"

    # Class method — works with the class, not instance
    @classmethod
    def get_count(cls):
        return f"Total employees: {cls.employee_count}"

    @classmethod
    def from_string(cls, data_string):
        """Alternative constructor: 'Name:Salary'"""
        name, salary = data_string.split(":")
        return cls(name.strip(), int(salary.strip()))

    # Static method — no access to class or instance
    @staticmethod
    def is_valid_salary(salary):
        return isinstance(salary, (int, float)) and salary > 0

# Usage
e1 = Employee("Alice", 75000)
e2 = Employee("Bob",   85000)
e3 = Employee.from_string("Charlie: 90000")

print(e1.get_info())
print(Employee.get_count())          # Total employees: 3
print(Employee.is_valid_salary(-500)) # False
print(e1.company)                    # TechCorp (class attr)
print(Employee.company)              # TechCorp

class ShoppingCart:
    """A shopping cart for an e-commerce app."""

    def __init__(self, owner):
        self.owner = owner
        self.items = []        # list of (name, price, qty) tuples

    def add_item(self, name, price, qty=1):
        """Add an item to the cart."""
        if price < 0 or qty < 1:
            raise ValueError("Invalid price or quantity")
        self.items.append({"name": name, "price": price, "qty": qty})
        print(f"Added: {qty}x {name} @ ${price:.2f}")

    def remove_item(self, name):
        """Remove an item by name."""
        self.items = [i for i in self.items if i["name"] != name]

    def get_total(self):
        """Calculate total price."""
        return sum(i["price"] * i["qty"] for i in self.items)

    def apply_discount(self, percent):
        """Apply a percentage discount."""
        discount = self.get_total() * (percent / 100)
        return self.get_total() - discount

    def print_receipt(self):
        """Print a formatted receipt."""
        print(f"\n{'='*35}")
        print(f"  Cart for: {self.owner}")
        print(f"{'='*35}")
        for item in self.items:
            line = item["price"] * item["qty"]
            print(f"  {item['name']:<18} ${line:>6.2f}")
        print(f"{'─'*35}")
        print(f"  {'TOTAL':<18} ${self.get_total():>6.2f}")
        print(f"{'='*35}")

    def is_empty(self):
        return len(self.items) == 0

cart = ShoppingCart("Alice")
cart.add_item("Python Book", 29.99)
cart.add_item("USB Hub",     19.99, qty=2)
cart.add_item("Coffee",       8.50, qty=3)
cart.print_receipt()

class BankAccount:
    """Demonstrates encapsulation."""

    def __init__(self, owner, balance=0):
        self.owner     = owner        # public   — accessible anywhere
        self._balance  = balance      # protected — convention: "internal use"
        self.__pin     = "0000"       # private  — name-mangled by Python

    # Public interface to access protected data
    def deposit(self, amount):
        if amount > 0:
            self._balance += amount
            return f"Deposited ${amount:.2f}. Balance: ${self._balance:.2f}"
        raise ValueError("Deposit must be positive")

    def withdraw(self, amount):
        if amount > self._balance:
            raise ValueError("Insufficient funds")
        self._balance -= amount
        return f"Withdrew ${amount:.2f}. Balance: ${self._balance:.2f}"

    def get_balance(self):
        return self._balance

    def change_pin(self, old_pin, new_pin):
        if old_pin != self.__pin:
            raise ValueError("Incorrect PIN")
        if len(new_pin) != 4 or not new_pin.isdigit():
            raise ValueError("PIN must be 4 digits")
        self.__pin = new_pin
        return "PIN changed successfully"

acc = BankAccount("Alice", 1000)
print(acc.deposit(500))
print(acc.withdraw(200))
print(acc.get_balance())   # 1300

# Public: accessible
print(acc.owner)           # Alice

# Protected: accessible but "please don't"
print(acc._balance)        # 1300 (works but bad practice)

# Private: name-mangled — hard to access accidentally
# print(acc.__pin)         # AttributeError!
print(acc._BankAccount__pin)  # "0000" (mangled name — avoid this!)

class Temperature:
    """Temperature with Celsius/Fahrenheit conversion."""

    def __init__(self, celsius=0):
        self.celsius = celsius   # uses the setter below

    @property
    def celsius(self):
        """Get temperature in Celsius."""
        return self._celsius

    @celsius.setter
    def celsius(self, value):
        """Set temperature — validates input."""
        if value < -273.15:
            raise ValueError(f"Temperature below absolute zero: {value}")
        self._celsius = value

    @property
    def fahrenheit(self):
        """Get temperature in Fahrenheit (computed)."""
        return self._celsius * 9/5 + 32

    @fahrenheit.setter
    def fahrenheit(self, value):
        """Set temperature via Fahrenheit."""
        self.celsius = (value - 32) * 5/9

    @property
    def kelvin(self):
        return self._celsius + 273.15

t = Temperature(25)
print(t.celsius)      # 25
print(t.fahrenheit)   # 77.0
print(t.kelvin)       # 298.15

t.fahrenheit = 98.6
print(f"{t.celsius:.1f}°C")   # 37.0°C

# try:
#     t.celsius = -300   # ValueError!

class Circle:
    def __init__(self, radius):
        self.radius = radius   # triggers setter

    @property
    def radius(self):
        return self._radius

    @radius.setter
    def radius(self, value):
        if value < 0:
            raise ValueError("Radius cannot be negative")
        self._radius = value

    @property
    def diameter(self):
        return self._radius * 2

    @property
    def area(self):
        import math
        return math.pi * self._radius ** 2

    @property
    def circumference(self):
        import math
        return 2 * math.pi * self._radius

    def __repr__(self):
        return f"Circle(radius={self._radius})"

c = Circle(5)
print(c.radius)          # 5
print(c.diameter)        # 10
print(f"{c.area:.2f}")   # 78.54
c.radius = 10            # update via setter
print(c.diameter)        # 20

class Book:
    def __init__(self, title, author, pages):
        self.title  = title
        self.author = author
        self.pages  = pages

    def __str__(self):
        """Human-readable string — used by print()"""
        return f'"{self.title}" by {self.author}'

    def __repr__(self):
        """Developer string — used in REPL/debugging"""
        return f"Book(title={self.title!r}, author={self.author!r}, pages={self.pages})"

    def __len__(self):
        """Called by len()"""
        return self.pages

    def __bool__(self):
        """Called in boolean context"""
        return self.pages > 0

book = Book("Python Crash Course", "Eric Matthes", 544)
print(book)          # "Python Crash Course" by Eric Matthes
print(repr(book))    # Book(title='Python Crash Course', ...)
print(len(book))     # 544
print(bool(book))    # True
if book:
    print("Book has content!")

class Vector:
    """2D Vector with math operations."""

    def __init__(self, x, y):
        self.x = x
        self.y = y

    def __str__(self):
        return f"Vector({self.x}, {self.y})"

    def __repr__(self):
        return f"Vector({self.x!r}, {self.y!r})"

    # Arithmetic
    def __add__(self, other):
        return Vector(self.x + other.x, self.y + other.y)

    def __sub__(self, other):
        return Vector(self.x - other.x, self.y - other.y)

    def __mul__(self, scalar):
        return Vector(self.x * scalar, self.y * scalar)

    def __rmul__(self, scalar):   # scalar * vector
        return self.__mul__(scalar)

    # Comparison
    def __eq__(self, other):
        return self.x == other.x and self.y == other.y

    def __lt__(self, other):
        return self.magnitude() < other.magnitude()

    # Other
    def __abs__(self):
        import math
        return math.sqrt(self.x**2 + self.y**2)

    def magnitude(self):
        return abs(self)

    def __neg__(self):
        return Vector(-self.x, -self.y)

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

print(v1 + v2)      # Vector(4, 6)
print(v2 - v1)      # Vector(2, 2)
print(v1 * 3)       # Vector(3, 6)
print(3 * v1)       # Vector(3, 6)
print(abs(v2))      # 5.0
print(v1 == v1)     # True
print(v1 < v2)      # True

class Playlist:
    """A music playlist that behaves like a container."""

    def __init__(self, name):
        self.name   = name
        self._songs = []

    def add(self, song):
        self._songs.append(song)

    def __len__(self):
        return len(self._songs)

    def __getitem__(self, index):
        return self._songs[index]

    def __contains__(self, song):
        return song in self._songs

    def __iter__(self):
        return iter(self._songs)

    def __str__(self):
        return f"Playlist '{self.name}' ({len(self)} songs)"

pl = Playlist("Chill Vibes")
pl.add("Song A")
pl.add("Song B")
pl.add("Song C")

print(pl)                   # Playlist 'Chill Vibes' (3 songs)
print(len(pl))              # 3
print(pl[0])                # Song A
print("Song B" in pl)       # True
for song in pl:             # iteration works!
    print(f"  ♪ {song}")

from dataclasses import dataclass, field
from typing import List

@dataclass
class Point:
    x: float
    y: float

p1 = Point(1.0, 2.0)
p2 = Point(1.0, 2.0)
print(p1)          # Point(x=1.0, y=2.0)
print(p1 == p2)    # True  (auto __eq__)

@dataclass
class Student:
    name:   str
    age:    int
    grades: List[float] = field(default_factory=list)
    active: bool = True

    def average(self):
        return sum(self.grades) / len(self.grades) if self.grades else 0

s = Student("Alice", 20, [90.0, 85.5, 92.0])
print(s)
print(f"Average: {s.average():.1f}")

# Frozen dataclass — immutable (like a named tuple)
@dataclass(frozen=True)
class Color:
    r: int
    g: int
    b: int

red = Color(255, 0, 0)
print(red)
# red.r = 100  ← FrozenInstanceError!

# Parent (base) class
class Animal:
    def __init__(self, name, age):
        self.name = name
        self.age  = age

    def eat(self):
        return f"{self.name} is eating."

    def sleep(self):
        return f"{self.name} is sleeping."

    def __str__(self):
        return f"{self.__class__.__name__}(name={self.name!r}, age={self.age})"

# Child class — inherits from Animal
class Dog(Animal):
    def __init__(self, name, age, breed):
        super().__init__(name, age)   # call parent __init__
        self.breed = breed            # add new attribute

    def bark(self):                   # new method
        return f"{self.name} says: Woof!"

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

class Cat(Animal):
    def __init__(self, name, age, indoor=True):
        super().__init__(name, age)
        self.indoor = indoor

    def meow(self):
        return f"{self.name} says: Meow!"

    def purr(self):
        return f"{self.name} purrs contentedly."

# Usage
dog = Dog("Rex", 3, "German Shepherd")
cat = Cat("Whiskers", 5)

# Inherited methods work on child objects
print(dog.eat())     # Rex is eating.
print(dog.sleep())   # Rex is sleeping.
print(dog.bark())    # Rex says: Woof!
print(cat.meow())    # Whiskers says: Meow!
print(dog)           # Dog(name='Rex', age=3)

class Vehicle:
    def __init__(self, make, model, year):
        self.make  = make
        self.model = model
        self.year  = year
        self.speed = 0

    def accelerate(self, amount):
        self.speed += amount
        return f"{self.make} {self.model} speed: {self.speed} km/h"

    def brake(self, amount):
        self.speed = max(0, self.speed - amount)
        return f"{self.make} {self.model} speed: {self.speed} km/h"

    def __str__(self):
        return f"{self.year} {self.make} {self.model}"

class ElectricVehicle(Vehicle):
    def __init__(self, make, model, year, battery_kwh):
        super().__init__(make, model, year)   # extend parent init
        self.battery_kwh  = battery_kwh
        self.charge_level = 100               # percent

    def charge(self, percent):
        self.charge_level = min(100, self.charge_level + percent)
        return f"Charged to {self.charge_level}%"

    # Override parent method AND extend it
    def accelerate(self, amount):
        if self.charge_level < 10:
            return f"Low battery! Charge before driving."
        self.charge_level -= amount * 0.1     # drain battery
        result = super().accelerate(amount)   # call parent method
        return f"{result} | Battery: {self.charge_level:.1f}%"

    def __str__(self):
        base = super().__str__()              # reuse parent __str__
        return f"{base} (Electric, {self.battery_kwh}kWh)"

ev = ElectricVehicle("Tesla", "Model 3", 2024, 75)
print(ev)
print(ev.accelerate(30))
print(ev.brake(10))
print(ev.charge(5))

class Shape:
    def __init__(self, color="black"):
        self.color = color

    def area(self):
        raise NotImplementedError("Subclasses must implement area()")

    def perimeter(self):
        raise NotImplementedError("Subclasses must implement perimeter()")

    def describe(self):
        return (f"{self.__class__.__name__}: color={self.color}, "
                f"area={self.area():.2f}, perimeter={self.perimeter():.2f}")

import math

class Circle(Shape):
    def __init__(self, radius, color="black"):
        super().__init__(color)
        self.radius = radius

    def area(self):           # override
        return math.pi * self.radius ** 2

    def perimeter(self):      # override
        return 2 * math.pi * self.radius

class Rectangle(Shape):
    def __init__(self, width, height, color="black"):
        super().__init__(color)
        self.width  = width
        self.height = height

    def area(self):
        return self.width * self.height

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

class Triangle(Shape):
    def __init__(self, a, b, c, color="black"):
        super().__init__(color)
        self.a, self.b, self.c = a, b, c

    def area(self):
        s = self.perimeter() / 2
        return math.sqrt(s*(s-self.a)*(s-self.b)*(s-self.c))

    def perimeter(self):
        return self.a + self.b + self.c

shapes = [Circle(5, "red"), Rectangle(4, 6, "blue"), Triangle(3, 4, 5)]
for shape in shapes:
    print(shape.describe())

class Dog:
    def speak(self):
        return "Woof!"

class Cat:
    def speak(self):
        return "Meow!"

class Duck:
    def speak(self):
        return "Quack!"

class Robot:
    def speak(self):
        return "Beep boop."

# Polymorphism — same interface, different behaviour
animals = [Dog(), Cat(), Duck(), Robot()]
for creature in animals:
    print(f"{creature.__class__.__name__}: {creature.speak()}")

# Duck typing — "if it walks like a duck and quacks like a duck..."
# Python doesn't care about the TYPE, only that the METHOD exists

class FileLogger:
    def log(self, message):
        print(f"[FILE]    {message}")

class ConsoleLogger:
    def log(self, message):
        print(f"[CONSOLE] {message}")

class DatabaseLogger:
    def log(self, message):
        print(f"[DB]      {message}")

def process_event(event, logger):
    # Works with ANY object that has a .log() method
    logger.log(f"Processing: {event}")

for logger in [FileLogger(), ConsoleLogger(), DatabaseLogger()]:
    process_event("user_login", logger)

from abc import ABC, abstractmethod
import math

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

    def __init__(self, color="black"):
        self.color = color

    @abstractmethod
    def area(self):
        """Must be implemented by every subclass."""
        pass

    @abstractmethod
    def perimeter(self):
        """Must be implemented by every subclass."""
        pass

    # Concrete method — shared by all subclasses
    def describe(self):
        return (f"{self.__class__.__name__}("
                f"area={self.area():.2f}, "
                f"perimeter={self.perimeter():.2f})")

class Circle(Shape):
    def __init__(self, radius, color="black"):
        super().__init__(color)
        self.radius = radius

    def area(self):
        return math.pi * self.radius ** 2

    def perimeter(self):
        return 2 * math.pi * self.radius

class Square(Shape):
    def __init__(self, side, color="black"):
        super().__init__(color)
        self.side = side

    def area(self):
        return self.side ** 2

    def perimeter(self):
        return 4 * self.side

# shape = Shape()   ← TypeError: Can't instantiate abstract class!
c = Circle(5)
s = Square(4)
print(c.describe())   # Circle(area=78.54, perimeter=31.42)
print(s.describe())   # Square(area=16.00, perimeter=16.00)

from abc import ABC, abstractmethod

class PaymentProcessor(ABC):
    """Abstract payment processor — defines the contract."""

    @abstractmethod
    def process_payment(self, amount):
        pass

    @abstractmethod
    def refund(self, amount):
        pass

    @abstractmethod
    def get_balance(self):
        pass

    # Concrete method available to all
    def validate_amount(self, amount):
        if amount <= 0:
            raise ValueError(f"Amount must be positive: {amount}")
        return True

class CreditCardProcessor(PaymentProcessor):
    def __init__(self, card_number):
        self._card   = card_number[-4:]
        self._balance = 5000.0

    def process_payment(self, amount):
        self.validate_amount(amount)
        self._balance -= amount
        return f"Credit card ****{self._card}: charged ${amount:.2f}"

    def refund(self, amount):
        self._balance += amount
        return f"Credit card ****{self._card}: refunded ${amount:.2f}"

    def get_balance(self):
        return self._balance

class PayPalProcessor(PaymentProcessor):
    def __init__(self, email):
        self._email   = email
        self._balance = 1000.0

    def process_payment(self, amount):
        self.validate_amount(amount)
        self._balance -= amount
        return f"PayPal ({self._email}): charged ${amount:.2f}"

    def refund(self, amount):
        self._balance += amount
        return f"PayPal ({self._email}): refunded ${amount:.2f}"

    def get_balance(self):
        return self._balance

def checkout(processor, amount):
    print(processor.process_payment(amount))
    print(f"Remaining balance: ${processor.get_balance():.2f}")

cc     = CreditCardProcessor("1234567890123456")
paypal = PayPalProcessor("[email protected]")
checkout(cc, 99.99)
checkout(paypal, 49.99)

class Flyable:
    def fly(self):
        return f"{self.__class__.__name__} is flying!"

    def land(self):
        return f"{self.__class__.__name__} has landed."

class Swimmable:
    def swim(self):
        return f"{self.__class__.__name__} is swimming!"

    def dive(self):
        return f"{self.__class__.__name__} dives underwater."

class Walkable:
    def walk(self):
        return f"{self.__class__.__name__} is walking."

# Duck can fly, swim, and walk
class Duck(Flyable, Swimmable, Walkable):
    def quack(self):
        return "Quack!"

# FlyingFish can fly and swim
class FlyingFish(Flyable, Swimmable):
    pass

duck = Duck()
print(duck.fly())    # Duck is flying!
print(duck.swim())   # Duck is swimming!
print(duck.walk())   # Duck is walking.
print(duck.quack())  # Quack!

# Method Resolution Order (MRO)
print(Duck.__mro__)
# (Duck, Flyable, Swimmable, Walkable, object)

class LogMixin:
    """Adds logging capability to any class."""
    def log(self, message, level="INFO"):
        print(f"[{level}] {self.__class__.__name__}: {message}")

class JsonMixin:
    """Adds JSON serialization to any class."""
    def to_json(self):
        import json
        return json.dumps(self.__dict__, indent=2, default=str)

class ValidateMixin:
    """Adds field validation to any class."""
    def validate(self):
        errors = []
        for attr, value in self.__dict__.items():
            if value is None:
                errors.append(f"'{attr}' cannot be None")
        return errors if errors else None

# Combine mixins with a real class
class User(LogMixin, JsonMixin, ValidateMixin):
    def __init__(self, name, email, age):
        self.name  = name
        self.email = email
        self.age   = age

    def save(self):
        errors = self.validate()
        if errors:
            for e in errors:
                self.log(e, "ERROR")
            return False
        self.log(f"Saving user: {self.name}")
        return True

user = User("Alice", "[email protected]", 25)
user.save()
print(user.to_json())
user.log("Profile updated")

class Animal: pass
class Dog(Animal): pass
class Cat(Animal): pass
class GoldenRetriever(Dog): pass

dog    = Dog()
cat    = Cat()
golden = GoldenRetriever()

# isinstance — checks if object is instance of class (or subclass)
print(isinstance(dog,    Dog))     # True
print(isinstance(dog,    Animal))  # True  (Dog IS-A Animal)
print(isinstance(dog,    Cat))     # False
print(isinstance(golden, Dog))     # True  (GoldenRetriever IS-A Dog)
print(isinstance(golden, Animal))  # True  (IS-A Animal too)

# Check against multiple types (tuple)
print(isinstance(dog, (Dog, Cat)))  # True

# issubclass — checks class hierarchy (not instances)
print(issubclass(Dog,             Animal))  # True
print(issubclass(GoldenRetriever, Dog))     # True
print(issubclass(GoldenRetriever, Animal))  # True
print(issubclass(Cat,             Dog))     # False

# Practical use — safe method dispatch
def make_sound(animal):
    if isinstance(animal, Dog):
        return "Woof!"
    elif isinstance(animal, Cat):
        return "Meow!"
    else:
        return "..."

# Better approach — polymorphism (no isinstance needed)
class Animal:
    def speak(self): return "..."
class Dog(Animal):
    def speak(self): return "Woof!"
class Cat(Animal):
    def speak(self): return "Meow!"

for a in [Dog(), Cat(), Animal()]:
    print(a.speak())

class A:
    def hello(self):
        return "Hello from A"

class B(A):
    def hello(self):
        return "Hello from B"

class C(A):
    def hello(self):
        return "Hello from C"

class D(B, C):
    pass

d = D()
print(d.hello())        # "Hello from B"  (B comes first in MRO)
print(D.__mro__)
# (D, B, C, A, object)

# Python uses C3 linearization — left to right, depth first
# D → B → C → A → object

# super() follows the MRO chain
class Base:
    def greet(self):
        return "Base"

class Left(Base):
    def greet(self):
        return f"Left -> {super().greet()}"

class Right(Base):
    def greet(self):
        return f"Right -> {super().greet()}"

class Child(Left, Right):
    def greet(self):
        return f"Child -> {super().greet()}"

c = Child()
print(c.greet())
# Child -> Left -> Right -> Base
print(Child.__mro__)
# (Child, Left, Right, Base, object)

from abc import ABC, abstractmethod

class Employee(ABC):
    def __init__(self, name, employee_id, department):
        self.name        = name
        self.employee_id = employee_id
        self.department  = department

    @abstractmethod
    def calculate_pay(self):
        pass

    def get_info(self):
        return (f"[{self.employee_id}] {self.name} "
                f"| Dept: {self.department} "
                f"| Pay: ${self.calculate_pay():,.2f}")

class FullTimeEmployee(Employee):
    def __init__(self, name, emp_id, dept, annual_salary):
        super().__init__(name, emp_id, dept)
        self.annual_salary = annual_salary

    def calculate_pay(self):
        return self.annual_salary / 12   # monthly pay

class PartTimeEmployee(Employee):
    def __init__(self, name, emp_id, dept, hourly_rate, hours_worked):
        super().__init__(name, emp_id, dept)
        self.hourly_rate  = hourly_rate
        self.hours_worked = hours_worked

    def calculate_pay(self):
        return self.hourly_rate * self.hours_worked

class Contractor(Employee):
    def __init__(self, name, emp_id, dept, day_rate, days_worked):
        super().__init__(name, emp_id, dept)
        self.day_rate    = day_rate
        self.days_worked = days_worked

    def calculate_pay(self):
        return self.day_rate * self.days_worked

class Manager(FullTimeEmployee):
    def __init__(self, name, emp_id, dept, annual_salary, bonus_pct):
        super().__init__(name, emp_id, dept, annual_salary)
        self.bonus_pct = bonus_pct
        self.reports   = []

    def calculate_pay(self):
        base  = super().calculate_pay()
        bonus = base * (self.bonus_pct / 100)
        return base + bonus

    def add_report(self, employee):
        self.reports.append(employee)

# Build the team
team = [
    Manager("Alice",   "E001", "Engineering", 120000, 15),
    FullTimeEmployee("Bob",     "E002", "Engineering", 80000),
    PartTimeEmployee("Carol",   "E003", "Design",  25, 80),
    Contractor("Dave",    "C001", "Marketing",  500, 10),
]

print("=== Monthly Payroll ===")
total = 0
for emp in team:
    print(emp.get_info())
    total += emp.calculate_pay()
print(f"\nTotal payroll: ${total:,.2f}")

# Iterable — has __iter__() method
my_list = [1, 2, 3]

# iter() creates an iterator from an iterable
it = iter(my_list)
print(type(it))         # 

# next() fetches the next value
print(next(it))         # 1
print(next(it))         # 2
print(next(it))         # 3
# print(next(it))       # StopIteration — exhausted!

# for loops do this automatically under the hood:
# 1. Call iter() on the iterable
# 2. Call next() repeatedly until StopIteration

# Strings, dicts, files are all iterables
for char in "Hello":
    print(char, end=" ")   # H e l l o

# Check if something is iterable
from collections.abc import Iterable, Iterator
print(isinstance([1,2,3], Iterable))    # True
print(isinstance(iter([1,2,3]), Iterator))  # True
print(isinstance(42, Iterable))         # False

class CountUp:
    """Counts from start to stop (inclusive)."""

    def __init__(self, start, stop, step=1):
        self.current = start
        self.stop    = stop
        self.step    = step

    def __iter__(self):
        return self       # the object itself is the iterator

    def __next__(self):
        if self.current > self.stop:
            raise StopIteration
        value        = self.current
        self.current += self.step
        return value

# Use in a for loop
for n in CountUp(1, 10, 2):
    print(n, end=" ")    # 1 3 5 7 9

# Use with next()
counter = CountUp(1, 3)
print(next(counter))    # 1
print(next(counter))    # 2
print(next(counter))    # 3

# Convert to list
print(list(CountUp(0, 8, 2)))   # [0, 2, 4, 6, 8]

class Fibonacci:
    """Generates Fibonacci numbers up to a limit."""

    def __init__(self, limit):
        self.limit = limit
        self.a, self.b = 0, 1

    def __iter__(self):
        return self

    def __next__(self):
        if self.a > self.limit:
            raise StopIteration
        value    = self.a
        self.a, self.b = self.b, self.a + self.b
        return value

# Print all Fibonacci numbers up to 100
print(list(Fibonacci(100)))
# [0, 1, 1, 2, 3, 5, 8, 13, 21, 34, 55, 89]

# Iterable class (separate iterator state each time)
class FibSequence:
    def __init__(self, limit):
        self.limit = limit

    def __iter__(self):
        a, b = 0, 1
        while a <= self.limit:
            yield a             # generator inside __iter__!
            a, b = b, a + b

fibs = FibSequence(50)
for n in fibs: print(n, end=" ")   # can iterate multiple times

def count_up(start, stop, step=1):
    """Generator version of CountUp — much simpler!"""
    current = start
    while current <= stop:
        yield current          # pause here, return value
        current += step        # resume here on next()

# Works exactly like the class-based iterator
for n in count_up(1, 10, 2):
    print(n, end=" ")   # 1 3 5 7 9

# Generator object
gen = count_up(1, 5)
print(type(gen))        # 
print(next(gen))        # 1
print(next(gen))        # 2

# Generators are LAZY — values computed on demand
def infinite_counter(start=0):
    """Infinite generator — never raises StopIteration."""
    n = start
    while True:
        yield n
        n += 1

counter = infinite_counter(10)
print([next(counter) for _ in range(5)])   # [10, 11, 12, 13, 14]

# Take first N items from any iterable
import itertools
first_10 = list(itertools.islice(infinite_counter(), 10))
print(first_10)   # [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

def demo_generator():
    print("Step 1: before first yield")
    yield 10
    print("Step 2: between yields")
    yield 20
    print("Step 3: before return")
    yield 30
    print("Step 4: generator exhausted")

gen = demo_generator()

print("Calling next() #1")
val = next(gen)
print(f"Got: {val}")

print("Calling next() #2")
val = next(gen)
print(f"Got: {val}")

print("Calling next() #3")
val = next(gen)
print(f"Got: {val}")

# Output:
# Calling next() #1
# Step 1: before first yield
# Got: 10
# Calling next() #2
# Step 2: between yields
# Got: 20
# Calling next() #3
# Step 3: before return
# Got: 30

import os

def read_large_file(filepath):
    """Read a huge file line by line — never loads all into memory."""
    with open(filepath, "r", encoding="utf-8") as f:
        for line in f:
            yield line.strip()

def find_files(directory, extension):
    """Walk a directory tree and yield matching files."""
    for root, dirs, files in os.walk(directory):
        for filename in files:
            if filename.endswith(extension):
                yield os.path.join(root, filename)

def pipeline(*funcs):
    """Chain generator functions into a pipeline."""
    def run(data):
        for func in funcs:
            data = func(data)
        return data
    return run

def only_errors(lines):
    for line in lines:
        if "ERROR" in line:
            yield line

def strip_prefix(lines):
    for line in lines:
        yield line.replace("ERROR: ", "")

# Usage: process only error lines from a large log file
# process = pipeline(only_errors, strip_prefix)
# for line in process(read_large_file("app.log")):
#     print(line)

# List comprehension — creates entire list in memory
squares_list = [x**2 for x in range(1000000)]   # 8 MB in memory!

# Generator expression — lazy, almost no memory
squares_gen  = (x**2 for x in range(1000000))   # ~200 bytes!

import sys
print(sys.getsizeof(squares_list))   # ~8,000,056 bytes
print(sys.getsizeof(squares_gen))    # 104 bytes

# Use generator expressions directly in functions
total = sum(x**2 for x in range(1000))        # no extra []
evens = list(x for x in range(20) if x % 2 == 0)

# Chaining generator expressions (pipeline)
numbers  = range(1, 101)
evens    = (n for n in numbers if n % 2 == 0)
squares  = (n**2 for n in evens)
filtered = (n for n in squares if n > 100)
result   = list(filtered)
print(result[:5])   # [144, 196, 256, 324, 400]

# any() and all() with generators — short-circuit evaluation
data = [2, 4, 6, 8, 10]
print(all(x % 2 == 0 for x in data))   # True
print(any(x > 9     for x in data))    # True

# Basic syntax: [expression for item in iterable if condition]

# Basic transformation
squares = [x**2 for x in range(1, 11)]
print(squares)   # [1, 4, 9, 16, 25, 36, 49, 64, 81, 100]

# With condition (filter)
evens = [x for x in range(20) if x % 2 == 0]
print(evens)     # [0, 2, 4, 6, 8, 10, 12, 14, 16, 18]

# String processing
words  = ["hello", "world", "python", "rocks"]
upper  = [w.upper() for w in words]
long_w = [w for w in words if len(w) > 5]
print(upper)     # ['HELLO', 'WORLD', 'PYTHON', 'ROCKS']
print(long_w)    # ['python', 'rocks']

# Conditional expression (ternary) inside comprehension
numbers = range(-5, 6)
labels  = ["pos" if x > 0 else "neg" if x < 0 else "zero"
           for x in numbers]
print(labels)

# Flatten a nested list
matrix  = [[1,2,3],[4,5,6],[7,8,9]]
flat    = [n for row in matrix for n in row]
print(flat)   # [1, 2, 3, 4, 5, 6, 7, 8, 9]

# Equivalent for loop (for comparison)
result = []
for row in matrix:
    for n in row:
        result.append(n)

# Dict comprehension: {key: value for item in iterable}

# Square numbers as dict
squares = {x: x**2 for x in range(1, 6)}
print(squares)   # {1: 1, 2: 4, 3: 9, 4: 16, 5: 25}

# Invert a dictionary
original = {"a": 1, "b": 2, "c": 3}
inverted = {v: k for k, v in original.items()}
print(inverted)  # {1: 'a', 2: 'b', 3: 'c'}

# Filter a dict
scores = {"Alice": 92, "Bob": 75, "Carol": 88, "Dave": 60}
passing = {name: score for name, score in scores.items() if score >= 80}
print(passing)   # {'Alice': 92, 'Carol': 88}

# Transform values
upper_scores = {name.upper(): score for name, score in scores.items()}
print(upper_scores)

# From two lists (zip)
keys   = ["name", "age", "city"]
values = ["Alice", 25, "London"]
record = {k: v for k, v in zip(keys, values)}
print(record)   # {'name': 'Alice', 'age': 25, 'city': 'London'}

# Set comprehension: {expression for item in iterable}
unique_lengths = {len(w) for w in ["hi", "hello", "hey", "world"]}
print(unique_lengths)   # {2, 5}  (unordered, unique)

vowels = {c for c in "hello world" if c in "aeiou"}
print(vowels)   # {'e', 'o'}

# Matrix creation
rows, cols = 3, 4
matrix = [[0] * cols for _ in range(rows)]
print(matrix)   # [[0,0,0,0],[0,0,0,0],[0,0,0,0]]

# Multiplication table
table = [[i * j for j in range(1, 6)] for i in range(1, 6)]
for row in table:
    print([f"{n:3}" for n in row])

# Transpose a matrix
original  = [[1,2,3],[4,5,6],[7,8,9]]
transposed = [[row[i] for row in original] for i in range(3)]
print(transposed)   # [[1,4,7],[2,5,8],[3,6,9]]

# Flatten with condition
data = [[1,-2,3],[-4,5,-6],[7,-8,9]]
positives = [n for row in data for n in row if n > 0]
print(positives)   # [1, 3, 5, 7, 9]

# Cartesian product
colors = ["red", "blue"]
sizes  = ["S", "M", "L"]
combos = [(c, s) for c in colors for s in sizes]
print(combos)
# [('red','S'),('red','M'),('red','L'),('blue','S'),...]

def gen_a():
    yield 1
    yield 2
    yield 3

def gen_b():
    yield 4
    yield 5

# Without yield from — verbose
def combined_verbose():
    for val in gen_a():
        yield val
    for val in gen_b():
        yield val

# With yield from — clean delegation
def combined():
    yield from gen_a()
    yield from gen_b()
    yield from range(6, 9)

print(list(combined()))   # [1, 2, 3, 4, 5, 6, 7, 8]

# Recursive generator with yield from
def flatten(nested):
    """Flatten arbitrarily nested lists."""
    for item in nested:
        if isinstance(item, list):
            yield from flatten(item)   # recurse
        else:
            yield item

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

# Tree traversal with yield from
def tree_values(node):
    """Yield all values in a nested dict tree."""
    if isinstance(node, dict):
        for child in node.values():
            yield from tree_values(child)
    else:
        yield node

tree = {"a": {"b": 1, "c": 2}, "d": {"e": 3, "f": {"g": 4}}}
print(list(tree_values(tree)))   # [1, 2, 3, 4]

def accumulator():
    """Generator that accumulates sent values."""
    total = 0
    while True:
        value = yield total    # yield sends total OUT, receives value IN
        if value is None:
            break
        total += value

acc = accumulator()
next(acc)          # prime the generator (advance to first yield)
print(acc.send(10))   # 10
print(acc.send(20))   # 30
print(acc.send(5))    # 35
print(acc.send(15))   # 50

# Running average generator
def running_average():
    total = 0
    count = 0
    average = None
    while True:
        value = yield average
        if value is None:
            return
        total   += value
        count   += 1
        average  = total / count

avg = running_average()
next(avg)
print(avg.send(10))    # 10.0
print(avg.send(20))    # 15.0
print(avg.send(30))    # 20.0
print(avg.send(40))    # 25.0

import itertools

# --- Infinite iterators ---
# count(start, step) — infinite counter
evens = itertools.islice(itertools.count(0, 2), 5)
print(list(evens))   # [0, 2, 4, 6, 8]

# cycle — repeat a sequence forever
colors = itertools.islice(itertools.cycle(["R","G","B"]), 7)
print(list(colors))  # ['R','G','B','R','G','B','R']

# repeat — repeat a value N times
print(list(itertools.repeat("x", 4)))   # ['x','x','x','x']

# --- Combining iterators ---
# chain — concatenate iterables
print(list(itertools.chain([1,2],[3,4],[5,6])))   # [1,2,3,4,5,6]

# zip_longest — zip with fill value
a = [1, 2, 3]
b = ["a", "b"]
print(list(itertools.zip_longest(a, b, fillvalue="-")))
# [(1,'a'),(2,'b'),(3,'-')]

# --- Filtering ---
# takewhile — take while condition is True
data = [2, 4, 6, 7, 8, 10]
print(list(itertools.takewhile(lambda x: x % 2 == 0, data)))
# [2, 4, 6]

# dropwhile — drop while condition is True
print(list(itertools.dropwhile(lambda x: x % 2 == 0, data)))
# [7, 8, 10]

# compress — filter by selector
data     = ["a","b","c","d","e"]
selector = [1, 0, 1, 0, 1]
print(list(itertools.compress(data, selector)))   # ['a','c','e']

# --- Combinatorics ---
print(list(itertools.combinations("ABCD", 2)))
print(list(itertools.permutations([1,2,3])))
print(list(itertools.product([0,1], repeat=3)))

# Functions can be assigned to variables
def greet(name):
    return f"Hello, {name}!"

say_hello = greet           # no () — we're assigning the function itself
print(say_hello("Alice"))   # Hello, Alice!
print(greet is say_hello)   # True — same object

# Functions can be passed as arguments
def apply(func, value):
    return func(value)

print(apply(str.upper, "hello"))   # HELLO
print(apply(len, "Python"))        # 6

# Functions can be returned from functions
def make_multiplier(n):
    def multiplier(x):
        return x * n          # n is captured from outer scope
    return multiplier         # return the inner function

double = make_multiplier(2)
triple = make_multiplier(3)
print(double(5))   # 10
print(triple(5))   # 15

# Functions can be stored in data structures
operations = {
    "add":  lambda x, y: x + y,
    "sub":  lambda x, y: x - y,
    "mul":  lambda x, y: x * y,
    "div":  lambda x, y: x / y,
}
print(operations["add"](3, 4))   # 7

def make_counter(start=0):
    """Returns a counter function that remembers its count."""
    count = start

    def counter():
        nonlocal count        # modify the enclosing variable
        count += 1
        return count

    return counter

c1 = make_counter()
c2 = make_counter(10)

print(c1())   # 1
print(c1())   # 2
print(c1())   # 3
print(c2())   # 11  (independent state)
print(c2())   # 12

# Inspect closure variables
print(c1.__closure__[0].cell_contents)   # 3 (current count)

# Practical closure — partial application
def make_validator(min_val, max_val):
    def validate(value):
        if not (min_val <= value <= max_val):
            raise ValueError(f"{value} not in [{min_val}, {max_val}]")
        return value
    return validate

validate_age    = make_validator(0, 150)
validate_score  = make_validator(0, 100)
validate_rating = make_validator(1, 5)

print(validate_age(25))      # 25
print(validate_score(95))    # 95
# validate_rating(6)         # ValueError!

# Step 1 — manual wrapping (what decorators do under the hood)
def shout(func):
    def wrapper(*args, **kwargs):
        result = func(*args, **kwargs)
        return result.upper()
    return wrapper

def greet(name):
    return f"Hello, {name}!"

greet = shout(greet)          # manual decoration
print(greet("Alice"))         # HELLO, ALICE!

# Step 2 — using @ syntax (cleaner, identical result)
def shout(func):
    def wrapper(*args, **kwargs):
        result = func(*args, **kwargs)
        return result.upper()
    return wrapper

@shout
def greet(name):
    return f"Hello, {name}!"

print(greet("Alice"))         # HELLO, ALICE!

# @shout is exactly equivalent to: greet = shout(greet)

from functools import wraps

# ❌ Without @wraps — metadata is lost
def bad_decorator(func):
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

@bad_decorator
def my_function():
    """This is my function's docstring."""
    pass

print(my_function.__name__)   # wrapper  ← WRONG!
print(my_function.__doc__)    # None     ← WRONG!

# ✅ With @wraps — metadata is preserved
def good_decorator(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

@good_decorator
def my_function():
    """This is my function's docstring."""
    pass

print(my_function.__name__)   # my_function  ← correct!
print(my_function.__doc__)    # This is my function's docstring.

import time
from functools import wraps

def timer(func):
    """Measure and print function execution time."""
    @wraps(func)
    def wrapper(*args, **kwargs):
        start  = time.perf_counter()
        result = func(*args, **kwargs)
        end    = time.perf_counter()
        print(f"⏱ {func.__name__}() took {end - start:.4f}s")
        return result
    return wrapper

@timer
def slow_sum(n):
    return sum(range(n))

@timer
def bubble_sort(lst):
    lst = lst[:]
    for i in range(len(lst)):
        for j in range(len(lst) - i - 1):
            if lst[j] > lst[j+1]:
                lst[j], lst[j+1] = lst[j+1], lst[j]
    return lst

print(slow_sum(1_000_000))
bubble_sort(list(range(500, 0, -1)))

from functools import wraps
from datetime import datetime

def logger(func):
    """Log function calls with arguments and return values."""
    @wraps(func)
    def wrapper(*args, **kwargs):
        timestamp = datetime.now().strftime("%H:%M:%S")
        args_str  = ", ".join(repr(a) for a in args)
        kwargs_str = ", ".join(f"{k}={v!r}" for k, v in kwargs.items())
        all_args  = ", ".join(filter(None, [args_str, kwargs_str]))
        print(f"[{timestamp}] CALL  {func.__name__}({all_args})")
        try:
            result = func(*args, **kwargs)
            print(f"[{timestamp}] RETURN {func.__name__} → {result!r}")
            return result
        except Exception as e:
            print(f"[{timestamp}] ERROR  {func.__name__} → {type(e).__name__}: {e}")
            raise
    return wrapper

@logger
def add(a, b):
    return a + b

@logger
def divide(a, b):
    return a / b

add(3, 4)
add(10, b=20)
divide(10, 2)
try:
    divide(5, 0)
except ZeroDivisionError:
    pass

import time
from functools import wraps

def retry(max_attempts=3, delay=1.0, exceptions=(Exception,)):
    """Retry a function on failure up to max_attempts times."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_error = None
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    last_error = e
                    print(f"Attempt {attempt}/{max_attempts} failed: {e}")
                    if attempt < max_attempts:
                        time.sleep(delay)
            raise last_error
        return wrapper
    return decorator

import random

@retry(max_attempts=4, delay=0.1, exceptions=(ValueError,))
def unstable_api_call():
    """Simulates an unreliable API call."""
    if random.random() < 0.7:
        raise ValueError("Connection timeout")
    return "Success!"

try:
    result = unstable_api_call()
    print(f"Result: {result}")
except ValueError:
    print("All attempts failed.")

from functools import wraps

def memoize(func):
    """Cache function results to avoid redundant computation."""
    cache = {}
    @wraps(func)
    def wrapper(*args):
        if args not in cache:
            cache[args] = func(*args)
        return cache[args]
    wrapper.cache = cache          # expose cache for inspection
    wrapper.clear_cache = cache.clear
    return wrapper

@memoize
def fibonacci(n):
    if n <= 1:
        return n
    return fibonacci(n - 1) + fibonacci(n - 2)

import time
start = time.perf_counter()
print(fibonacci(35))   # 9227465
print(f"Time: {time.perf_counter() - start:.4f}s")  # very fast!
print(f"Cache size: {len(fibonacci.cache)}")

# Python's built-in lru_cache — even better!
from functools import lru_cache

@lru_cache(maxsize=128)
def fib(n):
    return n if n <= 1 else fib(n-1) + fib(n-2)

print(fib(50))
print(fib.cache_info())   # CacheInfo(hits=..., misses=..., ...)

from functools import wraps

def repeat(times):
    """Call the decorated function 'times' times."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            results = []
            for _ in range(times):
                results.append(func(*args, **kwargs))
            return results
        return wrapper
    return decorator

@repeat(3)
def say(message):
    print(message)
    return message

say("Hello!")
# Hello!
# Hello!
# Hello!

# Validate argument types
def validate_types(**type_map):
    """Validate argument types at call time."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            import inspect
            sig    = inspect.signature(func)
            bound  = sig.bind(*args, **kwargs)
            bound.apply_defaults()
            for param, value in bound.arguments.items():
                if param in type_map:
                    expected = type_map[param]
                    if not isinstance(value, expected):
                        raise TypeError(
                            f"'{param}' must be {expected.__name__}, "
                            f"got {type(value).__name__}"
                        )
            return func(*args, **kwargs)
        return wrapper
    return decorator

@validate_types(name=str, age=int, score=float)
def create_student(name, age, score):
    return f"{name}, age {age}, score {score}"

print(create_student("Alice", 20, 95.5))
try:
    create_student("Bob", "twenty", 80.0)   # TypeError!
except TypeError as e:
    print(f"TypeError: {e}")

from functools import wraps
import time

def timer(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start  = time.perf_counter()
        result = func(*args, **kwargs)
        print(f"⏱ {func.__name__}: {time.perf_counter()-start:.4f}s")
        return result
    return wrapper

def logger(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        print(f"📝 Calling {func.__name__} with {args}, {kwargs}")
        result = func(*args, **kwargs)
        print(f"📝 {func.__name__} returned {result!r}")
        return result
    return wrapper

def validate_positive(func):
    @wraps(func)
    def wrapper(n, *args, **kwargs):
        if n < 0:
            raise ValueError(f"Expected positive number, got {n}")
        return func(n, *args, **kwargs)
    return wrapper

# Decorators apply BOTTOM-UP
# @timer → @logger → @validate_positive → original function
@timer
@logger
@validate_positive
def compute(n):
    """Compute sum of 0..n."""
    return sum(range(n))

result = compute(1000)
print(f"Result: {result}")

# Execution order:
# timer.wrapper → logger.wrapper → validate_positive.wrapper → compute

from functools import wraps, update_wrapper

class CallCounter:
    """Decorator that counts how many times a function is called."""

    def __init__(self, func):
        update_wrapper(self, func)
        self.func  = func
        self.count = 0

    def __call__(self, *args, **kwargs):
        self.count += 1
        print(f"Call #{self.count} to {self.func.__name__}")
        return self.func(*args, **kwargs)

    def reset(self):
        self.count = 0

@CallCounter
def greet(name):
    return f"Hello, {name}!"

greet("Alice")
greet("Bob")
greet("Carol")
print(f"Called {greet.count} times")   # Called 3 times
greet.reset()
print(f"After reset: {greet.count}")   # After reset: 0

# Class decorator with arguments
class RateLimit:
    """Limit how often a function can be called."""

    def __init__(self, calls_per_second):
        self.min_interval = 1.0 / calls_per_second
        self.last_called  = 0

    def __call__(self, func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            import time
            elapsed = time.time() - self.last_called
            if elapsed < self.min_interval:
                time.sleep(self.min_interval - elapsed)
            self.last_called = time.time()
            return func(*args, **kwargs)
        return wrapper

@RateLimit(calls_per_second=2)
def api_call(endpoint):
    return f"Response from {endpoint}"

import time
for ep in ["/users", "/posts", "/comments"]:
    print(api_call(ep))

class ManagedFile:
    """Context manager for safe file handling."""

    def __init__(self, filename, mode="r", encoding="utf-8"):
        self.filename = filename
        self.mode     = mode
        self.encoding = encoding
        self.file     = None

    def __enter__(self):
        print(f"Opening {self.filename}")
        self.file = open(self.filename, self.mode, encoding=self.encoding)
        return self.file          # value bound to 'as' variable

    def __exit__(self, exc_type, exc_val, exc_tb):
        print(f"Closing {self.filename}")
        if self.file:
            self.file.close()
        if exc_type:
            print(f"Exception handled: {exc_type.__name__}: {exc_val}")
        return False              # False = don't suppress exceptions

# Usage — identical to built-in open()
with ManagedFile("data.txt", "w") as f:
    f.write("Hello from context manager!\n")

with ManagedFile("data.txt", "r") as f:
    print(f.read())

# __exit__ parameters:
# exc_type — exception class (None if no exception)
# exc_val  — exception instance
# exc_tb   — traceback object
# return True  → suppress the exception
# return False → let the exception propagate

from contextlib import contextmanager
import time

@contextmanager
def timer_ctx(label=""):
    """Context manager that times a block of code."""
    start = time.perf_counter()
    try:
        yield                          # the 'with' block runs here
    finally:
        elapsed = time.perf_counter() - start
        print(f"⏱ {label}: {elapsed:.4f}s")

with timer_ctx("List creation"):
    data = [x**2 for x in range(100_000)]

with timer_ctx("Sum"):
    total = sum(data)

# Context manager with a value
@contextmanager
def managed_db_connection(db_name):
    """Simulate a database connection."""
    print(f"Connecting to {db_name}...")
    connection = {"db": db_name, "open": True}   # simulated connection
    try:
        yield connection
    except Exception as e:
        print(f"Rolling back due to: {e}")
        connection["open"] = False
        raise
    finally:
        connection["open"] = False
        print(f"Disconnected from {db_name}")

with managed_db_connection("users.db") as conn:
    print(f"Running query on {conn['db']}")
    print(f"Connection open: {conn['open']}")

from contextlib import contextmanager, suppress
import os

# Suppress specific exceptions
with suppress(FileNotFoundError):
    os.remove("nonexistent_file.txt")
print("No crash — exception suppressed!")

# Temporary directory change
@contextmanager
def working_directory(path):
    original = os.getcwd()
    try:
        os.chdir(path)
        yield path
    finally:
        os.chdir(original)

# Indent context (for pretty printing)
@contextmanager
def indented(level=1, char="  "):
    indent = char * level
    class Printer:
        def print(self, *args, **kwargs):
            print(indent + " ".join(str(a) for a in args), **kwargs)
    yield Printer()

with indented(2) as p:
    p.print("This is indented by 4 spaces")
    p.print("So is this")

# Redirect stdout
from contextlib import redirect_stdout
import io

buffer = io.StringIO()
with redirect_stdout(buffer):
    print("This goes to the buffer, not the console")
    print("So does this")

captured = buffer.getvalue()
print(f"Captured: {captured.strip()}")

from functools import wraps

# Simulate a current user session
current_user = {"name": "Alice", "role": "admin", "logged_in": True}

def login_required(func):
    """Ensure user is logged in."""
    @wraps(func)
    def wrapper(*args, **kwargs):
        if not current_user.get("logged_in"):
            raise PermissionError("You must be logged in.")
        return func(*args, **kwargs)
    return wrapper

def require_role(*roles):
    """Ensure user has one of the required roles."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            user_role = current_user.get("role")
            if user_role not in roles:
                raise PermissionError(
                    f"Role '{user_role}' not allowed. "
                    f"Required: {roles}"
                )
            return func(*args, **kwargs)
        return wrapper
    return decorator

@login_required
@require_role("admin", "moderator")
def delete_user(user_id):
    return f"User {user_id} deleted."

@login_required
def view_profile(user_id):
    return f"Profile of user {user_id}."

print(delete_user(42))
print(view_profile(42))

current_user["role"] = "viewer"
try:
    delete_user(42)
except PermissionError as e:
    print(f"Denied: {e}")

# CONCURRENCY — dealing with multiple tasks at once
# Tasks may not run simultaneously, but progress is interleaved
# Best for: I/O-bound tasks (network, disk, database)
# Tools: threading, asyncio

# PARALLELISM — tasks run simultaneously on multiple CPU cores
# True simultaneous execution
# Best for: CPU-bound tasks (math, image processing, ML)
# Tools: multiprocessing, concurrent.futures

# ASYNC — cooperative multitasking (single thread)
# Tasks voluntarily yield control while waiting
# Best for: high-concurrency I/O (thousands of connections)
# Tools: asyncio, async/await

# THE GIL (Global Interpreter Lock)
# CPython allows only ONE thread to execute Python bytecode at a time
# Impact: threading does NOT give true parallelism for CPU-bound work
# Solution: use multiprocessing for CPU-bound, threading for I/O-bound

# Quick decision guide:
# ┌─────────────────────────────────────────────────┐
# │ Task type    │ Best tool        │ Why             │
# ├─────────────────────────────────────────────────┤
# │ I/O-bound    │ asyncio          │ Lightweight     │
# │ I/O-bound    │ threading        │ Simple API      │
# │ CPU-bound    │ multiprocessing  │ Bypasses GIL    │
# │ Mixed        │ concurrent.futures│ Unified API    │
# └─────────────────────────────────────────────────┘

import threading
import time

def download_file(filename, duration):
    """Simulate downloading a file."""
    print(f"  Starting download: {filename}")
    time.sleep(duration)          # simulate I/O wait
    print(f"  Finished download: {filename} ({duration}s)")

# Sequential — takes sum of all durations
start = time.perf_counter()
download_file("file_a.zip", 2)
download_file("file_b.zip", 3)
download_file("file_c.zip", 1)
print(f"Sequential: {time.perf_counter()-start:.1f}s\n")

# Threaded — takes max duration (all run concurrently)
start = time.perf_counter()
threads = [
    threading.Thread(target=download_file, args=("file_a.zip", 2)),
    threading.Thread(target=download_file, args=("file_b.zip", 3)),
    threading.Thread(target=download_file, args=("file_c.zip", 1)),
]
for t in threads:
    t.start()
for t in threads:
    t.join()           # wait for all threads to finish
print(f"Threaded:   {time.perf_counter()-start:.1f}s")

import threading
import time

# Subclass Thread for more control
class WorkerThread(threading.Thread):
    def __init__(self, task_id, duration):
        super().__init__(name=f"Worker-{task_id}")
        self.task_id  = task_id
        self.duration = duration
        self.result   = None

    def run(self):
        """This runs in the new thread."""
        print(f"[{self.name}] Starting task {self.task_id}")
        time.sleep(self.duration)
        self.result = f"Task {self.task_id} done in {self.duration}s"
        print(f"[{self.name}] {self.result}")

workers = [WorkerThread(i, i * 0.5) for i in range(1, 5)]
for w in workers:
    w.start()
for w in workers:
    w.join()

print("\nResults:")
for w in workers:
    print(f"  {w.result}")

# Daemon threads — die when main thread exits
def background_monitor():
    while True:
        print("  [Monitor] Heartbeat...")
        time.sleep(1)

monitor = threading.Thread(target=background_monitor, daemon=True)
monitor.start()
print("Main thread doing work...")
time.sleep(2.5)
print("Main thread done — daemon thread will stop automatically")

import threading

# Race condition — UNSAFE shared counter
counter = 0

def unsafe_increment(n):
    global counter
    for _ in range(n):
        counter += 1   # read-modify-write: NOT atomic!

threads = [threading.Thread(target=unsafe_increment, args=(10000,))
           for _ in range(5)]
for t in threads: t.start()
for t in threads: t.join()
print(f"Unsafe counter: {counter}")   # likely < 50000!

# Fix with a Lock
counter = 0
lock    = threading.Lock()

def safe_increment(n):
    global counter
    for _ in range(n):
        with lock:         # acquire lock, release on exit
            counter += 1

threads = [threading.Thread(target=safe_increment, args=(10000,))
           for _ in range(5)]
for t in threads: t.start()
for t in threads: t.join()
print(f"Safe counter:   {counter}")   # always 50000

import threading
import queue
import time

# Event — signal between threads
ready = threading.Event()

def producer():
    print("Producer: preparing data...")
    time.sleep(1)
    ready.set()             # signal that data is ready
    print("Producer: data ready!")

def consumer():
    print("Consumer: waiting for data...")
    ready.wait()            # block until event is set
    print("Consumer: processing data!")

t1 = threading.Thread(target=producer)
t2 = threading.Thread(target=consumer)
t2.start(); t1.start()
t1.join();  t2.join()

# Semaphore — limit concurrent access
semaphore = threading.Semaphore(3)   # max 3 threads at once

def limited_resource(thread_id):
    with semaphore:
        print(f"Thread {thread_id}: using resource")
        time.sleep(0.5)

threads = [threading.Thread(target=limited_resource, args=(i,))
           for i in range(8)]
for t in threads: t.start()
for t in threads: t.join()

# Queue — thread-safe producer/consumer
task_queue = queue.Queue()

def worker():
    while True:
        task = task_queue.get()
        if task is None:
            break
        print(f"Processing: {task}")
        time.sleep(0.1)
        task_queue.task_done()

w = threading.Thread(target=worker)
w.start()
for item in ["task1", "task2", "task3"]:
    task_queue.put(item)
task_queue.put(None)    # sentinel to stop worker
w.join()

import multiprocessing
import time

def cpu_intensive(n):
    """Simulate CPU-bound work."""
    return sum(i * i for i in range(n))

if __name__ == "__main__":   # REQUIRED for multiprocessing on Windows/Mac
    numbers = [5_000_000] * 4

    # Sequential
    start = time.perf_counter()
    results = [cpu_intensive(n) for n in numbers]
    print(f"Sequential: {time.perf_counter()-start:.2f}s")

    # Multiprocessing Pool — distributes work across CPU cores
    start = time.perf_counter()
    with multiprocessing.Pool() as pool:
        results = pool.map(cpu_intensive, numbers)
    print(f"Parallel:   {time.perf_counter()-start:.2f}s")
    print(f"CPU cores:  {multiprocessing.cpu_count()}")

import multiprocessing
import time

def worker_with_result(task_id, result_queue):
    """Worker that sends result back via Queue."""
    result = task_id ** 2
    time.sleep(0.1)
    result_queue.put((task_id, result))

if __name__ == "__main__":
    result_queue = multiprocessing.Queue()
    processes = [
        multiprocessing.Process(
            target=worker_with_result,
            args=(i, result_queue)
        )
        for i in range(1, 6)
    ]
    for p in processes: p.start()
    for p in processes: p.join()

    results = {}
    while not result_queue.empty():
        task_id, result = result_queue.get()
        results[task_id] = result
    print("Results:", dict(sorted(results.items())))

    # Shared memory — Value and Array
    shared_counter = multiprocessing.Value("i", 0)
    shared_array   = multiprocessing.Array("d", [1.0, 2.0, 3.0])

    def increment(counter, lock):
        for _ in range(1000):
            with lock:
                counter.value += 1

    lock = multiprocessing.Lock()
    procs = [multiprocessing.Process(target=increment,
             args=(shared_counter, lock)) for _ in range(4)]
    for p in procs: p.start()
    for p in procs: p.join()
    print(f"Shared counter: {shared_counter.value}")   # 4000

from concurrent.futures import ThreadPoolExecutor, as_completed
import time

def fetch_data(url):
    """Simulate fetching data from a URL."""
    time.sleep(0.5)   # simulate network delay
    return f"Data from {url}"

urls = [f"https://api.example.com/item/{i}" for i in range(1, 9)]

# Sequential
start = time.perf_counter()
results = [fetch_data(url) for url in urls]
print(f"Sequential: {time.perf_counter()-start:.2f}s")

# ThreadPoolExecutor — submit all, collect results
start = time.perf_counter()
with ThreadPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(fetch_data, urls))
print(f"Threaded:   {time.perf_counter()-start:.2f}s")

# as_completed — process results as they finish (not in order)
start = time.perf_counter()
with ThreadPoolExecutor(max_workers=4) as executor:
    futures = {executor.submit(fetch_data, url): url for url in urls}
    for future in as_completed(futures):
        url    = futures[future]
        result = future.result()
        print(f"  Done: {url[-6:]} → {result[-15:]}")
print(f"as_completed: {time.perf_counter()-start:.2f}s")

from concurrent.futures import ProcessPoolExecutor, as_completed
import time

def is_prime(n):
    """Check if n is prime — CPU-bound."""
    if n < 2: return False
    for i in range(2, int(n**0.5) + 1):
        if n % i == 0:
            return False
    return True

def count_primes(start, end):
    """Count primes in range — CPU-bound."""
    return sum(1 for n in range(start, end) if is_prime(n))

if __name__ == "__main__":
    ranges = [(0, 250_000), (250_000, 500_000),
              (500_000, 750_000), (750_000, 1_000_000)]

    # Sequential
    start = time.perf_counter()
    total = sum(count_primes(s, e) for s, e in ranges)
    print(f"Sequential: {time.perf_counter()-start:.2f}s | Primes: {total}")

    # ProcessPoolExecutor — true parallelism
    start = time.perf_counter()
    with ProcessPoolExecutor() as executor:
        futures = [executor.submit(count_primes, s, e) for s, e in ranges]
        total   = sum(f.result() for f in futures)
    print(f"Parallel:   {time.perf_counter()-start:.2f}s | Primes: {total}")

    # Exception handling with futures
    def risky_task(n):
        if n == 3: raise ValueError(f"Bad input: {n}")
        return n * 10

    with ProcessPoolExecutor() as executor:
        futures = {executor.submit(risky_task, i): i for i in range(5)}
        for future in as_completed(futures):
            try:
                print(f"Result: {future.result()}")
            except ValueError as e:
                print(f"Error: {e}")

import asyncio

# async def defines a coroutine function
async def greet(name, delay):
    print(f"Hello, {name}!")
    await asyncio.sleep(delay)   # yield control while waiting
    print(f"Goodbye, {name}!")

# Run a single coroutine
asyncio.run(greet("Alice", 1))

# Run multiple coroutines CONCURRENTLY with gather
async def main():
    # All three run concurrently — total time ≈ max(delays)
    await asyncio.gather(
        greet("Alice",  1),
        greet("Bob",    2),
        greet("Carol",  1),
    )

asyncio.run(main())

# Coroutine vs regular function
async def add(a, b):
    return a + b

# add(1, 2)          → returns a coroutine object (not the result!)
# await add(1, 2)    → returns 3  (must be inside async def)
# asyncio.run(add(1, 2)) → returns 3  (entry point)

import asyncio
import time

async def fetch_page(url, delay):
    """Simulate async HTTP request."""
    print(f"  Fetching {url}...")
    await asyncio.sleep(delay)
    return f"{url}"

async def main():
    urls = [
        ("https://example.com",     0.5),
        ("https://python.org",      1.0),
        ("https://github.com",      0.3),
        ("https://stackoverflow.com", 0.8),
    ]

    start = time.perf_counter()

    # gather — run all concurrently, wait for all
    results = await asyncio.gather(
        *[fetch_page(url, delay) for url, delay in urls]
    )

    elapsed = time.perf_counter() - start
    print(f"\nAll done in {elapsed:.2f}s (sequential would be "
          f"{sum(d for _, d in urls):.1f}s)")
    for r in results:
        print(f"  {r[:40]}")

asyncio.run(main())

import asyncio

async def slow_operation(name, seconds):
    try:
        print(f"  {name}: starting ({seconds}s)")
        await asyncio.sleep(seconds)
        print(f"  {name}: done!")
        return f"{name} result"
    except asyncio.CancelledError:
        print(f"  {name}: CANCELLED")
        raise

async def main():
    # Create tasks explicitly
    task1 = asyncio.create_task(slow_operation("Task A", 1))
    task2 = asyncio.create_task(slow_operation("Task B", 5))
    task3 = asyncio.create_task(slow_operation("Task C", 2))

    # Cancel a task
    await asyncio.sleep(0.1)
    task2.cancel()

    # Wait for remaining tasks
    results = await asyncio.gather(task1, task2, task3,
                                   return_exceptions=True)
    for r in results:
        print(f"  Result: {r}")

    # Timeout — cancel if takes too long
    try:
        result = await asyncio.wait_for(
            slow_operation("Slow task", 10),
            timeout=2.0
        )
    except asyncio.TimeoutError:
        print("  Timed out after 2 seconds!")

asyncio.run(main())

import asyncio

# Async generator
async def async_range(start, stop, delay=0.1):
    for i in range(start, stop):
        await asyncio.sleep(delay)
        yield i

# Async for loop
async def consume_async_range():
    async for value in async_range(0, 5):
        print(f"  Got: {value}")

# Async comprehension
async def async_squares():
    squares = [i**2 async for i in async_range(1, 6, 0.05)]
    print(f"  Squares: {squares}")

# Async context manager
class AsyncDB:
    async def __aenter__(self):
        print("  DB: connecting...")
        await asyncio.sleep(0.1)
        return self

    async def __aexit__(self, *args):
        print("  DB: disconnecting...")
        await asyncio.sleep(0.05)

    async def query(self, sql):
        await asyncio.sleep(0.1)
        return f"Results for: {sql}"

async def main():
    await consume_async_range()
    await async_squares()
    async with AsyncDB() as db:
        result = await db.query("SELECT * FROM users")
        print(f"  {result}")

asyncio.run(main())

import asyncio
import random

# Simulated async HTTP client (replace with aiohttp in real code)
async def async_get(url, session_id=1):
    """Simulate async HTTP GET request."""
    delay = random.uniform(0.1, 0.8)
    await asyncio.sleep(delay)
    status = 200 if random.random() > 0.1 else 404
    return {"url": url, "status": status, "time": delay}

async def fetch_with_retry(url, retries=3):
    """Fetch URL with automatic retry on failure."""
    for attempt in range(1, retries + 1):
        try:
            response = await async_get(url)
            if response["status"] == 200:
                return response
            raise ValueError(f"HTTP {response['status']}")
        except ValueError as e:
            if attempt < retries:
                wait = 2 ** attempt * 0.1   # exponential backoff
                print(f"  Retry {attempt} for {url[-10:]}: {e}")
                await asyncio.sleep(wait)
            else:
                return {"url": url, "status": "FAILED", "time": 0}

async def fetch_all(urls, concurrency=5):
    """Fetch all URLs with limited concurrency."""
    semaphore = asyncio.Semaphore(concurrency)

    async def bounded_fetch(url):
        async with semaphore:
            return await fetch_with_retry(url)

    return await asyncio.gather(*[bounded_fetch(url) for url in urls])

async def main():
    urls = [f"https://api.example.com/data/{i}" for i in range(1, 13)]
    print(f"Fetching {len(urls)} URLs (max 5 concurrent)...\n")

    import time
    start   = time.perf_counter()
    results = await fetch_all(urls)
    elapsed = time.perf_counter() - start

    ok   = sum(1 for r in results if r["status"] == 200)
    fail = len(results) - ok
    print(f"\n✅ Success: {ok}  ❌ Failed: {fail}")
    print(f"⏱ Total time: {elapsed:.2f}s")

asyncio.run(main())

import asyncio
import random

async def producer(queue, n_items):
    """Produce items and put them in the queue."""
    for i in range(n_items):
        item = f"item-{i:03d}"
        await queue.put(item)
        print(f"  📦 Produced: {item}")
        await asyncio.sleep(random.uniform(0.05, 0.15))
    await queue.put(None)   # sentinel

async def consumer(queue, consumer_id):
    """Consume items from the queue."""
    while True:
        item = await queue.get()
        if item is None:
            await queue.put(None)   # pass sentinel to next consumer
            break
        await asyncio.sleep(random.uniform(0.1, 0.3))
        print(f"  ✅ Consumer {consumer_id} processed: {item}")
        queue.task_done()

async def main():
    queue = asyncio.Queue(maxsize=5)
    await asyncio.gather(
        producer(queue, 10),
        consumer(queue, 1),
        consumer(queue, 2),
    )

asyncio.run(main())

import asyncio
import threading
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import time

# ── asyncio ── best for: many I/O tasks, web servers, APIs
async def async_io_example():
    async def task(n):
        await asyncio.sleep(0.1)
        return n * 2
    results = await asyncio.gather(*[task(i) for i in range(100)])
    return results

# ── threading ── best for: simple I/O, legacy code, blocking calls
def thread_io_example():
    results = []
    lock = threading.Lock()
    def task(n):
        time.sleep(0.1)
        with lock:
            results.append(n * 2)
    threads = [threading.Thread(target=task, args=(i,)) for i in range(10)]
    for t in threads: t.start()
    for t in threads: t.join()
    return results

# ── ProcessPoolExecutor ── best for: CPU-heavy computation
def cpu_example():
    def heavy(n):
        return sum(i**2 for i in range(n))
    with ProcessPoolExecutor() as ex:
        return list(ex.map(heavy, [100_000]*4))

# ── ThreadPoolExecutor ── best for: blocking I/O in sync code
def thread_pool_example():
    def blocking_io(n):
        time.sleep(0.1)
        return n * 2
    with ThreadPoolExecutor(max_workers=10) as ex:
        return list(ex.map(blocking_io, range(10)))

# Summary:
# asyncio.run(async_io_example())  → 100 tasks in ~0.1s
# thread_io_example()              → 10 tasks in ~0.1s
# cpu_example()                    → 4 CPU tasks in parallel
# thread_pool_example()            → 10 blocking tasks in ~0.1s
print("Async IO:", asyncio.run(async_io_example())[:5])

# Testing pyramid — from fast/cheap to slow/expensive:
#
#         /\
#        /E2E\       End-to-end tests  (few, slow)
#       /──────\
#      /Integr. \    Integration tests (some)
#     /──────────\
#    /  Unit Tests \  Unit tests       (many, fast)
#   /______________\
#
# UNIT TEST      — test one function/class in isolation
# INTEGRATION    — test how components work together
# END-TO-END     — test the full system from user perspective
#
# Good tests are:
# F — Fast         (milliseconds, not seconds)
# I — Independent  (no shared state between tests)
# R — Repeatable   (same result every time)
# S — Self-validating (pass or fail, no manual check)
# T — Timely       (written alongside the code)
#
# Test naming convention:
# test___
# e.g. test_divide_by_zero_raises_error

import unittest

# Code under test
def add(a, b):       return a + b
def divide(a, b):
    if b == 0: raise ZeroDivisionError("Cannot divide by zero")
    return a / b
def is_palindrome(s):
    s = s.lower().replace(" ", "")
    return s == s[::-1]

class TestMathFunctions(unittest.TestCase):

    # setUp runs BEFORE each test method
    def setUp(self):
        self.test_values = [(1,2,3),(0,0,0),(-1,1,0),(100,200,300)]

    # tearDown runs AFTER each test method
    def tearDown(self):
        pass   # cleanup if needed

    def test_add_positive_numbers(self):
        self.assertEqual(add(2, 3), 5)

    def test_add_negative_numbers(self):
        self.assertEqual(add(-1, -1), -2)

    def test_add_zero(self):
        self.assertEqual(add(5, 0), 5)

    def test_divide_normal(self):
        self.assertEqual(divide(10, 2), 5.0)

    def test_divide_by_zero_raises(self):
        with self.assertRaises(ZeroDivisionError):
            divide(5, 0)

    def test_divide_returns_float(self):
        self.assertIsInstance(divide(7, 2), float)

    def test_palindrome_true(self):
        self.assertTrue(is_palindrome("racecar"))
        self.assertTrue(is_palindrome("A man a plan a canal Panama"))

    def test_palindrome_false(self):
        self.assertFalse(is_palindrome("hello"))

# Run tests
if __name__ == "__main__":
    unittest.main(verbosity=2)

import unittest

class TestAssertions(unittest.TestCase):

    def test_equality(self):
        self.assertEqual(1 + 1, 2)           # a == b
        self.assertNotEqual("a", "b")         # a != b

    def test_truth(self):
        self.assertTrue(5 > 3)               # bool(x) is True
        self.assertFalse(5 < 3)              # bool(x) is False

    def test_identity(self):
        self.assertIs(None, None)            # a is b
        self.assertIsNone(None)              # x is None
        self.assertIsNotNone(42)             # x is not None

    def test_membership(self):
        self.assertIn("a", ["a","b","c"])    # a in b
        self.assertNotIn("z", ["a","b","c"]) # a not in b

    def test_types(self):
        self.assertIsInstance(42, int)       # isinstance(a, b)
        self.assertIsInstance("hi", str)

    def test_numeric(self):
        self.assertAlmostEqual(3.14159, 3.14, places=2)
        self.assertGreater(5, 3)             # a > b
        self.assertLess(3, 5)               # a < b
        self.assertGreaterEqual(5, 5)        # a >= b

    def test_sequences(self):
        self.assertListEqual([1,2,3], [1,2,3])
        self.assertDictEqual({"a":1}, {"a":1})
        self.assertSetEqual({1,2,3}, {3,2,1})

    def test_raises(self):
        with self.assertRaises(ValueError) as ctx:
            int("not a number")
        self.assertIn("invalid literal", str(ctx.exception))

# Install: pip install pytest
# Run:     pytest              (discover all test_*.py files)
#          pytest -v           (verbose)
#          pytest -v -s        (show print output)
#          pytest test_math.py (specific file)
#          pytest -k "add"     (run tests matching "add")
#          pytest --tb=short   (shorter tracebacks)

# File: test_math.py
# pytest discovers functions starting with test_

def add(a, b):    return a + b
def divide(a, b):
    if b == 0: raise ZeroDivisionError
    return a / b

# No class needed — just plain functions!
def test_add_basic():
    assert add(2, 3) == 5

def test_add_negative():
    assert add(-1, -1) == -2

def test_add_zero():
    assert add(5, 0) == 5

def test_divide_normal():
    assert divide(10, 2) == 5.0

def test_divide_by_zero():
    import pytest
    with pytest.raises(ZeroDivisionError):
        divide(5, 0)

def test_divide_raises_with_message():
    import pytest
    with pytest.raises(ZeroDivisionError, match="divide"):
        divide(5, 0)

# pytest shows EXACTLY what failed:
# AssertionError: assert 4 == 5
#   where 4 = add(2, 2)

import pytest

# Simple fixture
@pytest.fixture
def sample_data():
    return {"name": "Alice", "age": 25, "scores": [90, 85, 92]}

def test_name(sample_data):
    assert sample_data["name"] == "Alice"

def test_average(sample_data):
    avg = sum(sample_data["scores"]) / len(sample_data["scores"])
    assert avg == pytest.approx(89.0, rel=0.01)

# Fixture with setup AND teardown (yield)
@pytest.fixture
def temp_file(tmp_path):
    file = tmp_path / "test.txt"
    file.write_text("Hello, pytest!")
    yield file                      # test runs here
    # teardown: tmp_path is auto-cleaned by pytest

def test_file_content(temp_file):
    assert temp_file.read_text() == "Hello, pytest!"

def test_file_exists(temp_file):
    assert temp_file.exists()

# Fixture scopes
@pytest.fixture(scope="module")     # created once per module
def db_connection():
    print("\nConnecting to DB...")
    conn = {"connected": True, "queries": 0}
    yield conn
    print("\nClosing DB connection...")

# scope options: "function" (default), "class", "module", "session"

# Parametrized fixture
@pytest.fixture(params=["sqlite", "postgres", "mysql"])
def database(request):
    return request.param

def test_db_type(database):
    assert database in ["sqlite", "postgres", "mysql"]

import pytest

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

def celsius_to_fahrenheit(c):
    return c * 9/5 + 32

# Run same test with multiple inputs
@pytest.mark.parametrize("n, expected", [
    (2,  True),
    (3,  True),
    (4,  False),
    (17, True),
    (25, False),
    (97, True),
    (1,  False),
    (0,  False),
])
def test_is_prime(n, expected):
    assert is_prime(n) == expected

# Multiple parameters
@pytest.mark.parametrize("celsius, fahrenheit", [
    (0,   32.0),
    (100, 212.0),
    (-40, -40.0),
    (37,  98.6),
])
def test_celsius_to_fahrenheit(celsius, fahrenheit):
    assert celsius_to_fahrenheit(celsius) == pytest.approx(fahrenheit)

# Parametrize with IDs for readable output
@pytest.mark.parametrize("text, expected", [
    pytest.param("racecar", True,  id="palindrome"),
    pytest.param("hello",   False, id="not-palindrome"),
    pytest.param("",        True,  id="empty-string"),
])
def test_palindrome(text, expected):
    result = text == text[::-1]
    assert result == expected

from unittest.mock import Mock, MagicMock, patch, call
import pytest

# Basic Mock
mock = Mock()
mock.return_value = 42
print(mock())          # 42
print(mock.called)     # True
print(mock.call_count) # 1

# Mock with spec — only allows real attributes
class Database:
    def get_user(self, user_id): pass
    def save_user(self, user): pass

db_mock = Mock(spec=Database)
db_mock.get_user.return_value = {"id": 1, "name": "Alice"}
result = db_mock.get_user(1)
print(result)   # {'id': 1, 'name': 'Alice'}

# Verify calls
db_mock.get_user.assert_called_once_with(1)
db_mock.get_user.assert_called_with(1)

# patch — replace a real object during a test
import json

def load_config(filepath):
    with open(filepath) as f:
        return json.load(f)

def test_load_config():
    mock_data = '{"debug": true, "port": 8080}'
    with patch("builtins.open", create=True) as mock_open:
        mock_open.return_value.__enter__.return_value.read.return_value = mock_data
        # test continues...

from unittest.mock import patch, MagicMock
import pytest

# Code under test
class UserService:
    def __init__(self, db, email_service):
        self.db            = db
        self.email_service = email_service

    def register(self, name, email):
        if self.db.user_exists(email):
            raise ValueError(f"User {email} already exists")
        user = {"name": name, "email": email, "id": 42}
        self.db.save_user(user)
        self.email_service.send_welcome(email)
        return user

    def get_user(self, user_id):
        user = self.db.get_user(user_id)
        if not user:
            raise KeyError(f"User {user_id} not found")
        return user

# Tests using mocks
def test_register_new_user():
    mock_db    = MagicMock()
    mock_email = MagicMock()
    mock_db.user_exists.return_value = False

    service = UserService(mock_db, mock_email)
    user    = service.register("Alice", "[email protected]")

    assert user["name"]  == "Alice"
    assert user["email"] == "[email protected]"
    mock_db.save_user.assert_called_once_with(user)
    mock_email.send_welcome.assert_called_once_with("[email protected]")

def test_register_duplicate_raises():
    mock_db    = MagicMock()
    mock_email = MagicMock()
    mock_db.user_exists.return_value = True

    service = UserService(mock_db, mock_email)
    with pytest.raises(ValueError, match="already exists"):
        service.register("Alice", "[email protected]")

    mock_db.save_user.assert_not_called()
    mock_email.send_welcome.assert_not_called()

def test_get_user_not_found():
    mock_db = MagicMock()
    mock_db.get_user.return_value = None
    service = UserService(mock_db, MagicMock())
    with pytest.raises(KeyError):
        service.get_user(999)

# Install: pip install pytest-cov

# Run tests with coverage
# pytest --cov=mymodule tests/
# pytest --cov=mymodule --cov-report=term-missing tests/
# pytest --cov=mymodule --cov-report=html tests/   (HTML report)

# Example output:
# Name              Stmts   Miss  Cover   Missing
# -----------------------------------------------
# mymodule.py          45      5    89%   23-25, 41, 67
# -----------------------------------------------
# TOTAL                45      5    89%

# .coveragerc — configure coverage
# [run]
# source = mypackage
# omit = tests/*, setup.py
#
# [report]
# fail_under = 80    ← fail if coverage < 80%
# show_missing = True

# What coverage tells you:
# 100% coverage ≠ bug-free code
# But < 80% coverage = likely untested edge cases

# Example: code with branches
def classify_age(age):
    if age < 0:       return "invalid"    # branch 1
    elif age < 13:    return "child"      # branch 2
    elif age < 18:    return "teen"       # branch 3
    elif age < 65:    return "adult"      # branch 4
    else:             return "senior"     # branch 5

# To get 100% branch coverage, test ALL 5 branches:
# classify_age(-1), classify_age(5), classify_age(15),
# classify_age(30), classify_age(70)

# Method 1: breakpoint() — Python 3.7+ (recommended)
def buggy_function(data):
    result = []
    for item in data:
        breakpoint()          # execution pauses here
        processed = item * 2
        result.append(processed)
    return result

# Method 2: import pdb
import pdb
def another_function(x):
    pdb.set_trace()           # old-style breakpoint
    return x + 1

# pdb commands:
# n  (next)      — execute next line (don't step into calls)
# s  (step)      — step into function calls
# c  (continue)  — run until next breakpoint
# q  (quit)      — exit debugger
# l  (list)      — show current code
# p expr         — print expression value
# pp expr        — pretty-print expression
# w  (where)     — show call stack
# u  (up)        — go up one frame in stack
# d  (down)      — go down one frame in stack
# b 42           — set breakpoint at line 42
# b func_name    — set breakpoint at function
# cl             — clear all breakpoints
# h  (help)      — show all commands

# Post-mortem debugging — debug after a crash
import pdb, traceback, sys

def run_with_debugger():
    try:
        buggy_function([1, 2, "three"])
    except Exception:
        traceback.print_exc()
        pdb.post_mortem()     # open debugger at crash point

import logging

# Log levels (lowest to highest):
# DEBUG    — detailed diagnostic info
# INFO     — confirmation things work
# WARNING  — something unexpected (default threshold)
# ERROR    — serious problem
# CRITICAL — program may not continue

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

logger = logging.getLogger(__name__)

logger.debug("Debug message — detailed info")
logger.info("Info message — normal operation")
logger.warning("Warning — something unexpected")
logger.error("Error — something went wrong")
logger.critical("Critical — system failure!")

# Log exceptions with traceback
try:
    result = 1 / 0
except ZeroDivisionError:
    logger.exception("Caught an exception!")   # includes traceback

# Log with extra data
logger.info("User logged in", extra={"user_id": 42, "ip": "192.168.1.1"})

import logging
import logging.handlers
import sys

def setup_logger(name, log_file=None, level=logging.DEBUG):
    """Create a configured logger with console and file handlers."""
    logger    = logging.getLogger(name)
    logger.setLevel(level)
    formatter = logging.Formatter(
        "%(asctime)s | %(levelname)-8s | %(name)s:%(lineno)d | %(message)s",
        datefmt="%H:%M:%S"
    )

    # Console handler — INFO and above
    console = logging.StreamHandler(sys.stdout)
    console.setLevel(logging.INFO)
    console.setFormatter(formatter)
    logger.addHandler(console)

    # File handler — DEBUG and above
    if log_file:
        file_handler = logging.FileHandler(log_file, encoding="utf-8")
        file_handler.setLevel(logging.DEBUG)
        file_handler.setFormatter(formatter)
        logger.addHandler(file_handler)

    # Rotating file handler — max 5MB, keep 3 backups
    if log_file:
        rotating = logging.handlers.RotatingFileHandler(
            log_file + ".rotating",
            maxBytes=5 * 1024 * 1024,
            backupCount=3
        )
        rotating.setFormatter(formatter)
        logger.addHandler(rotating)

    return logger

log = setup_logger("myapp", "app.log")
log.debug("Starting application")
log.info("Server listening on port 8080")
log.warning("Config file not found, using defaults")
log.error("Failed to connect to database")

# Logger hierarchy — child loggers inherit parent config
parent = logging.getLogger("myapp")
child  = logging.getLogger("myapp.database")
child.info("This also goes to myapp handlers")

# Type hints — document expected types (Python 3.5+)
# Install mypy: pip install mypy
# Run:          mypy myfile.py

from typing import List, Dict, Optional, Tuple, Union, Any

# Basic type hints
def greet(name: str) -> str:
    return f"Hello, {name}!"

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

# Optional — value or None
def find_user(user_id: int) -> Optional[Dict[str, Any]]:
    users = {1: {"name": "Alice"}}
    return users.get(user_id)

# List, Dict, Tuple
def process_scores(scores: List[float]) -> Dict[str, float]:
    return {
        "mean": sum(scores) / len(scores),
        "max":  max(scores),
        "min":  min(scores),
    }

# Union — one of several types
def stringify(value: Union[int, float, str]) -> str:
    return str(value)

# Python 3.10+ — cleaner union syntax
def new_stringify(value: int | float | str) -> str:
    return str(value)

# Class with type hints
class Stack:
    def __init__(self) -> None:
        self._items: List[Any] = []

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

    def pop(self) -> Any:
        if not self._items:
            raise IndexError("Stack is empty")
        return self._items.pop()

    def peek(self) -> Optional[Any]:
        return self._items[-1] if self._items else None

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

# ── LINTING ──────────────────────────────────────────
# flake8  — style checker (PEP 8 compliance)
#   pip install flake8
#   flake8 myfile.py
#   Common errors:
#   E501 line too long (> 79 chars)
#   E302 expected 2 blank lines
#   W291 trailing whitespace
#   F401 imported but unused

# pylint — comprehensive linter
#   pip install pylint
#   pylint myfile.py
#   Gives a score out of 10

# ruff — extremely fast linter (replaces flake8 + isort)
#   pip install ruff
#   ruff check myfile.py
#   ruff check --fix myfile.py

# ── FORMATTING ───────────────────────────────────────
# black — opinionated auto-formatter
#   pip install black
#   black myfile.py        (format in place)
#   black --check myfile.py (check without changing)

# isort — sort imports automatically
#   pip install isort
#   isort myfile.py

# ── TYPE CHECKING ────────────────────────────────────
# mypy — static type checker
#   pip install mypy
#   mypy myfile.py

# ── ALL-IN-ONE CONFIG ────────────────────────────────
# pyproject.toml
# [tool.black]
# line-length = 88
#
# [tool.isort]
# profile = "black"
#
# [tool.mypy]
# strict = true
#
# [tool.ruff]
# line-length = 88
# select = ["E", "F", "W", "I"]

import pytest

# TDD cycle:
# 1. RED   — write a failing test
# 2. GREEN — write minimum code to pass
# 3. REFACTOR — improve code, keep tests green

# Step 1: RED — write tests FIRST (code doesn't exist yet)
class TestShoppingCart:

    def test_empty_cart_total(self):
        cart = ShoppingCart()
        assert cart.total() == 0.0

    def test_add_item(self):
        cart = ShoppingCart()
        cart.add("Apple", 1.50)
        assert cart.total() == 1.50

    def test_add_multiple_items(self):
        cart = ShoppingCart()
        cart.add("Apple",  1.50)
        cart.add("Banana", 0.75)
        assert cart.total() == 2.25

    def test_add_item_with_quantity(self):
        cart = ShoppingCart()
        cart.add("Apple", 1.50, qty=3)
        assert cart.total() == 4.50

    def test_remove_item(self):
        cart = ShoppingCart()
        cart.add("Apple", 1.50)
        cart.remove("Apple")
        assert cart.total() == 0.0

    def test_apply_discount(self):
        cart = ShoppingCart()
        cart.add("Apple", 10.00)
        cart.apply_discount(10)    # 10% off
        assert cart.total() == pytest.approx(9.00)

# Step 2: GREEN — implement minimum code to pass
class ShoppingCart:
    def __init__(self):
        self._items    = {}
        self._discount = 0

    def add(self, name, price, qty=1):
        self._items[name] = {"price": price, "qty": qty}

    def remove(self, name):
        self._items.pop(name, None)

    def apply_discount(self, percent):
        self._discount = percent

    def total(self):
        subtotal = sum(i["price"] * i["qty"] for i in self._items.values())
        return subtotal * (1 - self._discount / 100)

# Step 3: REFACTOR — improve, add validation, etc.

# File modes:
# "r"  — read text (default)
# "w"  — write text (overwrites)
# "a"  — append text
# "x"  — exclusive create (fails if exists)
# "rb" — read binary
# "wb" — write binary
# "r+" — read and write

# Always use with — auto-closes the file
with open("data.txt", "r", encoding="utf-8") as f:
    content = f.read()          # read entire file as string
print(content)

# Read line by line — memory efficient for large files
with open("data.txt", "r", encoding="utf-8") as f:
    for line in f:
        print(line.strip())     # strip removes \n

# Read all lines into a list
with open("data.txt", "r", encoding="utf-8") as f:
    lines = f.readlines()       # ["line1\n", "line2\n", ...]

# Read single line
with open("data.txt", "r", encoding="utf-8") as f:
    first_line = f.readline()

# Read in chunks — for very large files
with open("bigfile.txt", "r", encoding="utf-8") as f:
    while chunk := f.read(8192):   # 8KB chunks
        pass   # process(chunk)

# File position
with open("data.txt", "r", encoding="utf-8") as f:
    f.seek(0)           # go to start
    pos = f.tell()      # current position in bytes
    print(pos)          # 0

# Write — creates or overwrites
with open("output.txt", "w", encoding="utf-8") as f:
    f.write("Hello, World!\n")
    f.write("Second line\n")

# writelines — write a list of strings (no auto newline)
lines = ["Line 1\n", "Line 2\n", "Line 3\n"]
with open("output.txt", "w", encoding="utf-8") as f:
    f.writelines(lines)

# Append — adds to end of existing file
with open("log.txt", "a", encoding="utf-8") as f:
    f.write("New log entry\n")

# Write with print() — convenient for formatted output
with open("report.txt", "w", encoding="utf-8") as f:
    print("=== Report ===", file=f)
    print(f"Total: {42}", file=f)

# Atomic write — write to temp, rename on success
import os, tempfile

def atomic_write(filepath, content, encoding="utf-8"):
    """Write content atomically — never leaves partial file."""
    dir_name = os.path.dirname(filepath) or "."
    with tempfile.NamedTemporaryFile(
        mode="w", dir=dir_name,
        encoding=encoding, delete=False
    ) as tmp:
        tmp.write(content)
        tmp_path = tmp.name
    os.replace(tmp_path, filepath)   # atomic rename

atomic_write("config.txt", "key=value\n")

from pathlib import Path

# Create paths — OS-independent (uses / on all platforms)
p = Path("/home/user/documents/report.txt")
p = Path.home() / "documents" / "report.txt"

# Path components
print(p.name)       # "report.txt"
print(p.stem)       # "report"
print(p.suffix)     # ".txt"
print(p.parent)     # /home/user/documents
print(p.parts)      # ('/', 'home', 'user', 'documents', 'report.txt')

# Check existence
print(p.exists())   # True/False
print(p.is_file())  # True if file
print(p.is_dir())   # True if directory

# Read and write — built into Path!
p = Path("data.txt")
p.write_text("Hello from pathlib!\n", encoding="utf-8")
content = p.read_text(encoding="utf-8")
print(content)

# Binary
p.write_bytes(b"\x00\x01\x02")
data = p.read_bytes()

# Directory operations
d = Path("my_project")
d.mkdir(exist_ok=True)
d.mkdir(parents=True, exist_ok=True)

# List directory contents
for item in Path(".").iterdir():
    print(item)

# Glob — find files by pattern
for py_file in Path(".").glob("*.py"):
    print(py_file)

for py_file in Path(".").rglob("*.py"):   # recursive
    print(py_file)

# File info
stat = p.stat()
print(stat.st_size)    # size in bytes
print(stat.st_mtime)   # modification time

from pathlib import Path
import shutil

# Rename / move
# src.rename(dst)              # rename in same dir
# src.replace(dst)             # replace even if dst exists

# Copy (use shutil)
# shutil.copy2(src, dst)       # copy file with metadata
# shutil.copytree(src, dst)    # copy directory tree

# Delete
# p.unlink()                   # delete file
# p.unlink(missing_ok=True)    # no error if missing
# d.rmdir()                    # delete empty directory
# shutil.rmtree(d)             # delete directory tree

# Resolve — get absolute path
rel      = Path("../data/file.txt")
abs_path = rel.resolve()

# Change suffix / name
p     = Path("report.txt")
new_p = p.with_suffix(".md")         # report.md
new_p = p.with_name("summary.txt")  # summary.txt

# Find all large Python files
def find_large_files(directory, extension="*.py", min_kb=10):
    results = []
    for f in Path(directory).rglob(extension):
        size_kb = f.stat().st_size / 1024
        if size_kb >= min_kb:
            results.append((f, size_kb))
    return sorted(results, key=lambda x: -x[1])

# Ensure directory structure exists
def ensure_dirs(*paths):
    for p in paths:
        Path(p).mkdir(parents=True, exist_ok=True)

ensure_dirs("logs", "data/raw", "data/processed", "output")

import csv

# Read as dicts — header becomes keys (recommended)
with open("employees.csv", "r", encoding="utf-8", newline="") as f:
    reader    = csv.DictReader(f)
    employees = list(reader)

for emp in employees:
    print(f"{emp['name']}: ${int(emp['salary']):,}")

# Read as rows (list of lists)
with open("employees.csv", "r", encoding="utf-8", newline="") as f:
    reader = csv.reader(f)
    header = next(reader)          # skip header row
    for row in reader:
        name, age, city, salary = row
        print(f"{name} ({age}) — {city}: ${int(salary):,}")

# Handle different delimiters
with open("data.tsv", "r", encoding="utf-8", newline="") as f:
    reader = csv.reader(f, delimiter="\t")   # tab-separated
    for row in reader:
        print(row)

# csv.Sniffer — auto-detect dialect
with open("unknown.csv", "r", encoding="utf-8") as f:
    sample  = f.read(1024)
    dialect = csv.Sniffer().sniff(sample)
    f.seek(0)
    reader  = csv.reader(f, dialect)

# Compute stats from CSV
salaries = [int(e["salary"]) for e in employees]
print(f"Average salary: ${sum(salaries)/len(salaries):,.0f}")

import csv, io

employees = [
    {"name": "Alice", "age": 28, "city": "London", "salary": 95000},
    {"name": "Bob",   "age": 35, "city": "Paris",  "salary": 72000},
    {"name": "Carol", "age": 31, "city": "Berlin", "salary": 88000},
]

# Write with DictWriter (recommended)
fieldnames = ["name", "age", "city", "salary"]
with open("output.csv", "w", encoding="utf-8", newline="") as f:
    writer = csv.DictWriter(f, fieldnames=fieldnames)
    writer.writeheader()
    writer.writerows(employees)

# Write with writer (list of lists)
rows = [["name", "age"], ["Alice", 28], ["Bob", 35]]
with open("output2.csv", "w", encoding="utf-8", newline="") as f:
    writer = csv.writer(f, quoting=csv.QUOTE_NONNUMERIC)
    writer.writerows(rows)

# Write to string buffer (no file needed)
buffer = io.StringIO()
writer = csv.DictWriter(buffer, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(employees)
csv_string = buffer.getvalue()
print(csv_string[:120])

import json

data = {
    "name":    "Alice",
    "age":     28,
    "active":  True,
    "scores":  [95, 87, 92],
    "address": {"city": "London", "zip": "EC1A"}
}

# Python → JSON string
json_str = json.dumps(data, indent=2, sort_keys=True)
print(json_str)

# JSON string → Python
parsed = json.loads(json_str)
print(parsed["name"])    # Alice

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

# Read JSON from file
with open("data.json", "r", encoding="utf-8") as f:
    loaded = json.load(f)

# JSON type mapping:
# Python dict   ↔  JSON object  {}
# Python list   ↔  JSON array   []
# Python str    ↔  JSON string  ""
# Python int    ↔  JSON number  42
# Python float  ↔  JSON number  3.14
# Python True   ↔  JSON true
# Python False  ↔  JSON false
# Python None   ↔  JSON null

import json
from datetime import datetime
from decimal import Decimal
from dataclasses import dataclass, asdict

class AppJSONEncoder(json.JSONEncoder):
    """Handle types JSON doesn't natively support."""
    def default(self, obj):
        if isinstance(obj, datetime):
            return obj.isoformat()
        if isinstance(obj, Decimal):
            return float(obj)
        if isinstance(obj, set):
            return sorted(list(obj))
        if hasattr(obj, "__dict__"):
            return obj.__dict__
        return super().default(obj)

data = {
    "created_at": datetime.now(),
    "price":      Decimal("19.99"),
    "tags":       {"python", "coding", "tutorial"},
}
json_str = json.dumps(data, cls=AppJSONEncoder, indent=2)
print(json_str)

# Custom decoder — restore Python objects
def decode_dates(dct):
    for key, value in dct.items():
        if isinstance(value, str):
            try:
                dct[key] = datetime.fromisoformat(value)
            except ValueError:
                pass
    return dct

restored = json.loads(json_str, object_hook=decode_dates)

# Dataclass to JSON
@dataclass
class Product:
    name:  str
    price: float
    stock: int

p = Product("Widget", 9.99, 100)
print(json.dumps(asdict(p), indent=2))

import xml.etree.ElementTree as ET

xml_data = """

    
        
        F. Scott Fitzgerald
        1925
        12.99
    
    
        
        Frank Herbert
        1965
        14.99
    
"""

root = ET.fromstring(xml_data)
print(root.tag)   # library

# Iterate children
for book in root.findall("book"):
    book_id = book.get("id")
    title   = book.find("title").text
    author  = book.find("author").text
    price   = float(book.find("price").text)
    print(f"[{book_id}] {title} by {author} — ${price:.2f}")

# XPath-like queries
titles = root.findall(".//title")
for t in titles:
    print(t.text)

# Find by attribute
first = root.find("book[@id='1']")
print(first.find("title").text)   # The Great Gatsby

import xml.etree.ElementTree as ET

products = [
    {"id": "P001", "name": "Widget",   "price": 9.99,  "stock": 100},
    {"id": "P002", "name": "Gadget",   "price": 24.99, "stock": 50},
    {"id": "P003", "name": "Doohickey","price": 4.99,  "stock": 200},
]

root = ET.Element("catalog")
root.set("version", "1.0")

for p in products:
    item = ET.SubElement(root, "product")
    item.set("id", p["id"])
    for field in ("name", "price", "stock"):
        el      = ET.SubElement(item, field)
        el.text = str(p[field])

# Pretty-print (Python 3.9+)
ET.indent(root, space="  ")
tree = ET.ElementTree(root)
tree.write("catalog.xml", encoding="unicode", xml_declaration=True)

# Print to string
xml_str = ET.tostring(root, encoding="unicode")
print(xml_str[:200])

import pickle, struct

# pickle — serialize ANY Python object to binary
data = {
    "model":   "LinearRegression",
    "weights": [0.5, 1.2, -0.3, 0.8],
    "bias":    0.1,
    "trained": True
}

# Save to file
with open("model.pkl", "wb") as f:
    pickle.dump(data, f)

# Load from file
with open("model.pkl", "rb") as f:
    loaded = pickle.load(f)
print(loaded["model"])   # LinearRegression

# Pickle to/from bytes (no file)
serialized   = pickle.dumps(data)
deserialized = pickle.loads(serialized)

# ⚠️ WARNING: Never unpickle data from untrusted sources!
# pickle can execute arbitrary code on load.

# struct — pack/unpack fixed-format binary data
packed = struct.pack("if2s", 42, 3.14, b"AB")
print(f"Packed: {len(packed)} bytes")

n, f_val, chars = struct.unpack("if2s", packed)
print(n, f_val, chars)   # 42  3.14  b'AB'

# Read binary file header (e.g. PNG dimensions)
def read_png_dimensions(filepath):
    with open(filepath, "rb") as f:
        f.seek(16)   # PNG width/height at byte 16
        width, height = struct.unpack(">II", f.read(8))
    return width, height

import zipfile, gzip, shutil, json
from pathlib import Path

# ── ZIP files ─────────────────────────────────────────
# Create a ZIP archive
with zipfile.ZipFile("archive.zip", "w",
                      compression=zipfile.ZIP_DEFLATED) as zf:
    zf.write("data.txt")
    zf.write("report.csv")
    zf.write("output.json", arcname="results/output.json")

# Read a ZIP archive
with zipfile.ZipFile("archive.zip", "r") as zf:
    print(zf.namelist())         # list all files
    zf.extractall("extracted/")  # extract all

    # Read without extracting
    with zf.open("data.txt") as f:
        content = f.read().decode("utf-8")

# ── GZIP ──────────────────────────────────────────────
# Compress a file
with open("data.txt", "rb") as f_in:
    with gzip.open("data.txt.gz", "wb") as f_out:
        shutil.copyfileobj(f_in, f_out)

# Read gzip directly
with gzip.open("data.txt.gz", "rt", encoding="utf-8") as f:
    content = f.read()

# Write compressed JSON
data = {"records": list(range(10000))}
with gzip.open("data.json.gz", "wt", encoding="utf-8") as f:
    json.dump(data, f)

# shutil — high-level archive operations
shutil.make_archive("backup", "zip", "my_project/")
shutil.unpack_archive("backup.zip", "restored/")

import tempfile
from pathlib import Path

# NamedTemporaryFile — auto-deleted on close
with tempfile.NamedTemporaryFile(
    mode="w", suffix=".txt",
    encoding="utf-8", delete=True
) as tmp:
    tmp.write("Temporary content\n")
    print(f"Temp file: {tmp.name}")
# File is deleted here

# TemporaryDirectory — auto-deleted on exit
with tempfile.TemporaryDirectory() as tmp_dir:
    tmp_path = Path(tmp_dir)
    (tmp_path / "file1.txt").write_text("Hello")
    (tmp_path / "file2.txt").write_text("World")
    files = list(tmp_path.iterdir())
    print(f"Temp files: {[f.name for f in files]}")
# Directory and all contents deleted here

# SpooledTemporaryFile — in memory until size limit
with tempfile.SpooledTemporaryFile(
    max_size=1024 * 1024,   # 1MB in memory, then disk
    mode="w"
) as tmp:
    tmp.write("In-memory content")
    tmp.seek(0)
    print(tmp.read())

# Get system temp directory
print(tempfile.gettempdir())   # /tmp on Linux, %TEMP% on Windows

import csv, json
from pathlib import Path
from datetime import datetime

def csv_to_json(csv_path, json_path, type_map=None):
    """Convert CSV to JSON with optional type conversion."""
    type_map = type_map or {}
    records  = []

    with open(csv_path, "r", encoding="utf-8", newline="") as f:
        for row in csv.DictReader(f):
            record = {}
            for key, value in row.items():
                if key in type_map:
                    try:
                        record[key] = type_map[key](value)
                    except (ValueError, TypeError):
                        record[key] = value
                else:
                    record[key] = value
            records.append(record)

    output = {
        "generated_at": datetime.now().isoformat(),
        "count":        len(records),
        "records":      records
    }

    Path(json_path).parent.mkdir(parents=True, exist_ok=True)
    with open(json_path, "w", encoding="utf-8") as f:
        json.dump(output, f, indent=2, ensure_ascii=False)

    return len(records)

def merge_json_files(input_dir, output_file):
    """Merge all JSON files in a directory into one."""
    all_records = []
    for json_file in sorted(Path(input_dir).glob("*.json")):
        with open(json_file, "r", encoding="utf-8") as f:
            data = json.load(f)
            if isinstance(data, list):
                all_records.extend(data)
            else:
                all_records.append(data)
    with open(output_file, "w", encoding="utf-8") as f:
        json.dump(all_records, f, indent=2)
    return len(all_records)

import re
from collections import Counter

def analyze_log_file(log_path):
    """Parse and analyze an Apache/Nginx access log."""
    pattern = re.compile(
        r'(?P\S+) \S+ \S+ \[(?P[^\]]+)\] '
        r'"(?P\S+) (?P\S+) \S+" '
        r'(?P\d{3}) (?P\S+)'
    )
    stats = {
        "total_requests": 0,
        "status_codes":   Counter(),
        "top_paths":      Counter(),
        "top_ips":        Counter(),
        "total_bytes":    0,
        "errors":         [],
    }
    with open(log_path, "r", encoding="utf-8") as f:
        for line in f:
            m = pattern.match(line.strip())
            if not m: continue
            stats["total_requests"] += 1
            status = int(m.group("status"))
            stats["status_codes"][status] += 1
            stats["top_paths"][m.group("path")] += 1
            stats["top_ips"][m.group("ip")] += 1
            size = m.group("size")
            if size.isdigit():
                stats["total_bytes"] += int(size)
            if status >= 400:
                stats["errors"].append({
                    "ip": m.group("ip"),
                    "path": m.group("path"),
                    "status": status
                })
    return stats

def print_log_report(stats):
    print(f"Total requests: {stats['total_requests']:,}")
    print(f"Total data:     {stats['total_bytes']/1024/1024:.1f} MB")
    print(f"\nTop 5 paths:")
    for path, count in stats["top_paths"].most_common(5):
        print(f"  {count:>6,}  {path}")
    print(f"\nStatus codes:")
    for code, count in sorted(stats["status_codes"].items()):
        print(f"  {code}: {count:,}")

# File: mathutils.py  ← this IS the module
PI = 3.14159265358979

def add(a, b):
    return a + b

def multiply(a, b):
    return a * b

def circle_area(radius):
    return PI * radius ** 2

class Vector:
    def __init__(self, x, y):
        self.x, self.y = x, y
    def __repr__(self):
        return f"Vector({self.x}, {self.y})"
    def magnitude(self):
        return (self.x**2 + self.y**2) ** 0.5

# ── In another file: main.py ──────────────────────────

# Import entire module
import mathutils
print(mathutils.PI)
print(mathutils.add(3, 4))
v = mathutils.Vector(3, 4)

# Import specific names
from mathutils import add, PI, Vector
print(add(3, 4))
print(PI)

# Import with alias
import mathutils as mu
print(mu.circle_area(5))

from mathutils import circle_area as area
print(area(5))

# Import everything (avoid in production — pollutes namespace)
from mathutils import *

# File: calculator.py

def add(a, b):      return a + b
def subtract(a, b): return a - b
def multiply(a, b): return a * b
def divide(a, b):
    if b == 0: raise ZeroDivisionError("Cannot divide by zero")
    return a / b

# This block ONLY runs when you do: python calculator.py
# It does NOT run when you do: import calculator
if __name__ == "__main__":
    print("Running calculator directly")
    print(f"add(3, 4)      = {add(3, 4)}")
    print(f"multiply(3, 4) = {multiply(3, 4)}")
    print(f"divide(10, 2)  = {divide(10, 2)}")

# ── Why this matters ─────────────────────────────────
# Without the guard, importing calculator.py would
# immediately print all those lines — unwanted side effect!

# Common pattern — module with CLI entry point
def main():
    import sys
    if len(sys.argv) != 3:
        print("Usage: python calculator.py  ")
        sys.exit(1)
    a, b = float(sys.argv[1]), float(sys.argv[2])
    print(f"Sum: {add(a, b)}")

if __name__ == "__main__":
    main()

import sys

# Python searches for modules in this order:
# 1. sys.modules cache (already imported modules)
# 2. Built-in modules (math, os, sys, ...)
# 3. Frozen modules
# 4. sys.path directories (in order)

# View the search path
for path in sys.path:
    print(path)
# Typically includes:
# '' or '.'          ← current directory (FIRST)
# PYTHONPATH dirs    ← env variable
# stdlib dirs        ← standard library
# site-packages      ← installed packages

# Add a directory to the path at runtime
sys.path.insert(0, "/path/to/my/modules")
sys.path.append("/another/path")

# Check if a module is cached
print("math" in sys.modules)   # True after import math

# Reload a module (useful during development)
import importlib
import mathutils
importlib.reload(mathutils)

# Find where a module lives
import os
print(os.__file__)    # /usr/lib/python3.x/os.py

import math
print(math.__file__)  # .../lib-dynload/math.cpython-3x.so

# A package is a directory with __init__.py:
#
# mypackage/
# ├── __init__.py          ← makes it a package
# ├── core.py
# ├── utils.py
# └── models/
#     ├── __init__.py      ← nested package
#     ├── user.py
#     └── product.py

# ── mypackage/__init__.py ────────────────────────────
"""MyPackage — a demonstration package."""

__version__ = "1.0.0"
__author__  = "Alice Smith"

# Control what 'from mypackage import *' exposes
__all__ = ["core", "utils"]

# Make key items available at package level
from .core import process, validate
from .utils import format_output

# ── mypackage/core.py ────────────────────────────────
def process(data):
    return f"Processed: {data}"

def validate(data):
    return bool(data)

# ── mypackage/utils.py ───────────────────────────────
def format_output(result):
    return f"[OUTPUT] {result}"

# ── Usage ────────────────────────────────────────────
import mypackage
print(mypackage.__version__)   # 1.0.0
print(mypackage.process("hello"))

from mypackage import utils
print(utils.format_output("test"))

from mypackage.models import user
from mypackage.models.product import Product

# Package structure:
# myapp/
# ├── __init__.py
# ├── main.py
# ├── config.py
# └── services/
#     ├── __init__.py
#     ├── auth.py
#     └── email.py

# ── ABSOLUTE imports (recommended) ───────────────────
# Always specify the full path from the package root
# Works from anywhere, always unambiguous

# In myapp/services/auth.py:
from myapp.config import Settings         # full path
from myapp.services.email import send     # full path

# ── RELATIVE imports (within a package) ──────────────
# Use dots to navigate relative to current file
# . = current package
# .. = parent package
# ... = grandparent package

# In myapp/services/auth.py:
from ..config import Settings             # go up one level
from .email import send                   # same package

# In myapp/services/email.py:
from . import auth                        # import sibling module
from .. import main                       # import from parent

# ── Rules ────────────────────────────────────────────
# Use ABSOLUTE imports for:
#   - Top-level scripts
#   - Cross-package imports
#   - Better readability
#
# Use RELATIVE imports for:
#   - Within the same package
#   - When renaming/moving the package
#   - Avoiding circular imports

# ── Create a virtual environment ─────────────────────
# python -m venv venv          ← create in ./venv/
# python -m venv .venv         ← hidden dir (common)
# python3.11 -m venv venv      ← specific Python version

# ── Activate ─────────────────────────────────────────
# Windows:
#   venv\Scripts\activate
#   venv\Scripts\Activate.ps1  (PowerShell)
#
# macOS / Linux:
#   source venv/bin/activate
#
# Prompt changes to: (venv) $

# ── Deactivate ────────────────────────────────────────
# deactivate

# ── Install packages ──────────────────────────────────
# pip install requests
# pip install "django>=4.0,<5.0"
# pip install -r requirements.txt

# ── Freeze dependencies ───────────────────────────────
# pip freeze > requirements.txt
# pip freeze | grep -v "^-e" > requirements.txt

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

# ── Useful pip commands ───────────────────────────────
# pip list                     ← list installed packages
# pip show requests            ← info about a package
# pip install --upgrade pip    ← upgrade pip itself
# pip uninstall requests       ← remove a package
# pip check                    ← check for conflicts

# ── Always add to .gitignore ──────────────────────────
# venv/
# .venv/
# __pycache__/
# *.pyc
# *.egg-info/
# dist/
# build/

# ── requirements.txt ─────────────────────────────────
# Pinned versions (reproducible builds):
# requests==2.31.0
# numpy==1.26.0
# pandas==2.1.0
#
# Flexible versions (development):
# requests>=2.28.0
# numpy>=1.24.0,<2.0.0
#
# Install from git:
# git+https://github.com/user/repo.git@main
#
# Editable install (local development):
# -e .

# ── requirements split by environment ─────────────────
# requirements/
# ├── base.txt        ← shared dependencies
# ├── dev.txt         ← development only
# └── prod.txt        ← production only
#
# dev.txt:
# -r base.txt
# pytest>=7.0
# black>=23.0
# mypy>=1.0
#
# prod.txt:
# -r base.txt
# gunicorn>=21.0

# ── pip programmatic usage ────────────────────────────
import subprocess, sys

def install_package(package):
    subprocess.check_call([sys.executable, "-m", "pip",
                           "install", package])

def get_installed_packages():
    import importlib.metadata
    return {
        dist.name: dist.version
        for dist in importlib.metadata.distributions()
    }

packages = get_installed_packages()
print(f"Installed packages: {len(packages)}")
if "requests" in packages:
    print(f"requests version: {packages['requests']}")

# Modern Python project structure (src layout):
#
# my_project/
# ├── src/
# │   └── mypackage/
# │       ├── __init__.py
# │       ├── core.py
# │       ├── models/
# │       │   ├── __init__.py
# │       │   └── user.py
# │       └── utils/
# │           ├── __init__.py
# │           └── helpers.py
# ├── tests/
# │   ├── __init__.py
# │   ├── test_core.py
# │   └── test_models/
# │       └── test_user.py
# ├── docs/
# │   └── index.md
# ├── .github/
# │   └── workflows/
# │       └── ci.yml
# ├── .gitignore
# ├── pyproject.toml       ← modern config (replaces setup.py)
# ├── README.md
# ├── LICENSE
# └── CHANGELOG.md
#
# Flat layout (simpler, for smaller projects):
#
# my_project/
# ├── mypackage/
# │   ├── __init__.py
# │   └── core.py
# ├── tests/
# │   └── test_core.py
# ├── pyproject.toml
# └── README.md

# pyproject.toml

[build-system]
requires      = ["hatchling"]
build-backend = "hatchling.build"

[project]
name        = "mypackage"
version     = "1.0.0"
description = "A demonstration Python package"
readme      = "README.md"
license     = { file = "LICENSE" }
authors     = [{ name = "Alice Smith", email = "[email protected]" }]
keywords    = ["python", "example", "package"]
classifiers = [
    "Development Status :: 4 - Beta",
    "Programming Language :: Python :: 3",
    "Programming Language :: Python :: 3.10",
    "Programming Language :: Python :: 3.11",
    "License :: OSI Approved :: MIT License",
    "Operating System :: OS Independent",
]
requires-python = ">=3.10"
dependencies    = [
    "requests>=2.28.0",
    "click>=8.0.0",
]

[project.optional-dependencies]
dev  = ["pytest>=7.0", "black>=23.0", "mypy>=1.0", "ruff>=0.1"]
docs = ["mkdocs>=1.5", "mkdocs-material>=9.0"]

[project.urls]
Homepage      = "https://github.com/alice/mypackage"
Documentation = "https://mypackage.readthedocs.io"
Repository    = "https://github.com/alice/mypackage"
"Bug Tracker" = "https://github.com/alice/mypackage/issues"

[project.scripts]
mypackage = "mypackage.cli:main"   # creates CLI command

[tool.pytest.ini_options]
testpaths    = ["tests"]
addopts      = "-v --tb=short"

[tool.black]
line-length    = 88
target-version = ["py310", "py311"]

[tool.ruff]
line-length = 88
select      = ["E", "F", "W", "I", "N"]

[tool.mypy]
python_version = "3.11"
strict         = true

# ── Install build tools ───────────────────────────────
# pip install build twine

# ── Build the package ─────────────────────────────────
# python -m build
#
# Creates:
# dist/
# ├── mypackage-1.0.0.tar.gz       ← source distribution
# └── mypackage-1.0.0-py3-none-any.whl  ← wheel (binary)

# ── Test on TestPyPI first ────────────────────────────
# twine upload --repository testpypi dist/*
# pip install --index-url https://test.pypi.org/simple/ mypackage

# ── Publish to PyPI ───────────────────────────────────
# twine upload dist/*
# (requires PyPI account and API token)

# ── Install your published package ───────────────────
# pip install mypackage

# ── Semantic versioning ───────────────────────────────
# MAJOR.MINOR.PATCH
# 1.0.0  → initial release
# 1.0.1  → bug fix (patch)
# 1.1.0  → new feature, backward compatible (minor)
# 2.0.0  → breaking change (major)

# ── Version in code ───────────────────────────────────
# mypackage/__init__.py
__version__ = "1.0.0"

# Access version from metadata (after install)
from importlib.metadata import version
ver = version("mypackage")
print(ver)   # 1.0.0

# uv — written in Rust, 10-100x faster than pip
# Install: curl -LsSf https://astral.sh/uv/install.sh | sh

# Create project
# uv init myproject
# cd myproject

# Create virtual environment
# uv venv

# Add dependencies (updates pyproject.toml automatically)
# uv add requests
# uv add "django>=4.0"
# uv add --dev pytest black mypy

# Install all dependencies
# uv sync

# Run commands in the venv
# uv run python main.py
# uv run pytest

# Lock file for reproducible installs
# uv lock        ← creates uv.lock
# uv sync        ← installs from lock file

# ── Poetry — alternative tool ─────────────────────────
# Install: pip install poetry

# Create project
# poetry new myproject
# poetry init    ← interactive setup

# Add dependencies
# poetry add requests
# poetry add --group dev pytest

# Install
# poetry install

# Run
# poetry run python main.py
# poetry run pytest

# Build and publish
# poetry build
# poetry publish

import os, sys, math, random, datetime
import collections, itertools, functools
import re, json, csv, pathlib
import threading, multiprocessing, asyncio
import unittest, logging, argparse
import hashlib, hmac, secrets
import urllib.request, urllib.parse
import sqlite3, pickle, shelve
import time, timeit, cProfile

# ── os — operating system interface ──────────────────
print(os.getcwd())                # current directory
print(os.environ.get("HOME"))     # env variable
os.makedirs("a/b/c", exist_ok=True)
print(os.listdir("."))            # list directory

# ── sys — interpreter info ────────────────────────────
print(sys.version)                # Python version
print(sys.platform)               # 'linux', 'win32', 'darwin'
print(sys.argv)                   # command-line arguments
sys.exit(0)                       # exit with code

# ── collections — specialised containers ─────────────
from collections import Counter, defaultdict, OrderedDict, deque, namedtuple

words   = ["apple", "banana", "apple", "cherry", "banana", "apple"]
counter = Counter(words)
print(counter.most_common(2))     # [('apple', 3), ('banana', 2)]

dd = defaultdict(list)
dd["fruits"].append("apple")

Point = namedtuple("Point", ["x", "y"])
p = Point(3, 4)
print(p.x, p.y)

# ── hashlib — cryptographic hashing ──────────────────
import hashlib
h = hashlib.sha256(b"Hello, World!").hexdigest()
print(h[:16])   # first 16 chars of hash

# ── secrets — cryptographically secure random ─────────
import secrets
token  = secrets.token_hex(32)    # 64-char hex token
passwd = secrets.token_urlsafe(16) # URL-safe token

import sqlite3

# Connect — creates file if it doesn't exist
# conn = sqlite3.connect("mydb.sqlite")

# In-memory database — fast, no file, lost on close
with sqlite3.connect(":memory:") as conn:
    cursor = conn.cursor()

    # Create table
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS users (
            id       INTEGER PRIMARY KEY AUTOINCREMENT,
            name     TEXT    NOT NULL,
            email    TEXT    UNIQUE NOT NULL,
            age      INTEGER CHECK(age >= 0),
            created  TEXT    DEFAULT (datetime('now'))
        )
    """)

    # Insert one row — use ? placeholders (NEVER string format!)
    cursor.execute(
        "INSERT INTO users (name, email, age) VALUES (?, ?, ?)",
        ("Alice", "[email protected]", 28)
    )

    # Insert many rows
    users = [
        ("Bob",   "[email protected]",   35),
        ("Carol", "[email protected]", 31),
        ("Dave",  "[email protected]",  27),
    ]
    cursor.executemany(
        "INSERT INTO users (name, email, age) VALUES (?, ?, ?)",
        users
    )

    conn.commit()
    print(f"Inserted rows, last ID: {cursor.lastrowid}")

import sqlite3

with sqlite3.connect(":memory:") as conn:
    conn.row_factory = sqlite3.Row   # access columns by name
    cursor = conn.cursor()

    cursor.execute("""
        CREATE TABLE users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT, email TEXT, age INTEGER, dept TEXT
        )
    """)
    cursor.executemany(
        "INSERT INTO users (name,email,age,dept) VALUES (?,?,?,?)",
        [("Alice","[email protected]",28,"Eng"),
         ("Bob","[email protected]",35,"Mkt"),
         ("Carol","[email protected]",31,"Eng"),
         ("Dave","[email protected]",27,"HR")]
    )
    conn.commit()

    # Fetch all rows
    cursor.execute("SELECT * FROM users")
    for row in cursor.fetchall():
        print(dict(row))

    # Fetch one row
    cursor.execute("SELECT * FROM users WHERE id = ?", (1,))
    user = cursor.fetchone()
    print(f"Found: {user['name']}")

    # Parameterized WHERE
    cursor.execute(
        "SELECT name, age FROM users WHERE age > ? AND dept = ?",
        (25, "Eng")
    )
    for row in cursor.fetchall():
        print(f"  {row['name']}: {row['age']}")

    # Aggregates
    cursor.execute("SELECT COUNT(*), AVG(age), MAX(age) FROM users")
    count, avg, max_age = cursor.fetchone()
    print(f"Count={count}, Avg={avg:.1f}, Max={max_age}")

    # ORDER BY, LIMIT
    cursor.execute(
        "SELECT name, age FROM users ORDER BY age DESC LIMIT 3"
    )
    print([dict(r) for r in cursor.fetchall()])

import sqlite3

with sqlite3.connect(":memory:") as conn:
    conn.row_factory = sqlite3.Row
    cur = conn.cursor()
    cur.execute("""CREATE TABLE accounts (
        id INTEGER PRIMARY KEY, owner TEXT, balance REAL
    )""")
    cur.executemany("INSERT INTO accounts VALUES (?,?,?)",
                    [(1,"Alice",1000.0),(2,"Bob",500.0),(3,"Carol",750.0)])
    conn.commit()

    # UPDATE
    cur.execute(
        "UPDATE accounts SET balance = balance + ? WHERE owner = ?",
        (200.0, "Alice")
    )
    print(f"Updated {cur.rowcount} rows")

    # DELETE
    cur.execute("DELETE FROM accounts WHERE balance < ?", (600.0,))
    print(f"Deleted {cur.rowcount} rows")

    # TRANSACTION — all-or-nothing transfer
    def transfer(conn, from_id, to_id, amount):
        try:
            cur = conn.cursor()
            cur.execute("SELECT balance FROM accounts WHERE id=?", (from_id,))
            row = cur.fetchone()
            if not row or row["balance"] < amount:
                raise ValueError("Insufficient funds")
            cur.execute(
                "UPDATE accounts SET balance = balance - ? WHERE id = ?",
                (amount, from_id)
            )
            cur.execute(
                "UPDATE accounts SET balance = balance + ? WHERE id = ?",
                (amount, to_id)
            )
            conn.commit()
            print(f"Transferred ${amount:.2f}")
        except Exception as e:
            conn.rollback()
            print(f"Transfer failed: {e}")
            raise

    transfer(conn, 1, 3, 300.0)

    cur.execute("SELECT owner, balance FROM accounts ORDER BY id")
    for row in cur.fetchall():
        print(f"  {row['owner']}: ${row['balance']:.2f}")

import sqlite3

with sqlite3.connect(":memory:") as conn:
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA foreign_keys = ON")
    cur = conn.cursor()

    cur.executescript("""
        CREATE TABLE departments (
            id   INTEGER PRIMARY KEY,
            name TEXT NOT NULL UNIQUE
        );
        CREATE TABLE employees (
            id      INTEGER PRIMARY KEY AUTOINCREMENT,
            name    TEXT NOT NULL,
            salary  REAL NOT NULL,
            dept_id INTEGER REFERENCES departments(id) ON DELETE CASCADE
        );
        CREATE TABLE projects (
            id   INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL
        );
        CREATE TABLE employee_projects (
            emp_id  INTEGER REFERENCES employees(id),
            proj_id INTEGER REFERENCES projects(id),
            role    TEXT,
            PRIMARY KEY (emp_id, proj_id)
        );
        CREATE INDEX idx_emp_dept   ON employees(dept_id);
        CREATE INDEX idx_emp_salary ON employees(salary);
    """)

    cur.executemany("INSERT INTO departments VALUES (?,?)",
                    [(1,"Engineering"),(2,"Marketing"),(3,"HR")])
    cur.executemany(
        "INSERT INTO employees (name,salary,dept_id) VALUES (?,?,?)",
        [("Alice",95000,1),("Bob",72000,2),
         ("Carol",88000,1),("Dave",58000,3)]
    )
    cur.executemany("INSERT INTO projects VALUES (?,?)",
                    [(1,"Alpha"),(2,"Beta")])
    cur.executemany("INSERT INTO employee_projects VALUES (?,?,?)",
                    [(1,1,"Lead"),(1,2,"Member"),(3,1,"Member")])
    conn.commit()

    # INNER JOIN
    cur.execute("""
        SELECT e.name, e.salary, d.name AS dept
        FROM employees e
        JOIN departments d ON e.dept_id = d.id
        ORDER BY e.salary DESC
    """)
    print("Employees with departments:")
    for r in cur.fetchall():
        print(f"  {r['name']:<10} ${r['salary']:>8,.0f}  {r['dept']}")

    # Many-to-many JOIN
    cur.execute("""
        SELECT e.name, p.name AS project, ep.role
        FROM employees e
        JOIN employee_projects ep ON e.id = ep.emp_id
        JOIN projects p ON ep.proj_id = p.id
    """)
    print("\nEmployee projects:")
    for r in cur.fetchall():
        print(f"  {r['name']:<10} → {r['project']} ({r['role']})")

import sqlite3
from contextlib import contextmanager

class Database:
    """Thread-safe SQLite wrapper with context manager support."""

    def __init__(self, db_path=":memory:"):
        self.db_path = db_path
        self._conn   = None

    def connect(self):
        self._conn = sqlite3.connect(
            self.db_path,
            check_same_thread=False,
            detect_types=sqlite3.PARSE_DECLTYPES
        )
        self._conn.row_factory = sqlite3.Row
        self._conn.execute("PRAGMA foreign_keys = ON")
        self._conn.execute("PRAGMA journal_mode = WAL")
        return self

    def close(self):
        if self._conn:
            self._conn.close()
            self._conn = None

    def __enter__(self):  return self.connect()
    def __exit__(self, *args): self.close()

    @contextmanager
    def transaction(self):
        try:
            yield self._conn
            self._conn.commit()
        except Exception:
            self._conn.rollback()
            raise

    def execute(self, sql, params=()):
        return self._conn.execute(sql, params)

    def executemany(self, sql, params_list):
        return self._conn.executemany(sql, params_list)

    def fetchall(self, sql, params=()):
        return self._conn.execute(sql, params).fetchall()

    def fetchone(self, sql, params=()):
        return self._conn.execute(sql, params).fetchone()

    def fetchval(self, sql, params=()):
        row = self.fetchone(sql, params)
        return row[0] if row else None

    def table_exists(self, name):
        return bool(self.fetchval(
            "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?",
            (name,)
        ))

# Usage
with Database(":memory:") as db:
    db.execute("""CREATE TABLE products (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL, price REAL, stock INTEGER DEFAULT 0
    )""")
    with db.transaction():
        db.executemany(
            "INSERT INTO products (name,price,stock) VALUES (?,?,?)",
            [("Widget",9.99,100),("Gadget",24.99,50),("Doohickey",4.99,200)]
        )
    rows = db.fetchall("SELECT * FROM products ORDER BY price DESC")
    for r in rows:
        print(f"  {r['name']:<12} ${r['price']:>6.2f}  stock={r['stock']}")

# pip install sqlalchemy

from sqlalchemy import (
    create_engine, Column, Integer, String,
    Float, Boolean, DateTime, ForeignKey
)
from sqlalchemy.orm import (
    DeclarativeBase, relationship, sessionmaker
)
from datetime import datetime

class Base(DeclarativeBase):
    pass

class Department(Base):
    __tablename__ = "departments"
    id        = Column(Integer, primary_key=True)
    name      = Column(String(100), nullable=False, unique=True)
    employees = relationship("Employee", back_populates="department",
                             cascade="all, delete-orphan")
    def __repr__(self):
        return f"Department(name={self.name!r})"

class Employee(Base):
    __tablename__ = "employees"
    id         = Column(Integer, primary_key=True)
    name       = Column(String(100), nullable=False)
    email      = Column(String(200), unique=True)
    salary     = Column(Float, default=0.0)
    active     = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    dept_id    = Column(Integer, ForeignKey("departments.id"))
    department = relationship("Department", back_populates="employees")
    def __repr__(self):
        return f"Employee(name={self.name!r}, salary={self.salary})"

engine       = create_engine("sqlite:///:memory:", echo=False)
Base.metadata.create_all(engine)
SessionLocal = sessionmaker(bind=engine)

with SessionLocal() as session:
    eng  = Department(name="Engineering")
    mkt  = Department(name="Marketing")
    session.add_all([eng, mkt])
    session.flush()

    session.add_all([
        Employee(name="Alice", email="[email protected]",
                 salary=95000, department=eng),
        Employee(name="Bob",   email="[email protected]",
                 salary=72000, department=mkt),
        Employee(name="Carol", email="[email protected]",
                 salary=88000, department=eng),
    ])
    session.commit()
    print("ORM models created successfully")

# Continuing from 18.6 setup...

with SessionLocal() as session:
    # Seed
    eng = Department(name="Engineering")
    mkt = Department(name="Marketing")
    session.add_all([eng, mkt])
    session.add_all([
        Employee(name="Alice", salary=95000, department=eng),
        Employee(name="Bob",   salary=72000, department=mkt),
        Employee(name="Carol", salary=88000, department=eng),
        Employee(name="Dave",  salary=58000, department=mkt),
    ])
    session.commit()

    # SELECT all
    for e in session.query(Employee).all():
        print(f"  {e.name}: ${e.salary:,.0f}")

    # Filter
    high = (session.query(Employee)
            .filter(Employee.salary > 80000)
            .order_by(Employee.salary.desc())
            .all())
    print(f"High earners: {[e.name for e in high]}")

    # Get by primary key
    emp = session.get(Employee, 1)
    print(f"Employee #1: {emp.name}")
    print(f"Department:  {emp.department.name}")

    # Aggregate
    from sqlalchemy import func
    avg = session.query(func.avg(Employee.salary)).scalar()
    cnt = session.query(func.count(Employee.id)).scalar()
    print(f"Avg: ${avg:,.0f}  Count: {cnt}")

    # JOIN
    results = (session.query(
                   Employee.name,
                   Employee.salary,
                   Department.name.label("dept"))
               .join(Department)
               .order_by(Employee.salary.desc())
               .all())
    for name, salary, dept in results:
        print(f"  {name:<10} ${salary:>8,.0f}  {dept}")

# Continuing from 18.6 setup...
from sqlalchemy import update, delete

with SessionLocal() as session:
    eng   = Department(name="Engineering")
    alice = Employee(name="Alice", salary=95000, department=eng)
    bob   = Employee(name="Bob",   salary=72000, department=eng)
    session.add_all([eng, alice, bob])
    session.commit()

    # UPDATE — modify object, then commit
    alice = session.get(Employee, alice.id)
    alice.salary = 100000
    alice.name   = "Alice Smith"
    session.commit()
    print(f"Updated: {alice.name} — ${alice.salary:,.0f}")

    # Bulk UPDATE
    session.execute(
        update(Employee)
        .where(Employee.salary < 80000)
        .values(salary=Employee.salary * 1.10)
    )
    session.commit()

    # DELETE single object
    emp = session.get(Employee, bob.id)
    if emp:
        session.delete(emp)
        session.commit()
        print(f"Deleted: {emp.name}")

    # Bulk DELETE
    session.execute(
        delete(Employee).where(Employee.active == False)
    )
    session.commit()
    print("Bulk operations complete")

import sqlite3
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional, List

@dataclass
class User:
    name:       str
    email:      str
    age:        int
    id:         Optional[int] = None
    created_at: Optional[str] = None

class UserRepository(ABC):
    @abstractmethod
    def add(self, user: User) -> User: pass
    @abstractmethod
    def get_by_id(self, user_id: int) -> Optional[User]: pass
    @abstractmethod
    def get_by_email(self, email: str) -> Optional[User]: pass
    @abstractmethod
    def get_all(self) -> List[User]: pass
    @abstractmethod
    def update(self, user: User) -> User: pass
    @abstractmethod
    def delete(self, user_id: int) -> bool: pass
    @abstractmethod
    def count(self) -> int: pass

class SQLiteUserRepository(UserRepository):
    def __init__(self, conn):
        self._conn = conn
        self._conn.row_factory = sqlite3.Row
        self._conn.execute("""
            CREATE TABLE IF NOT EXISTS users (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name TEXT NOT NULL, email TEXT UNIQUE NOT NULL,
                age INTEGER, created_at TEXT DEFAULT (datetime('now'))
            )""")
        self._conn.commit()

    def _row_to_user(self, row):
        return User(id=row["id"], name=row["name"],
                    email=row["email"], age=row["age"],
                    created_at=row["created_at"])

    def add(self, user):
        cur = self._conn.execute(
            "INSERT INTO users (name,email,age) VALUES (?,?,?)",
            (user.name, user.email, user.age))
        self._conn.commit()
        return self.get_by_id(cur.lastrowid)

    def get_by_id(self, uid):
        row = self._conn.execute(
            "SELECT * FROM users WHERE id=?", (uid,)).fetchone()
        return self._row_to_user(row) if row else None

    def get_by_email(self, email):
        row = self._conn.execute(
            "SELECT * FROM users WHERE email=?", (email,)).fetchone()
        return self._row_to_user(row) if row else None

    def get_all(self):
        return [self._row_to_user(r) for r in
                self._conn.execute("SELECT * FROM users ORDER BY name")]

    def update(self, user):
        self._conn.execute(
            "UPDATE users SET name=?,email=?,age=? WHERE id=?",
            (user.name, user.email, user.age, user.id))
        self._conn.commit()
        return self.get_by_id(user.id)

    def delete(self, uid):
        cur = self._conn.execute("DELETE FROM users WHERE id=?", (uid,))
        self._conn.commit()
        return cur.rowcount > 0

    def count(self):
        return self._conn.execute(
            "SELECT COUNT(*) FROM users").fetchone()[0]

class InMemoryUserRepository(UserRepository):
    """For testing — no database needed."""
    def __init__(self):
        self._store   = {}
        self._next_id = 1

    def add(self, user):
        user.id = self._next_id
        self._store[self._next_id] = user
        self._next_id += 1
        return user

    def get_by_id(self, uid): return self._store.get(uid)
    def get_by_email(self, email):
        return next((u for u in self._store.values()
                     if u.email == email), None)
    def get_all(self):
        return sorted(self._store.values(), key=lambda u: u.name)
    def update(self, user):
        self._store[user.id] = user
        return user
    def delete(self, uid): return bool(self._store.pop(uid, None))
    def count(self): return len(self._store)

class UserService:
    def __init__(self, repo: UserRepository):
        self.repo = repo

    def register(self, name, email, age):
        if self.repo.get_by_email(email):
            raise ValueError(f"Email already registered: {email}")
        return self.repo.add(User(name=name, email=email, age=age))

    def get_profile(self, user_id):
        user = self.repo.get_by_id(user_id)
        if not user:
            raise KeyError(f"User {user_id} not found")
        return user

# Demo
conn    = sqlite3.connect(":memory:")
repo    = SQLiteUserRepository(conn)
service = UserService(repo)

alice = service.register("Alice", "[email protected]", 28)
bob   = service.register("Bob",   "[email protected]",   35)
carol = service.register("Carol", "[email protected]", 31)

print(f"Registered {repo.count()} users")
for user in repo.get_all():
    print(f"  [{user.id}] {user.name} ({user.age}) — {user.email}")

import sqlite3

class MigrationManager:
    def __init__(self, conn):
        self._conn = conn
        self._conn.execute("""
            CREATE TABLE IF NOT EXISTS schema_migrations (
                version     TEXT PRIMARY KEY,
                applied_at  TEXT DEFAULT (datetime('now')),
                description TEXT
            )""")
        self._conn.commit()

    def _is_applied(self, version):
        return bool(self._conn.execute(
            "SELECT 1 FROM schema_migrations WHERE version=?",
            (version,)).fetchone())

    def apply(self, version, description, up_sql):
        if self._is_applied(version):
            print(f"  ⏭  Already applied: {version}")
            return False
        try:
            self._conn.executescript(up_sql)
            self._conn.execute(
                "INSERT INTO schema_migrations (version,description) VALUES (?,?)",
                (version, description))
            self._conn.commit()
            print(f"  ✅ Applied: {version} — {description}")
            return True
        except Exception as e:
            self._conn.rollback()
            print(f"  ❌ Failed:  {version} — {e}")
            raise

    def status(self):
        rows = self._conn.execute(
            "SELECT version, applied_at, description "
            "FROM schema_migrations ORDER BY version"
        ).fetchall()
        print(f"\n{'Version':<15} {'Applied At':<22} Description")
        print("─" * 60)
        for r in rows:
            print(f"{r[0]:<15} {r[1]:<22} {r[2]}")

MIGRATIONS = [
    ("001_initial",        "Create users table",
     "CREATE TABLE users (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, email TEXT UNIQUE NOT NULL);"),
    ("002_add_age",        "Add age column",
     "ALTER TABLE users ADD COLUMN age INTEGER DEFAULT 0;"),
    ("003_add_timestamps", "Add created_at column",
     "ALTER TABLE users ADD COLUMN created_at TEXT DEFAULT (datetime('now'));"),
    ("004_add_products",   "Create products table",
     "CREATE TABLE products (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, price REAL NOT NULL, stock INTEGER DEFAULT 0); CREATE INDEX idx_products_price ON products(price);"),
]

conn = sqlite3.connect(":memory:")
mgr  = MigrationManager(conn)
print("Running migrations...")
for version, description, sql in MIGRATIONS:
    mgr.apply(version, description, sql)
mgr.status()

# pip install flask

from flask import Flask, request, jsonify, redirect, url_for

app = Flask(__name__)
app.config["SECRET_KEY"] = "your-secret-key-here"
app.config["DEBUG"]      = True

# Basic route
@app.route("/")
def index():
    return "<h1>Hello, Flask!</h1>"

# Route with variable
@app.route("/user/<username>")
def user_profile(username):
    return f"<p>Profile: {username}</p>"

# Typed variable — only matches integers
@app.route("/post/<int:post_id>")
def get_post(post_id):
    return jsonify({"id": post_id, "title": f"Post {post_id}"})

# Multiple HTTP methods
@app.route("/login", methods=["GET", "POST"])
def login():
    if request.method == "POST":
        username = request.form.get("username")
        password = request.form.get("password")
        if username == "admin" and password == "secret":
            return redirect(url_for("dashboard"))
        return "<p>Invalid credentials</p>", 401
    return """
        <form method="POST">
            <input name="username" placeholder="Username">
            <input name="password" type="password">
            <button type="submit">Login</button>
        </form>
    """

@app.route("/dashboard")
def dashboard():
    return "<h1>Dashboard</h1>"

if __name__ == "__main__":
    app.run(debug=True, port=5000)

from flask import Flask, request, jsonify, make_response

app = Flask(__name__)

@app.route("/api/echo", methods=["POST"])
def echo():
    data = request.get_json()
    if not data:
        return jsonify({"error": "No JSON body"}), 400
    fmt  = request.args.get("format", "default")   # query param
    auth = request.headers.get("Authorization", "")
    return jsonify({"received": data, "format": fmt, "auth": bool(auth)})

@app.route("/api/upload", methods=["POST"])
def upload():
    name = request.form.get("name")
    if "file" not in request.files:
        return jsonify({"error": "No file"}), 400
    file = request.files["file"]
    if file.filename == "":
        return jsonify({"error": "Empty filename"}), 400
    from werkzeug.utils import secure_filename
    filename = secure_filename(file.filename)
    return jsonify({"saved": filename, "uploaded_by": name})

# Custom response — headers, cookies, status
@app.route("/api/custom")
def custom_response():
    response = make_response(jsonify({"message": "Custom"}), 200)
    response.headers["X-Custom-Header"] = "MyValue"
    response.set_cookie("session_id", "abc123",
                        httponly=True, secure=True, max_age=3600)
    return response

# Global error handlers
@app.errorhandler(404)
def not_found(error):
    return jsonify({"error": "Resource not found"}), 404

@app.errorhandler(500)
def server_error(error):
    return jsonify({"error": "Internal server error"}), 500

from flask import Flask, request, jsonify

app     = Flask(__name__)
_db     = {}
_nxt_id = 1

def make_error(msg, code):
    return jsonify({"error": msg}), code

# GET  /api/users  — list all
# POST /api/users  — create
@app.route("/api/users", methods=["GET", "POST"])
def users():
    global _nxt_id
    if request.method == "GET":
        result  = list(_db.values())
        min_age = request.args.get("min_age", type=int)
        sort_by = request.args.get("sort")
        if min_age:
            result = [u for u in result if u["age"] >= min_age]
        if sort_by in ("name", "age", "id"):
            result.sort(key=lambda u: u[sort_by])
        return jsonify({"users": result, "count": len(result)})

    data    = request.get_json()
    if not data:
        return make_error("JSON body required", 400)
    missing = [f for f in ("name","email","age") if f not in data]
    if missing:
        return make_error(f"Missing fields: {missing}", 400)
    user = {"id": _nxt_id, "name": data["name"],
            "email": data["email"], "age": data["age"]}
    _db[_nxt_id] = user
    _nxt_id += 1
    return jsonify(user), 201

# GET    /api/users/<id>  — get one
# PUT    /api/users/<id>  — replace
# PATCH  /api/users/<id>  — partial update
# DELETE /api/users/<id>  — delete
@app.route("/api/users/<int:uid>",
           methods=["GET","PUT","PATCH","DELETE"])
def user(uid):
    if uid not in _db:
        return make_error(f"User {uid} not found", 404)
    if request.method == "GET":
        return jsonify(_db[uid])
    if request.method == "DELETE":
        return jsonify({"deleted": _db.pop(uid)}), 200
    data = request.get_json() or {}
    if request.method == "PUT":
        missing = [f for f in ("name","email","age") if f not in data]
        if missing:
            return make_error(f"Missing: {missing}", 400)
        _db[uid] = {"id": uid, "name": data["name"],
                    "email": data["email"], "age": data["age"]}
    elif request.method == "PATCH":
        for key in ("name", "email", "age"):
            if key in data:
                _db[uid][key] = data[key]
    return jsonify(_db[uid])

# myapp/__init__.py — Application Factory
from flask import Flask

def create_app(config_name="development"):
    app = Flask(__name__)
    from .config import config_map
    app.config.from_object(config_map[config_name])

    from .routes.auth  import auth_bp
    from .routes.users import users_bp
    from .routes.api   import api_bp
    app.register_blueprint(auth_bp,  url_prefix="/auth")
    app.register_blueprint(users_bp, url_prefix="/users")
    app.register_blueprint(api_bp,   url_prefix="/api/v1")
    return app

# myapp/config.py
class Config:
    SECRET_KEY = "dev-secret"
    DEBUG      = False
    TESTING    = False

class DevelopmentConfig(Config):
    DEBUG = True

class ProductionConfig(Config):
    SECRET_KEY = "prod-secret-from-env"

class TestingConfig(Config):
    TESTING  = True
    DATABASE = ":memory:"

config_map = {
    "development": DevelopmentConfig,
    "production":  ProductionConfig,
    "testing":     TestingConfig,
}

# myapp/routes/users.py — Blueprint
from flask import Blueprint, jsonify, request

users_bp = Blueprint("users", __name__)

@users_bp.route("/")
def list_users():
    return jsonify({"users": []})

@users_bp.route("/<int:user_id>")
def get_user(user_id):
    return jsonify({"id": user_id})

# run.py
from myapp import create_app
app = create_app("development")

if __name__ == "__main__":
    app.run()

from flask import Flask, request, jsonify, g
from functools import wraps
import time
from collections import defaultdict

app = Flask(__name__)
VALID_TOKENS = {"token-alice": "alice", "token-bob": "bob"}

def require_auth(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        token = request.headers.get("Authorization","").replace("Bearer ","")
        if not token or token not in VALID_TOKENS:
            return jsonify({"error": "Unauthorized"}), 401
        g.current_user = VALID_TOKENS[token]
        return f(*args, **kwargs)
    return decorated

def require_json(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        if not request.is_json:
            return jsonify({"error": "application/json required"}), 415
        return f(*args, **kwargs)
    return decorated

@app.before_request
def start_timer():
    g.start_time = time.time()

@app.after_request
def add_timing(response):
    ms = (time.time() - getattr(g, "start_time", time.time())) * 1000
    response.headers["X-Response-Time"] = f"{ms:.2f}ms"
    return response

@app.route("/api/profile")
@require_auth
def profile():
    return jsonify({"user": g.current_user})

@app.route("/api/data", methods=["POST"])
@require_auth
@require_json
def post_data():
    return jsonify({"user": g.current_user, "data": request.get_json()})

# Simple rate limiter
_counts = defaultdict(list)

def rate_limit(max_req=10, window=60):
    def decorator(f):
        @wraps(f)
        def decorated(*args, **kwargs):
            ip  = request.remote_addr
            now = time.time()
            _counts[ip] = [t for t in _counts[ip] if now - t < window]
            if len(_counts[ip]) >= max_req:
                return jsonify({"error": "Rate limit exceeded"}), 429
            _counts[ip].append(now)
            return f(*args, **kwargs)
        return decorated
    return decorator

@app.route("/api/limited")
@rate_limit(max_req=5, window=60)
def limited():
    return jsonify({"message": "OK"})

# pip install fastapi uvicorn

from fastapi import FastAPI, HTTPException, Query, Path
from pydantic import BaseModel, Field, field_validator
from typing import Optional, List
from datetime import datetime

app = FastAPI(
    title="My API",
    description="A demonstration FastAPI application",
    version="1.0.0",
)

# Pydantic models — request/response schemas
class UserCreate(BaseModel):
    name:  str           = Field(..., min_length=1, max_length=100)
    email: str           = Field(..., description="User email address")
    age:   int           = Field(..., ge=0, le=150)
    bio:   Optional[str] = Field(None, max_length=500)

    @field_validator("name")
    @classmethod
    def name_not_blank(cls, v):
        if not v.strip():
            raise ValueError("Name cannot be blank")
        return v.strip()

class UserResponse(BaseModel):
    id:         int
    name:       str
    email:      str
    age:        int
    created_at: str

class UserUpdate(BaseModel):
    name:  Optional[str] = None
    email: Optional[str] = None
    age:   Optional[int] = Field(None, ge=0, le=150)

_users   = {}
_next_id = 1

@app.get("/")
def root():
    return {"message": "Welcome to My API", "docs": "/docs"}

@app.get("/health")
def health_check():
    return {"status": "healthy", "timestamp": datetime.utcnow().isoformat()}

# Run with: uvicorn main:app --reload
# Docs at:  http://localhost:8000/docs

# Continuing from 19.6 setup...

@app.post("/users", status_code=201)
def create_user(user: UserCreate):
    global _next_id
    for u in _users.values():
        if u["email"] == user.email:
            raise HTTPException(409, "Email already registered")
    new_user = {
        "id":         _next_id,
        "name":       user.name,
        "email":      user.email,
        "age":        user.age,
        "bio":        user.bio,
        "created_at": datetime.utcnow().isoformat(),
    }
    _users[_next_id] = new_user
    _next_id += 1
    return new_user

@app.get("/users")
def list_users(
    skip:    int           = Query(0, ge=0),
    limit:   int           = Query(10, ge=1, le=100),
    min_age: Optional[int] = Query(None, ge=0),
    search:  Optional[str] = Query(None),
):
    result = list(_users.values())
    if min_age is not None:
        result = [u for u in result if u["age"] >= min_age]
    if search:
        result = [u for u in result
                  if search.lower() in u["name"].lower()]
    total  = len(result)
    result = result[skip: skip + limit]
    return {"users": result, "total": total, "skip": skip, "limit": limit}

@app.get("/users/{user_id}")
def get_user(user_id: int = Path(..., ge=1)):
    if user_id not in _users:
        raise HTTPException(404, f"User {user_id} not found")
    return _users[user_id]

@app.patch("/users/{user_id}")
def update_user(user_id: int, updates: UserUpdate):
    if user_id not in _users:
        raise HTTPException(404, f"User {user_id} not found")
    update_data = updates.model_dump(exclude_unset=True)
    _users[user_id].update(update_data)
    return _users[user_id]

@app.delete("/users/{user_id}", status_code=204)
def delete_user(user_id: int):
    if user_id not in _users:
        raise HTTPException(404, f"User {user_id} not found")
    del _users[user_id]

# pip install python-jose[cryptography] passlib[bcrypt]

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel
from datetime import datetime, timedelta
from typing import Optional
import hashlib, hmac, json, base64

app    = FastAPI()
SECRET = "super-secret-key-change-in-production"

def create_token(data: dict, expires_minutes=30):
    payload     = {**data,
                   "exp": (datetime.utcnow() +
                            timedelta(minutes=expires_minutes)).isoformat()}
    payload_b64 = base64.urlsafe_b64encode(
        json.dumps(payload).encode()).decode()
    sig = hmac.new(SECRET.encode(), payload_b64.encode(),
                   hashlib.sha256).hexdigest()
    return f"{payload_b64}.{sig}"

def verify_token(token: str):
    try:
        payload_b64, sig = token.rsplit(".", 1)
        expected = hmac.new(SECRET.encode(),
                            payload_b64.encode(),
                            hashlib.sha256).hexdigest()
        if not hmac.compare_digest(sig, expected):
            return None
        payload = json.loads(base64.urlsafe_b64decode(payload_b64))
        if datetime.fromisoformat(payload["exp"]) < datetime.utcnow():
            return None
        return payload
    except Exception:
        return None

USERS_DB      = {
    "alice": {"username":"alice","password":"hashed_alice","role":"admin"},
    "bob":   {"username":"bob",  "password":"hashed_bob",  "role":"user"},
}
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")

def get_current_user(token: str = Depends(oauth2_scheme)):
    payload = verify_token(token)
    if not payload:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED,
                            "Invalid or expired token",
                            headers={"WWW-Authenticate": "Bearer"})
    user = USERS_DB.get(payload.get("sub"))
    if not user:
        raise HTTPException(401, "User not found")
    return user

def require_admin(user=Depends(get_current_user)):
    if user["role"] != "admin":
        raise HTTPException(403, "Admin only")
    return user

@app.post("/auth/token")
def login(form: OAuth2PasswordRequestForm = Depends()):
    user = USERS_DB.get(form.username)
    if not user or form.password != "password":
        raise HTTPException(401, "Bad credentials")
    token = create_token({"sub": user["username"], "role": user["role"]})
    return {"access_token": token, "token_type": "bearer"}

@app.get("/me")
def get_me(current_user=Depends(get_current_user)):
    return {"username": current_user["username"],
            "role":     current_user["role"]}

@app.get("/admin/stats")
def admin_stats(admin=Depends(require_admin)):
    return {"total_users": len(USERS_DB), "admin": admin["username"]}

# pip install fastapi uvicorn aiosqlite

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from typing import Optional, List
from contextlib import asynccontextmanager
import aiosqlite

DATABASE = ":memory:"

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with aiosqlite.connect(DATABASE) as db:
        await db.execute("""
            CREATE TABLE IF NOT EXISTS items (
                id          INTEGER PRIMARY KEY AUTOINCREMENT,
                title       TEXT NOT NULL,
                description TEXT,
                price       REAL NOT NULL,
                in_stock    INTEGER DEFAULT 1
            )
        """)
        await db.commit()
    print("Database ready")
    yield
    print("Shutting down")

app = FastAPI(lifespan=lifespan)

async def get_db():
    async with aiosqlite.connect(DATABASE) as db:
        db.row_factory = aiosqlite.Row
        yield db

class ItemCreate(BaseModel):
    title:       str
    description: Optional[str] = None
    price:       float
    in_stock:    bool = True

class Item(ItemCreate):
    id: int

@app.post("/items", status_code=201)
async def create_item(item: ItemCreate, db=Depends(get_db)):
    cursor = await db.execute(
        "INSERT INTO items (title,description,price,in_stock) VALUES (?,?,?,?)",
        (item.title, item.description, item.price, int(item.in_stock))
    )
    await db.commit()
    return {**item.model_dump(), "id": cursor.lastrowid}

@app.get("/items", response_model=List[Item])
async def list_items(skip: int=0, limit: int=10, db=Depends(get_db)):
    async with db.execute(
        "SELECT * FROM items LIMIT ? OFFSET ?", (limit, skip)
    ) as cursor:
        rows = await cursor.fetchall()
    return [dict(r) for r in rows]

@app.get("/items/{item_id}", response_model=Item)
async def get_item(item_id: int, db=Depends(get_db)):
    async with db.execute(
        "SELECT * FROM items WHERE id=?", (item_id,)
    ) as cursor:
        row = await cursor.fetchone()
    if not row:
        raise HTTPException(404, "Item not found")
    return dict(row)

@app.delete("/items/{item_id}", status_code=204)
async def delete_item(item_id: int, db=Depends(get_db)):
    cursor = await db.execute("DELETE FROM items WHERE id=?", (item_id,))
    await db.commit()
    if cursor.rowcount == 0:
        raise HTTPException(404, "Item not found")

# Background tasks — run after response is sent
from fastapi import FastAPI, BackgroundTasks, WebSocket, WebSocketDisconnect
from typing import List
import time, logging

app    = FastAPI()
logger = logging.getLogger(__name__)

def send_email(to: str, subject: str, body: str):
    time.sleep(2)   # simulate slow operation
    logger.info(f"Email sent to {to}: {subject}")
    print(f"Email sent to {to}: {subject}")

def log_activity(user_id: int, action: str):
    logger.info(f"User {user_id}: {action}")

@app.post("/register")
def register(email: str, background_tasks: BackgroundTasks):
    # Responds immediately — email sends in background
    background_tasks.add_task(
        send_email, email, "Welcome!", "Thanks for registering."
    )
    background_tasks.add_task(log_activity, 1, "register")
    return {"message": "Registered! Welcome email on its way."}

# WebSocket chat
class ConnectionManager:
    def __init__(self):
        self.active: List[WebSocket] = []

    async def connect(self, ws: WebSocket):
        await ws.accept()
        self.active.append(ws)

    def disconnect(self, ws: WebSocket):
        if ws in self.active:
            self.active.remove(ws)

    async def broadcast(self, message: str):
        for ws in self.active:
            await ws.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/chat")
async def chat(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"Message: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast("A user disconnected")

import pytest
from flask import Flask, request, jsonify
from fastapi import FastAPI
from fastapi.testclient import TestClient

# ── Flask test setup ──────────────────────────────────
def create_test_flask_app():
    app = Flask(__name__)
    app.config["TESTING"] = True
    _db = {}; _id = [1]

    @app.route("/api/users", methods=["GET","POST"])
    def users():
        if request.method == "POST":
            data = request.get_json()
            user = {"id": _id[0], **data}
            _db[_id[0]] = user
            _id[0] += 1
            return jsonify(user), 201
        return jsonify(list(_db.values()))

    @app.route("/api/users/<int:uid>", methods=["GET","DELETE"])
    def user(uid):
        if uid not in _db:
            return jsonify({"error": "Not found"}), 404
        if request.method == "DELETE":
            return jsonify({"deleted": _db.pop(uid)})
        return jsonify(_db[uid])

    return app

@pytest.fixture
def flask_client():
    app = create_test_flask_app()
    with app.test_client() as client:
        yield client

def test_flask_create_user(flask_client):
    resp = flask_client.post("/api/users",
                             json={"name":"Alice","email":"[email protected]","age":28})
    assert resp.status_code == 201
    data = resp.get_json()
    assert data["name"] == "Alice"
    assert "id" in data

def test_flask_list_users(flask_client):
    flask_client.post("/api/users",
                      json={"name":"Alice","email":"[email protected]","age":28})
    resp = flask_client.get("/api/users")
    assert resp.status_code == 200
    assert len(resp.get_json()) == 1

def test_flask_not_found(flask_client):
    resp = flask_client.get("/api/users/999")
    assert resp.status_code == 404

# ── FastAPI test setup ────────────────────────────────
def create_test_fastapi_app():
    from pydantic import BaseModel
    test_app = FastAPI()
    _db = {}; _id = [1]

    class UserIn(BaseModel):
        name: str; email: str; age: int

    @test_app.get("/ping")
    def ping(): return {"pong": True}

    @test_app.post("/users", status_code=201)
    def create(user: UserIn):
        u = {"id": _id[0], **user.model_dump()}
        _db[_id[0]] = u; _id[0] += 1
        return u

    @test_app.get("/users/{uid}")
    def get(uid: int):
        from fastapi import HTTPException
        if uid not in _db: raise HTTPException(404, "Not found")
        return _db[uid]

    return test_app

@pytest.fixture
def fastapi_client():
    return TestClient(create_test_fastapi_app())

def test_fastapi_ping(fastapi_client):
    resp = fastapi_client.get("/ping")
    assert resp.status_code == 200
    assert resp.json() == {"pong": True}

def test_fastapi_create_user(fastapi_client):
    resp = fastapi_client.post("/users",
                               json={"name":"Bob","email":"[email protected]","age":30})
    assert resp.status_code == 201
    assert resp.json()["name"] == "Bob"

def test_fastapi_validation_error(fastapi_client):
    resp = fastapi_client.post("/users", json={"name":"Bob"})
    assert resp.status_code == 422   # Unprocessable Entity

# ── Production servers ───────────────────────────────
# Flask  → Gunicorn:  gunicorn "myapp:create_app()" --workers 4 --bind 0.0.0.0:8000
# FastAPI → Uvicorn:  uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

# ── Environment variables ─────────────────────────────
import os
from dotenv import load_dotenv   # pip install python-dotenv
load_dotenv()

DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./dev.db")
SECRET_KEY   = os.getenv("SECRET_KEY",   "dev-secret")
DEBUG        = os.getenv("DEBUG", "false").lower() == "true"

# .env file (never commit to git!):
# DATABASE_URL=postgresql://user:pass@localhost/mydb
# SECRET_KEY=super-secure-random-string-here
# DEBUG=false

# ── Dockerfile ────────────────────────────────────────
# FROM python:3.11-slim
# WORKDIR /app
# COPY requirements.txt .
# RUN pip install --no-cache-dir -r requirements.txt
# COPY . .
# EXPOSE 8000
# CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

# ── CORS (FastAPI built-in) ───────────────────────────
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://myfrontend.com", "http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# ── Health check endpoint ─────────────────────────────
import time
_start = time.time()

@app.get("/health")
def health():
    return {
        "status":   "healthy",
        "uptime_s": round(time.time() - _start),
        "version":  "1.0.0",
    }

# ── Rate limiting middleware ──────────────────────────
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse
from collections import defaultdict

class RateLimitMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, max_requests=100, window=60):
        super().__init__(app)
        self.max_req = max_requests
        self.window  = window
        self._store  = defaultdict(list)

    async def dispatch(self, request, call_next):
        ip  = request.client.host
        now = time.time()
        self._store[ip] = [t for t in self._store[ip]
                            if now - t < self.window]
        if len(self._store[ip]) >= self.max_req:
            return JSONResponse({"error": "Rate limit exceeded"}, 429)
        self._store[ip].append(now)
        return await call_next(request)

app.add_middleware(RateLimitMiddleware, max_requests=100, window=60)

# pip install numpy

import numpy as np

# ── Creating arrays ───────────────────────────────────
a = np.array([1, 2, 3, 4, 5])
b = np.array([[1, 2, 3],
              [4, 5, 6]])

print(a.shape)    # (5,)
print(b.shape)    # (2, 3)
print(b.ndim)     # 2
print(b.dtype)    # int64
print(b.size)     # 6

# Special arrays
np.zeros((3, 4))           # 3x4 array of zeros
np.ones((2, 3))            # 2x3 array of ones
np.eye(4)                  # 4x4 identity matrix
np.full((3, 3), 7)         # 3x3 filled with 7
np.arange(0, 10, 2)        # [0, 2, 4, 6, 8]
np.linspace(0, 1, 5)       # [0, 0.25, 0.5, 0.75, 1.0]
np.random.rand(3, 3)       # uniform random [0, 1)
np.random.randn(3, 3)      # standard normal distribution
np.random.randint(0, 10, (3, 3))  # random integers

# ── Vectorised operations — no loops! ─────────────────
a = np.array([1, 2, 3, 4, 5])
print(a * 2)          # [2, 4, 6, 8, 10]
print(a ** 2)         # [1, 4, 9, 16, 25]
print(a + 10)         # [11, 12, 13, 14, 15]
print(np.sqrt(a))     # [1.0, 1.41, 1.73, 2.0, 2.24]

# Element-wise operations between arrays
x = np.array([1, 2, 3])
y = np.array([4, 5, 6])
print(x + y)          # [5, 7, 9]
print(x * y)          # [4, 10, 18]
print(np.dot(x, y))   # 32  (dot product)

# ── Universal functions ───────────────────────────────
data = np.array([1.5, 2.7, 3.1, 4.9, 5.2])
print(np.floor(data))   # [1. 2. 3. 4. 5.]
print(np.ceil(data))    # [2. 3. 4. 5. 6.]
print(np.round(data))   # [2. 3. 3. 5. 5.]
print(np.abs(np.array([-1, -2, 3])))  # [1 2 3]

import numpy as np

a = np.array([[1,  2,  3,  4],
              [5,  6,  7,  8],
              [9, 10, 11, 12]])

# Indexing
print(a[0, 0])      # 1   (row 0, col 0)
print(a[2, 3])      # 12  (row 2, col 3)
print(a[-1, -1])    # 12  (last row, last col)

# Slicing — [row_start:row_end, col_start:col_end]
print(a[0, :])      # [1 2 3 4]       first row
print(a[:, 0])      # [1 5 9]         first column
print(a[0:2, 1:3])  # [[2 3],[6 7]]   submatrix
print(a[::2, ::2])  # [[1 3],[9 11]]  every other

# Boolean indexing — filter by condition
data = np.array([10, 25, 3, 47, 8, 33])
mask = data > 20
print(mask)           # [F T F T F T]
print(data[mask])     # [25 47 33]
print(data[data > 20])# same, inline

# Fancy indexing
print(a[[0, 2], :])   # rows 0 and 2
print(a[:, [0, 3]])   # cols 0 and 3

# Reshaping
flat = np.arange(12)
mat  = flat.reshape(3, 4)    # 3 rows, 4 cols
cube = flat.reshape(2, 2, 3) # 3D
back = mat.flatten()         # back to 1D

# Transpose
print(mat.T)                 # swap rows and cols
print(mat.T.shape)           # (4, 3)

# Stack arrays
x = np.array([1, 2, 3])
y = np.array([4, 5, 6])
print(np.vstack([x, y]))     # vertical stack → (2,3)
print(np.hstack([x, y]))     # horizontal → [1 2 3 4 5 6]
print(np.column_stack([x,y]))# → [[1 4],[2 5],[3 6]]

import numpy as np

data = np.array([[4, 7, 2, 9],
                 [1, 5, 8, 3],
                 [6, 2, 4, 7]])

# Aggregations — whole array
print(np.sum(data))      # 58
print(np.mean(data))     # 4.83
print(np.std(data))      # 2.37
print(np.var(data))      # 5.64
print(np.min(data))      # 1
print(np.max(data))      # 9
print(np.median(data))   # 4.5

# Axis-wise — axis=0: along rows (per column)
#              axis=1: along cols (per row)
print(np.sum(data, axis=0))    # [11 14 14 19] col sums
print(np.sum(data, axis=1))    # [22 17 19]    row sums
print(np.mean(data, axis=1))   # [5.5 4.25 4.75]

# Sorting
arr = np.array([3, 1, 4, 1, 5, 9, 2, 6])
print(np.sort(arr))            # sorted copy
print(np.argsort(arr))         # indices that sort arr

# Percentiles and quantiles
print(np.percentile(data, 25))  # Q1
print(np.percentile(data, 75))  # Q3

# Linear algebra
A = np.array([[1, 2], [3, 4]])
B = np.array([[5, 6], [7, 8]])

print(np.dot(A, B))        # matrix multiplication
print(A @ B)               # same with @ operator
print(np.linalg.det(A))    # determinant: -2.0
print(np.linalg.inv(A))    # inverse matrix
eigenvalues, eigenvectors = np.linalg.eig(A)

# Correlation and covariance
x = np.array([1, 2, 3, 4, 5])
y = np.array([2, 4, 5, 4, 5])
print(np.corrcoef(x, y))   # correlation matrix
print(np.cov(x, y))        # covariance matrix

# pip install pandas

import pandas as pd
import numpy as np

# ── Series ────────────────────────────────────────────
s = pd.Series([10, 20, 30, 40, 50],
              index=["a","b","c","d","e"])
print(s["b"])       # 20
print(s[s > 25])    # c 30, d 40, e 50
print(s.mean())     # 30.0

# ── DataFrame — from dict ─────────────────────────────
df = pd.DataFrame({
    "name":   ["Alice","Bob","Carol","Dave","Eve"],
    "age":    [28, 35, 31, 27, 42],
    "salary": [95000, 72000, 88000, 58000, 110000],
    "dept":   ["Eng","Mkt","Eng","HR","Eng"],
    "active": [True, True, True, False, True],
})

print(df.shape)         # (5, 5)
print(df.dtypes)        # column data types
print(df.head(3))       # first 3 rows
print(df.tail(2))       # last 2 rows
print(df.info())        # summary with nulls
print(df.describe())    # stats for numeric cols

# ── Selecting columns ─────────────────────────────────
print(df["name"])                    # Series
print(df[["name","salary"]])         # DataFrame
print(df.name)                       # attribute access

# ── Selecting rows ────────────────────────────────────
print(df.iloc[0])                    # first row by position
print(df.iloc[1:3])                  # rows 1-2
print(df.loc[0])                     # row by label/index
print(df.loc[0, "name"])             # single value: "Alice"

# ── Adding/removing columns ───────────────────────────
df["bonus"]       = df["salary"] * 0.10
df["senior"]      = df["age"] > 30
df["name_upper"]  = df["name"].str.upper()
df = df.drop(columns=["name_upper"])  # remove column

import pandas as pd

df = pd.DataFrame({
    "name":   ["Alice","Bob","Carol","Dave","Eve"],
    "age":    [28, 35, 31, 27, 42],
    "salary": [95000, 72000, 88000, 58000, 110000],
    "dept":   ["Eng","Mkt","Eng","HR","Eng"],
    "active": [True, True, True, False, True],
})

# ── Boolean filtering ─────────────────────────────────
print(df[df["age"] > 30])
print(df[df["dept"] == "Eng"])
print(df[(df["age"] > 28) & (df["salary"] > 80000)])
print(df[(df["dept"] == "Eng") | (df["dept"] == "Mkt")])
print(df[~df["active"]])                # NOT active

# ── .query() — SQL-like string syntax ─────────────────
print(df.query("age > 30 and dept == 'Eng'"))
print(df.query("salary > 80000 and active == True"))

# Use variable in query with @
min_salary = 75000
print(df.query("salary > @min_salary"))

# ── .isin() — filter by list ──────────────────────────
print(df[df["dept"].isin(["Eng","HR"])])
print(df[~df["name"].isin(["Bob","Dave"])])

# ── .between() ───────────────────────────────────────
print(df[df["age"].between(28, 35)])

# ── Sorting ───────────────────────────────────────────
print(df.sort_values("salary", ascending=False))
print(df.sort_values(["dept","salary"],
                      ascending=[True, False]))

# ── nlargest / nsmallest ──────────────────────────────
print(df.nlargest(3, "salary"))
print(df.nsmallest(2, "age"))

import pandas as pd
import numpy as np

df = pd.DataFrame({
    "name":   ["Alice","Bob",None,"Dave","Alice"],
    "age":    [28, np.nan, 31, 27, 28],
    "salary": [95000, 72000, np.nan, 58000, 95000],
    "dept":   ["Eng","Mkt","Eng",None,"Eng"],
})

# ── Detect missing values ─────────────────────────────
print(df.isnull())           # True where NaN/None
print(df.isnull().sum())     # count per column
print(df.isnull().sum().sum())  # total missing

# ── Drop missing ──────────────────────────────────────
df.dropna()                  # drop rows with ANY null
df.dropna(how="all")         # drop rows where ALL null
df.dropna(subset=["name"])   # drop only if name is null
df.dropna(thresh=3)          # keep rows with >= 3 non-null

# ── Fill missing ──────────────────────────────────────
df["age"].fillna(df["age"].mean())         # fill with mean
df["dept"].fillna("Unknown")               # fill with value
df["salary"].fillna(method="ffill")        # forward fill
df["salary"].fillna(method="bfill")        # backward fill

# ── Duplicates ────────────────────────────────────────
print(df.duplicated())                     # True for dupes
print(df.duplicated(subset=["name","age"]))
df_clean = df.drop_duplicates()
df_clean = df.drop_duplicates(subset=["name"], keep="first")

# ── Type conversion ───────────────────────────────────
df["age"]    = df["age"].astype(float)
df["salary"] = pd.to_numeric(df["salary"], errors="coerce")
df["name"]   = df["name"].astype(str)

# ── String cleaning ───────────────────────────────────
df["name"] = df["name"].str.strip()
df["name"] = df["name"].str.lower()
df["dept"] = df["dept"].str.replace("Eng","Engineering")
df["name"] = df["name"].str.title()

# ── Rename columns ────────────────────────────────────
df = df.rename(columns={"dept": "department", "age": "years"})

# ── Reset index ───────────────────────────────────────
df = df.reset_index(drop=True)

import pandas as pd
import numpy as np

df = pd.DataFrame({
    "name":   ["Alice","Bob","Carol","Dave","Eve","Frank"],
    "dept":   ["Eng","Mkt","Eng","HR","Eng","Mkt"],
    "salary": [95000,72000,88000,58000,110000,65000],
    "age":    [28,35,31,27,42,29],
    "active": [True,True,True,False,True,True],
})

# ── Basic groupby ─────────────────────────────────────
grouped = df.groupby("dept")

print(grouped["salary"].mean())     # avg salary per dept
print(grouped["salary"].sum())      # total salary per dept
print(grouped["salary"].count())    # headcount per dept
print(grouped.size())               # same as count

# ── Multiple aggregations ─────────────────────────────
result = grouped["salary"].agg(["mean","min","max","std"])
print(result)

# ── agg with dict — different agg per column ──────────
result = grouped.agg({
    "salary": ["mean","max"],
    "age":    ["mean","min"],
    "active": "sum",
})
print(result)

# ── Named aggregations (clean column names) ───────────
result = df.groupby("dept").agg(
    avg_salary  = ("salary", "mean"),
    max_salary  = ("salary", "max"),
    headcount   = ("name",   "count"),
    avg_age     = ("age",    "mean"),
).round(0)
print(result)

# ── Transform — broadcast group stats back ────────────
df["dept_avg_salary"] = df.groupby("dept")["salary"].transform("mean")
df["salary_vs_avg"]   = df["salary"] - df["dept_avg_salary"]

# ── Filter groups ─────────────────────────────────────
# Keep only departments with more than 1 employee
big_depts = df.groupby("dept").filter(lambda g: len(g) > 1)

# ── Apply custom function ─────────────────────────────
def dept_summary(group):
    return pd.Series({
        "count":      len(group),
        "avg_salary": group["salary"].mean(),
        "top_earner": group.loc[group["salary"].idxmax(), "name"],
    })

summary = df.groupby("dept").apply(dept_summary)
print(summary)

import pandas as pd

employees = pd.DataFrame({
    "emp_id": [1, 2, 3, 4],
    "name":   ["Alice","Bob","Carol","Dave"],
    "dept_id":[10, 20, 10, 30],
})
departments = pd.DataFrame({
    "dept_id":  [10, 20, 40],
    "dept_name":["Engineering","Marketing","Finance"],
})
salaries = pd.DataFrame({
    "emp_id": [1, 2, 3],
    "salary": [95000, 72000, 88000],
})

# ── merge — like SQL JOIN ─────────────────────────────
inner = pd.merge(employees, departments, on="dept_id", how="inner")
left  = pd.merge(employees, departments, on="dept_id", how="left")
outer = pd.merge(employees, departments, on="dept_id", how="outer")

# Merge on different column names
result = pd.merge(employees, salaries,
                  left_on="emp_id", right_on="emp_id")

# Chain merges
full = (employees
        .merge(departments, on="dept_id", how="left")
        .merge(salaries,    on="emp_id",  how="left"))
print(full)

# ── concat — stack DataFrames ─────────────────────────
df1 = pd.DataFrame({"a":[1,2],"b":[3,4]})
df2 = pd.DataFrame({"a":[5,6],"b":[7,8]})
stacked = pd.concat([df1, df2], ignore_index=True)
side_by_side = pd.concat([df1, df2], axis=1)

# ── pivot_table ───────────────────────────────────────
sales = pd.DataFrame({
    "month":   ["Jan","Jan","Feb","Feb","Mar","Mar"],
    "product": ["A","B","A","B","A","B"],
    "revenue": [100,200,150,180,120,220],
})
pivot = sales.pivot_table(
    values="revenue", index="month",
    columns="product", aggfunc="sum", fill_value=0
)
print(pivot)

# ── melt — wide to long format ────────────────────────
wide = pd.DataFrame({
    "name":  ["Alice","Bob"],
    "jan":   [100, 200],
    "feb":   [150, 180],
})
long = wide.melt(id_vars="name",
                 var_name="month", value_name="sales")
print(long)

# pip install matplotlib seaborn

import matplotlib.pyplot as plt
import numpy as np

# ── Figure and Axes ───────────────────────────────────
fig, ax = plt.subplots(figsize=(10, 6))

# Line plot
x = np.linspace(0, 2*np.pi, 100)
ax.plot(x, np.sin(x), label="sin(x)", color="blue",  linewidth=2)
ax.plot(x, np.cos(x), label="cos(x)", color="red",   linestyle="--")
ax.set_title("Sine and Cosine", fontsize=16)
ax.set_xlabel("x"); ax.set_ylabel("y")
ax.legend(); ax.grid(True, alpha=0.3)
plt.tight_layout(); plt.show()

# ── Multiple subplots ─────────────────────────────────
fig, axes = plt.subplots(2, 2, figsize=(12, 8))

# Bar chart
categories = ["Eng","Mkt","HR","Finance"]
values      = [95000, 72000, 58000, 88000]
axes[0,0].bar(categories, values, color=["#3498db","#e74c3c","#2ecc71","#f39c12"])
axes[0,0].set_title("Avg Salary by Department")
axes[0,0].set_ylabel("Salary ($)")

# Histogram
data = np.random.normal(50000, 15000, 1000)
axes[0,1].hist(data, bins=30, color="#3498db", edgecolor="white", alpha=0.8)
axes[0,1].set_title("Salary Distribution")
axes[0,1].set_xlabel("Salary")

# Scatter plot
x = np.random.rand(100) * 10
y = 2 * x + np.random.randn(100)
axes[1,0].scatter(x, y, alpha=0.6, c=y, cmap="viridis")
axes[1,0].set_title("Scatter Plot")

# Pie chart
sizes  = [35, 25, 20, 20]
labels = ["Eng","Mkt","HR","Finance"]
axes[1,1].pie(sizes, labels=labels, autopct="%1.1f%%",
              startangle=90)
axes[1,1].set_title("Headcount by Department")

plt.tight_layout(); plt.show()

# pip install seaborn

import seaborn as sns
import matplotlib.pyplot as plt
import pandas as pd
import numpy as np

sns.set_theme(style="whitegrid", palette="husl")

# Sample data
np.random.seed(42)
df = pd.DataFrame({
    "salary": np.random.normal(80000, 20000, 200).astype(int),
    "age":    np.random.randint(22, 60, 200),
    "dept":   np.random.choice(["Eng","Mkt","HR","Finance"], 200),
    "score":  np.random.uniform(60, 100, 200).round(1),
    "active": np.random.choice([True, False], 200),
})

fig, axes = plt.subplots(2, 3, figsize=(16, 10))

# Box plot — distribution by category
sns.boxplot(data=df, x="dept", y="salary", ax=axes[0,0])
axes[0,0].set_title("Salary by Department")

# Violin plot — distribution shape
sns.violinplot(data=df, x="dept", y="score", ax=axes[0,1])
axes[0,1].set_title("Score Distribution by Dept")

# Scatter with regression line
sns.regplot(data=df, x="age", y="salary",
            scatter_kws={"alpha":0.4}, ax=axes[0,2])
axes[0,2].set_title("Age vs Salary")

# Count plot
sns.countplot(data=df, x="dept", hue="active", ax=axes[1,0])
axes[1,0].set_title("Headcount by Dept and Status")

# Heatmap — correlation matrix
numeric_cols = df.select_dtypes(include=np.number)
corr = numeric_cols.corr()
sns.heatmap(corr, annot=True, fmt=".2f", cmap="coolwarm",
            ax=axes[1,1], vmin=-1, vmax=1)
axes[1,1].set_title("Correlation Matrix")

# KDE plot — density
for dept in df["dept"].unique():
    subset = df[df["dept"] == dept]["salary"]
    sns.kdeplot(subset, label=dept, ax=axes[1,2])
axes[1,2].set_title("Salary Density by Dept")
axes[1,2].legend()

plt.tight_layout(); plt.show()

import pandas as pd
import numpy as np

# ── Create date ranges ────────────────────────────────
dates  = pd.date_range("2024-01-01", periods=365, freq="D")
months = pd.date_range("2024-01-01", periods=12,  freq="MS")
hours  = pd.date_range("2024-01-01", periods=24,  freq="H")

# ── Build time series ─────────────────────────────────
np.random.seed(42)
ts = pd.Series(
    np.cumsum(np.random.randn(365)) + 100,
    index=dates,
    name="price"
)

# ── Indexing by date ──────────────────────────────────
print(ts["2024-03"])            # all of March
print(ts["2024-01":"2024-03"])  # Jan through Mar
print(ts.loc["2024-06-15"])     # specific date

# ── Resampling — change frequency ─────────────────────
monthly_mean = ts.resample("ME").mean()   # monthly average
weekly_sum   = ts.resample("W").sum()     # weekly total
quarterly    = ts.resample("QE").agg(["mean","min","max"])

# ── Rolling windows ───────────────────────────────────
ts_df = ts.to_frame()
ts_df["ma_7"]  = ts.rolling(window=7).mean()   # 7-day MA
ts_df["ma_30"] = ts.rolling(window=30).mean()  # 30-day MA
ts_df["std_7"] = ts.rolling(window=7).std()    # rolling std

# ── Lag features ──────────────────────────────────────
ts_df["lag_1"]  = ts.shift(1)    # yesterday's value
ts_df["lag_7"]  = ts.shift(7)    # last week
ts_df["diff_1"] = ts.diff(1)     # daily change
ts_df["pct_chg"]= ts.pct_change()# % daily change

# ── Date components ───────────────────────────────────
df = pd.DataFrame({"date": dates, "value": ts.values})
df["year"]      = df["date"].dt.year
df["month"]     = df["date"].dt.month
df["day"]       = df["date"].dt.day
df["weekday"]   = df["date"].dt.day_name()
df["quarter"]   = df["date"].dt.quarter
df["is_weekend"]= df["date"].dt.dayofweek >= 5

# ── Seasonal analysis ─────────────────────────────────
monthly_avg = df.groupby("month")["value"].mean()
weekday_avg = df.groupby("weekday")["value"].mean()
print("Monthly averages:\n", monthly_avg.round(2))

import pandas as pd
import numpy as np

# ── Step 1: Generate synthetic dataset ───────────────
np.random.seed(42)
n = 500
raw = pd.DataFrame({
    "emp_id":    range(1, n+1),
    "name":      [f"Employee_{i}" for i in range(1, n+1)],
    "age":       np.random.randint(22, 65, n),
    "salary":    np.random.normal(75000, 20000, n).round(0),
    "dept":      np.random.choice(["Eng","Mkt","HR","Finance","Sales"], n),
    "years_exp": np.random.randint(0, 30, n),
    "score":     np.random.uniform(50, 100, n).round(1),
    "active":    np.random.choice([True, False], n, p=[0.85, 0.15]),
    "joined":    pd.date_range("2010-01-01", periods=n, freq="W"),
})
# Introduce noise
raw.loc[raw.sample(20).index, "salary"] = np.nan
raw.loc[raw.sample(10).index, "dept"]   = None

# ── Step 2: Clean ─────────────────────────────────────
df = raw.copy()
df["salary"] = df["salary"].fillna(df["salary"].median())
df["dept"]   = df["dept"].fillna("Unknown")
df["salary"] = df["salary"].clip(lower=20000, upper=200000)
df = df.drop_duplicates(subset=["emp_id"])
df = df.reset_index(drop=True)

# ── Step 3: Feature engineering ───────────────────────
df["salary_band"] = pd.cut(df["salary"],
    bins=[0, 50000, 75000, 100000, float("inf")],
    labels=["Low","Mid","High","Senior"])
df["tenure_years"] = (pd.Timestamp.now() - df["joined"]).dt.days // 365
df["salary_per_exp"] = (df["salary"] / (df["years_exp"] + 1)).round(0)
df["is_senior"]      = df["years_exp"] >= 10

# ── Step 4: Analysis ──────────────────────────────────
print("=== Dataset Overview ===")
print(f"Rows: {len(df):,}  |  Columns: {df.shape[1]}")
print(f"Missing values: {df.isnull().sum().sum()}")

print("\n=== Department Summary ===")
summary = df.groupby("dept").agg(
    headcount    = ("emp_id",   "count"),
    avg_salary   = ("salary",   "mean"),
    avg_score    = ("score",    "mean"),
    active_count = ("active",   "sum"),
    avg_exp      = ("years_exp","mean"),
).round(1)
print(summary.sort_values("avg_salary", ascending=False))

print("\n=== Salary Band Distribution ===")
band_dist = df["salary_band"].value_counts().sort_index()
for band, count in band_dist.items():
    bar = "█" * (count // 5)
    print(f"  {band:<8} {count:>4}  {bar}")

print("\n=== Top 5 Earners ===")
top5 = df.nlargest(5, "salary")[["name","dept","salary","score"]]
print(top5.to_string(index=False))

# ── Step 5: Correlations ──────────────────────────────
numeric = df[["age","salary","years_exp","score","tenure_years"]]
corr    = numeric.corr()["salary"].drop("salary").sort_values(ascending=False)
print("\n=== Salary Correlations ===")
for col, val in corr.items():
    direction = "+" if val > 0 else "-"
    bar = "█" * int(abs(val) * 20)
    print(f"  {col:<15} {direction}{abs(val):.3f}  {bar}")