import sys

a = []  # refcount = 1
b = a   # refcount = 2
c = a   # refcount = 3

print(sys.getrefcount(a))  # 4 (includes temporary reference in getrefcount)

del b   # refcount = 2
del c   # refcount = 1
del a   # refcount = 0 -> IMMEDIATELY freed

import gc

class Node:
    def __init__(self):
        self.ref = None

# Create circular reference
a = Node()
b = Node()
a.ref = b
b.ref = a  # Circular!

del a
del b
# Reference counting can't free these (refcount never hits 0)
# Cyclic GC will detect and collect them

gc.collect()  # Force collection
print(gc.get_count())  # (generation0, generation1, generation2)

# BAD: Circular reference with __del__
class BadClass:
    def __del__(self):
        pass  # Prevents GC from collecting cycles!

# GOOD: Use weakref for circular references
import weakref

class GoodClass:
    def __init__(self):
        self.ref = None  # Use weakref.ref() for circular refs

class Singleton:
    _instance = None
    
    def __new__(cls, *args, **kwargs):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance
    
    def __init__(self, value):
        self.value = value  # WARNING: runs every time Singleton() is called!

s1 = Singleton(1)
s2 = Singleton(2)
print(s1 is s2)    # True
print(s1.value)    # 2 -- __init__ ran again and overwrote!

class A:
    def method(self):
        print("A")

class B(A):
    def method(self):
        print("B")

class C(A):
    def method(self):
        print("C")

class D(B, C):  # Diamond inheritance!
    pass

# Which method() gets called?
d = D()
d.method()  # "B" (follows MRO)

print(D.mro())
# [D, B, C, A, object]
# Left-to-right, depth-first, but respecting order

class A:
    def method(self):
        print("A")

class B(A):
    def method(self):
        print("B")
        super().method()  # Calls C.method, NOT A.method!

class C(A):
    def method(self):
        print("C")
        super().method()

class D(B, C):
    def method(self):
        print("D")
        super().method()

D().method()  # D -> B -> C -> A (follows MRO chain)

def log(func):
    def wrapper(*args):
        print("Call")
        return func(*args)
    return wrapper

@log
def add(): pass
# Equivalent to: add = log(add)

def repeat(times):
    def decorator(func):
        def wrapper(*args, **kwargs):
            for _ in range(times):
                result = func(*args, **kwargs)
            return result
        return wrapper
    return decorator

@repeat(times=3)
def greet(name):
    print(f"Hello {name}")

greet("Alice")  # Prints 3 times

def singleton(cls):
    instances = {}
    def get_instance(*args, **kwargs):
        if cls not in instances:
            instances[cls] = cls(*args, **kwargs)
        return instances[cls]
    return get_instance

@singleton
class Database:
    pass

db1 = Database()
db2 = Database()
assert db1 is db2  # Same instance!

from functools import wraps

def my_decorator(func):
    @wraps(func)  # Preserves __name__, __doc__, __module__, __qualname__, __dict__
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

# Generator pipeline for processing a large CSV
def read_lines(filepath):
    with open(filepath) as f:
        for line in f:
            yield line.strip()

def parse_csv(lines):
    for line in lines:
        yield line.split(',')

def filter_active(rows):
    for row in rows:
        if row[2] == 'active':
            yield row

# Entire pipeline processes ONE row at a time -- constant memory
pipeline = filter_active(parse_csv(read_lines('users.csv')))
for row in pipeline:
    process(row)

def flatten(nested):
    for item in nested:
        if isinstance(item, list):
            yield from flatten(item)  # Delegates to sub-generator
        else:
            yield item

def accumulator():
    total = 0
    while True:
        value = yield total
        if value is None:
            break
        total += value

gen = accumulator()
next(gen)        # Prime the generator -> 0
gen.send(10)     # -> 10
gen.send(20)     # -> 30

class DatabaseConnection:
    def __enter__(self):
        self.conn = psycopg2.connect(DSN)
        return self.conn
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is not None:
            self.conn.rollback()
        else:
            self.conn.commit()
        self.conn.close()
        return False  # Don't suppress exceptions

from contextlib import contextmanager

@contextmanager
def timer(label):
    start = time.perf_counter()
    try:
        yield  # Code inside `with` block runs here
    finally:
        elapsed = time.perf_counter() - start
        print(f"{label}: {elapsed:.3f}s")

with timer("query"):
    db.execute("SELECT * FROM users")

# GOOD: Short, clear intent
users.sort(key=lambda u: u.last_login)
max(products, key=lambda p: p.price)

# BAD: Complex lambda that should be a named function
process = lambda x: (x.strip().lower().replace('-', '_') if x else 'default')
# Better:
def normalize_key(x):
    """Normalize input string to snake_case key."""
    if not x:
        return 'default'
    return x.strip().lower().replace('-', '_')

# Bug: all lambdas capture the SAME variable `i`
funcs = [lambda: i for i in range(5)]
print([f() for f in funcs])  # [4, 4, 4, 4, 4] -- NOT [0, 1, 2, 3, 4]

# Fix: capture by default argument
funcs = [lambda i=i: i for i in range(5)]
print([f() for f in funcs])  # [0, 1, 2, 3, 4]

def f(positional, /, normal, *, keyword_only, **kwargs):
    pass
# positional: positional-only (Python 3.8+, before /)
# normal: positional or keyword
# keyword_only: keyword-only (after *)
# **kwargs: remaining keyword arguments

# 1. Decorator pass-through (the most common use)
def decorator(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        # Can inspect/modify args before passing through
        return func(*args, **kwargs)
    return wrapper

# 2. Configuration merging
def create_connection(host, port, **options):
    defaults = {'timeout': 30, 'retry': 3, 'ssl': True}
    config = {**defaults, **options}  # options override defaults
    return connect(host, port, **config)

# 3. Forwarding to parent class
class CustomWidget(BaseWidget):
    def __init__(self, color, *args, **kwargs):
        self.color = color
        super().__init__(*args, **kwargs)  # Forward everything else

import copy

l1 = [[1, 2], 3, 4]
l2 = copy.copy(l1)    # Shallow: l2[0] POINTS TO same inner list as l1[0]
l3 = copy.deepcopy(l1)# Deep: l3[0] is a completely independent NEW list

l1[0].append(99)
print(l2[0])  # [1, 2, 99] -- AFFECTED by mutation!
print(l3[0])  # [1, 2]     -- NOT affected

class Connection:
    def __copy__(self):
        # Shallow copy: share the socket, copy config
        new = Connection.__new__(Connection)
        new.config = self.config.copy()
        new.socket = self.socket  # Shared!
        return new
    
    def __deepcopy__(self, memo):
        # Deep copy: new socket, new config
        new = Connection.__new__(Connection)
        new.config = copy.deepcopy(self.config, memo)
        new.socket = create_new_socket()  # Fresh!
        return new

class BankAccount:
    def __init__(self):
        self.holder = "Alice"       # Public
        self._balance = 1000        # Protected by convention
        self.__pin = 1234           # Name-mangled to _BankAccount__pin

acc = BankAccount()
print(acc.holder)              # Fine
print(acc._balance)            # Works (just a warning to other devs)
print(acc._BankAccount__pin)   # Works! Name mangling is NOT security

class Base:
    def __init__(self):
        self.__id = 1  # Becomes _Base__id

class Child(Base):
    def __init__(self):
        super().__init__()
        self.__id = 2  # Becomes _Child__id -- no collision!

from abc import ABC, abstractmethod

class PaymentProcessor(ABC):
    @abstractmethod
    def charge(self, amount: float) -> bool:
        """Process a payment. Returns True on success."""
        pass
    
    @abstractmethod
    def refund(self, transaction_id: str) -> bool:
        pass
    
    def validate_amount(self, amount: float) -> bool:
        """Concrete method -- shared logic all processors use."""
        return amount > 0 and amount < 100_000

class StripeProcessor(PaymentProcessor):
    def charge(self, amount):
        # Must implement or TypeError on instantiation
        return stripe.Charge.create(amount=amount)
    
    def refund(self, transaction_id):
        return stripe.Refund.create(charge=transaction_id)

from typing import Protocol

class Drawable(Protocol):
    def draw(self) -> None: ...

# No inheritance needed -- if it has draw(), it's Drawable
class Circle:
    def draw(self) -> None:
        print("drawing circle")

def render(shape: Drawable):  # mypy checks this
    shape.draw()

import sys

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

class WithSlots:
    __slots__ = ['x', 'y']
    def __init__(self, x, y):
        self.x = x
        self.y = y

d = WithDict(1, 2)
s = WithSlots(1, 2)

# sys.getsizeof does not include __dict__ contents
# Total with dict: ~152 bytes per instance
# Total with slots: ~56 bytes per instance
# At 10M objects: 1.52 GB vs 560 MB -- saves ~1 GB of RAM

class User:
    def __init__(self, email):
        self._email = None
        self.email = email  # Triggers the setter with validation!
    
    @property
    def email(self):
        return self._email
    
    @email.setter
    def email(self, value):
        if '@' not in value:
            raise ValueError(f"Invalid email: {value}")
        self._email = value.lower().strip()
    
    @email.deleter
    def email(self):
        self._email = None

class Base:
    @property
    def value(self):
        return self._value

class Child(Base):
    @Base.value.setter  # Must reference Base.value explicitly!
    def value(self, val):
        self._value = val

# Modern Python typing (3.10+)
def process(items: list[str | int]) -> dict[str, int]:
    result: dict[str, int] = {}
    for item in items:
        match item:
            case str() as s:
                result[s] = len(s)
            case int() as n:
                result[str(n)] = n
    return result

from typing import Protocol

class Renderable(Protocol):
    def render(self) -> str: ...

# Any class with render() -> str satisfies this, no inheritance needed

# BAD: Deep inheritance hierarchy
class Animal:
    def move(self): ...
class FlyingAnimal(Animal):
    def fly(self): ...
class SwimmingAnimal(Animal):
    def swim(self): ...
class Duck(FlyingAnimal, SwimmingAnimal):  # Diamond! Fragile!
    pass

# GOOD: Composition with strategy pattern
class Animal:
    def __init__(self, movement_strategy, sound_strategy):
        self.movement = movement_strategy
        self.sound = sound_strategy

duck = Animal(
    movement_strategy=FlyAndSwim(),
    sound_strategy=Quack()
)

from functools import singledispatch

@singledispatch
def process(data):
    raise NotImplementedError(f"Cannot process {type(data)}")

@process.register(str)
def _(data):
    return data.upper()

@process.register(list)
def _(data):
    return [process(item) for item in data]

@process.register(int)
def _(data):
    return data * 2

process("hello")  # "HELLO"
process(42)       # 84
process([1, 2])   # [2, 4]

from typing import overload

@overload
def get(key: str) -> str: ...
@overload
def get(key: int) -> int: ...

def get(key):  # Actual implementation
    return cache[key]

# db.py
_connection = DatabaseConnection()

def get_connection():
    return _connection

class Singleton:
    _instance = None
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

# INSTEAD of a singleton DB:
def create_user(db: Database, user_data: dict):
    db.execute("INSERT INTO users ...")

# In tests: create_user(mock_db, test_data)
# In production: create_user(real_db, user_data)

from enum import Enum, auto, IntEnum

class Status(Enum):
    PENDING = auto()   # 1
    ACTIVE = auto()    # 2
    DISABLED = auto()  # 3

# Type-safe comparisons
status = Status.ACTIVE
status == Status.ACTIVE   # True
status == 2               # False! (not equal to int)
status == "ACTIVE"        # False! (not equal to string)

# Iteration
for s in Status:
    print(s.name, s.value)  # PENDING 1, ACTIVE 2, DISABLED 3

class Color(Enum):
    RED = "#FF0000"
    GREEN = "#00FF00"
    BLUE = "#0000FF"
    
    @property
    def rgb_tuple(self):
        hex_val = self.value.lstrip('#')
        return tuple(int(hex_val[i:i+2], 16) for i in (0, 2, 4))

Color.RED.rgb_tuple  # (255, 0, 0)

from dataclasses import dataclass, field

@dataclass(frozen=True)  # Immutable + hashable
class Point:
    x: float
    y: float

@dataclass
class User:
    name: str
    age: int
    tags: list = field(default_factory=list)  # MUST use field() for mutable defaults
    
    def __post_init__(self):
        if self.age < 0:
            raise ValueError("Age must be non-negative")

import time
import threading
import multiprocessing

def cpu_bound_task(n):
    """CPU-intensive: calculate sum"""
    total = 0
    for i in range(n):
        total += i ** 2
    return total

# Single-threaded baseline
start = time.time()
cpu_bound_task(10_000_000)
print(f"Single thread: {time.time() - start:.2f}s")  # ~1.5s

# Multi-threading (GIL limits this!)
def threaded_approach():
    threads = []
    for _ in range(4):
        t = threading.Thread(target=cpu_bound_task, args=(10_000_000,))
        threads.append(t)
        t.start()
    for t in threads:
        t.join()

start = time.time()
threaded_approach()
print(f"4 threads: {time.time() - start:.2f}s")  # ~1.5s (NO speedup!)

# Multi-processing (bypasses GIL!)
def multiprocess_approach():
    with multiprocessing.Pool(4) as pool:
        pool.map(cpu_bound_task, [10_000_000] * 4)

start = time.time()
multiprocess_approach()
print(f"4 processes: {time.time() - start:.2f}s")  # ~0.4s (4x speedup!)

import requests
import threading

def fetch_url(url):
    response = requests.get(url)
    return len(response.content)

urls = ['https://example.com'] * 10

# Single-threaded
start = time.time()
for url in urls:
    fetch_url(url)
print(f"Sequential: {time.time() - start:.2f}s")  # ~5s

# Multi-threaded (GIL released during I/O!)
threads = []
start = time.time()
for url in urls:
    t = threading.Thread(target=fetch_url, args=(url,))
    threads.append(t)
    t.start()
for t in threads:
    t.join()
print(f"Threaded: {time.time() - start:.2f}s")  # ~0.5s (10x speedup!)

import asyncio

async def fetch(url, delay):
    print(f"Start {url}")
    await asyncio.sleep(delay)  # Yields control to loop
    print(f"Done {url}")
    return url

async def main():
    # These run concurrently, NOT sequentially
    results = await asyncio.gather(
        fetch("A", 2),
        fetch("B", 1),
        fetch("C", 3),
    )
    # Total time: ~3s (not 6s), because they overlap during sleep

asyncio.run(main())

async def transfer(from_acct, to_acct, amount):
    balance = from_acct.balance  # Atomic
    # NO other task can run here (no await between reads)
    if balance >= amount:
        from_acct.balance -= amount
        # If another task modifies from_acct between these two awaits, you have a bug
        await db.save(from_acct)  # SUSPENSION POINT: other tasks run here!
        to_acct.balance += amount
        await db.save(to_acct)

balance = 100

async def withdraw(amount):
    global balance
    current = balance
    await asyncio.sleep(0)  # Yield to event loop -- another task can run!
    balance = current - amount

# Both tasks read balance=100, both write back 100-50=50
# Expected final balance: 0, Actual: 50 (lost update!)
await asyncio.gather(withdraw(50), withdraw(50))
print(balance)  # 50, not 0!

lock = asyncio.Lock()

async def safe_withdraw(amount):
    global balance
    async with lock:
        current = balance
        await asyncio.sleep(0)
        balance = current - amount

await asyncio.gather(safe_withdraw(50), safe_withdraw(50))
print(balance)  # 0, correct!

# DISASTER: blocks the ENTIRE event loop for 10 seconds
async def handler(request):
    time.sleep(10)  # BLOCKING! Not async!
    return Response("done")
# During these 10 seconds, ALL other requests queue up

# FIX 1: Use async equivalent
async def handler(request):
    await asyncio.sleep(10)  # Non-blocking. Loop runs other tasks.
    return Response("done")

# FIX 2: Offload CPU-bound work to a thread pool
async def handler(request):
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(None, cpu_heavy_function, arg)
    return Response(result)

# FIX 3: Offload to process pool for true CPU parallelism
from concurrent.futures import ProcessPoolExecutor
executor = ProcessPoolExecutor(max_workers=4)

async def handler(request):
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(executor, cpu_heavy_function, arg)
    return Response(result)

results = await asyncio.gather(
    fetch("url1"),
    fetch("url2"),
    fetch("url3"),
    return_exceptions=True,  # Don't crash on first failure
)
# results = ["data1", "data2", ConnectionError(...)]

async with asyncio.TaskGroup() as tg:
    task1 = tg.create_task(fetch("url1"))
    task2 = tg.create_task(fetch("url2"))
    task3 = tg.create_task(fetch("url3"))
# All tasks guaranteed complete (or cancelled) when exiting the block
# If any task raises, all others are cancelled and ExceptionGroup is raised

class AsyncDBPool:
    async def __aenter__(self):
        self.pool = await asyncpg.create_pool(DSN)
        return self.pool
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.pool.close()

async with AsyncDBPool() as pool:
    result = await pool.fetch("SELECT * FROM users")

from contextlib import asynccontextmanager

@asynccontextmanager
async def get_connection():
    conn = await asyncpg.connect(DSN)
    try:
        yield conn
    finally:
        await conn.close()

class AsyncPaginator:
    def __init__(self, url):
        self.url = url
        self.page = 0
    
    def __aiter__(self):
        return self
    
    async def __anext__(self):
        self.page += 1
        data = await fetch_page(self.url, self.page)
        if not data:
            raise StopAsyncIteration
        return data

async for page in AsyncPaginator("/api/users"):
    process(page)

async def stream_events(channel):
    async with connect_to_redis() as redis:
        while True:
            event = await redis.blpop(channel)
            yield event

async for event in stream_events("notifications"):
    handle(event)

import uvloop
# Option 1: Set as default policy
uvloop.install()

# Option 2: Explicit (legacy)
import asyncio
loop = uvloop.new_event_loop()
asyncio.set_event_loop(loop)

def application(environ, start_response):
    start_response('200 OK', [('Content-Type', 'text/plain')])
    return [b'Hello World']

async def application(scope, receive, send):
    await send({
        'type': 'http.response.start',
        'status': 200,
        'headers': [[b'content-type', b'text/plain']],
    })
    await send({
        'type': 'http.response.body',
        'body': b'Hello World',
    })

from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel, EmailStr

class UserCreate(BaseModel):
    email: EmailStr
    name: str
    age: int  # Auto-validates, auto-documents

app = FastAPI()

async def get_db():
    db = await create_connection()
    try:
        yield db  # Generator-based DI with cleanup
    finally:
        await db.close()

@app.post("/users", status_code=201)
async def create_user(user: UserCreate, db = Depends(get_db)):
    # user is already validated Pydantic model
    return await db.insert_user(user.model_dump())

# N+1 QUERIES: 1 query for books + N queries for authors
books = Book.objects.all()  # SELECT * FROM books (1 query)
for book in books:
    print(book.author.name)  # SELECT * FROM authors WHERE id=? (N queries!)
# 1000 books = 1001 queries. Each query has ~1-5ms network round trip.
# Total: 1-5 SECONDS for what should be a 10ms query.

# select_related: SQL JOIN (for ForeignKey / OneToOne)
books = Book.objects.select_related('author').all()
# SELECT books.*, authors.* FROM books JOIN authors ON ... (1 query!)

# prefetch_related: Separate query + Python-side join (for ManyToMany / Reverse FK)
authors = Author.objects.prefetch_related('books').all()
# SELECT * FROM authors; SELECT * FROM books WHERE author_id IN (...) (2 queries)

# Prefetch with custom queryset
from django.db.models import Prefetch
Author.objects.prefetch_related(
    Prefetch('books', queryset=Book.objects.filter(published=True).order_by('-date'))
)

from fastapi import FastAPI, Request
import time
import uuid

app = FastAPI()

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    request_id = str(uuid.uuid4())
    request.state.request_id = request_id
    
    start = time.perf_counter()
    response = await call_next(request)
    duration = time.perf_counter() - start
    
    response.headers["X-Request-ID"] = request_id
    response.headers["X-Process-Time"] = f"{duration:.4f}"
    
    logger.info(f"[{request_id}] {request.method} {request.url.path} -> {response.status_code} ({duration:.3f}s)")
    return response

<!-- In Django templates -->
<form method="POST">
    {% csrf_token %}
    <input type="text" name="amount" value="1000">
    <button>Transfer</button>
</form>

// Read token from cookie, send in header
const csrftoken = document.cookie.match(/csrftoken=([^;]+)/)?.[1];
fetch('/api/transfer', {
    method: 'POST',
    headers: {'X-CSRFToken': csrftoken},
    body: JSON.stringify({amount: 1000})
});

# Production command
gunicorn app:app \
    --workers 4 \
    --worker-class uvicorn.workers.UvicornWorker \
    --bind 0.0.0.0:8000 \
    --timeout 120 \
    --graceful-timeout 30 \
    --max-requests 10000 \
    --max-requests-jitter 1000 \
    --access-logfile -

# Alembic migration example
def upgrade():
    # SAFE: Adding nullable column does not lock table
    op.add_column('users', sa.Column('phone', sa.String(20), nullable=True))

def downgrade():
    op.drop_column('users', 'phone')

# DANGEROUS: Adding NOT NULL column with default on large table
# This rewrites the entire table in PostgreSQL < 11!
def upgrade():
    op.add_column('users', sa.Column('status', sa.String(10), 
                                      nullable=False, server_default='active'))

# Server creates session, stores in Redis
session_id = uuid4()
redis.set(f"session:{session_id}", json.dumps({"user_id": 42}), ex=3600)
response.set_cookie("session_id", session_id, httponly=True, secure=True, samesite="Lax")

import jwt
token = jwt.encode(
    {"user_id": 42, "exp": datetime.utcnow() + timedelta(hours=1)},
    SECRET_KEY, algorithm="HS256"
)
# Client sends: Authorization: Bearer <token>

from fastapi import Depends

async def get_db():
    db = SessionLocal()
    try:
        yield db  # Generator-based: cleanup runs after request
    finally:
        db.close()

async def get_current_user(token: str = Depends(oauth2_scheme), db = Depends(get_db)):
    user = await db.query(User).filter(User.id == decode_token(token).sub).first()
    if not user:
        raise HTTPException(401)
    return user

@app.get("/profile")
async def profile(user: User = Depends(get_current_user)):
    return user  # Both DB and auth handled by DI chain

# Override dependencies in tests
async def mock_get_db():
    return FakeDB()

async def mock_current_user():
    return User(id=1, name="test")

app.dependency_overrides[get_db] = mock_get_db
app.dependency_overrides[get_current_user] = mock_current_user

# BAD: Tight coupling, hard to test
from myapp.db import database  # Global singleton
def get_user(user_id):
    return database.query(User, user_id)  # How do you test without a real DB?

# GOOD: Dependency injection
def get_user(user_id, db: Database):
    return db.query(User, user_id)  # Pass FakeDB in tests

def flatten(lst):
    result = []
    for item in lst:
        if isinstance(item, list):
            result.extend(flatten(item))
        else:
            result.append(item)
    return result

flatten([1, [2, [3, [4]], 5]])  # [1, 2, 3, 4, 5]

def flatten_gen(lst):
    for item in lst:
        if isinstance(item, list):
            yield from flatten_gen(item)
        else:
            yield item

def flatten_iter(lst):
    stack = list(reversed(lst))
    result = []
    while stack:
        item = stack.pop()
        if isinstance(item, list):
            stack.extend(reversed(item))
        else:
            result.append(item)
    return result

from collections import OrderedDict

class LRUCache:
    def __init__(self, capacity: int):
        self.cache = OrderedDict()
        self.capacity = capacity
    
    def get(self, key):
        if key not in self.cache:
            return -1
        self.cache.move_to_end(key)  # Mark as recently used
        return self.cache[key]
    
    def put(self, key, value):
        if key in self.cache:
            self.cache.move_to_end(key)
        self.cache[key] = value
        if len(self.cache) > self.capacity:
            self.cache.popitem(last=False)  # Remove oldest (LRU)

# BAD: Loads entire file into memory
with open('50gb_log.txt') as f:
    data = f.read()  # OOM kill on a 4GB machine

# GOOD: Iterator reads one line at a time (~0 memory overhead)
with open('50gb_log.txt') as f:
    for line in f:
        process(line)

# GOOD: Chunked reading for binary files
with open('50gb.bin', 'rb') as f:
    while chunk := f.read(8192):  # 8KB chunks (walrus operator)
        process_chunk(chunk)

# GOOD: Memory-mapped files for random access
import mmap
with open('large.bin', 'rb') as f:
    mm = mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ)
    data = mm[1000:2000]  # Read bytes 1000-2000 without loading entire file

# CSV: Use iterator, not read all
import csv
with open('huge.csv') as f:
    for row in csv.DictReader(f):
        process(row)

# JSON Lines (JSONL): One JSON object per line
with open('events.jsonl') as f:
    for line in f:
        event = json.loads(line)

# Parquet: Use column pruning
import pyarrow.parquet as pq
table = pq.read_table('huge.parquet', columns=['name', 'age'])  # Only needed columns

data = [
    {"name": "Alice", "age": 30},
    {"name": "Bob", "age": 25},
    {"name": "Charlie", "age": 35},
]

# Lambda approach
data.sort(key=lambda x: x['age'])

# operator.itemgetter (faster for simple key access)
from operator import itemgetter
data.sort(key=itemgetter('age'))

# Multiple keys
data.sort(key=lambda x: (x['age'], x['name']))

# Reverse
data.sort(key=itemgetter('age'), reverse=True)

a = [1, 2, 3]
b = [1, 2, 3]
a == b   # True (same values)
a is b   # False (different objects in memory)

c = a
a is c   # True (same object)