try:
    result = 10 / divisor
except ZeroDivisionError as e:
    print(f"Error: {e}")
except (TypeError, ValueError):
    print("Type or value error")
else:
    print(f"Success: {result}")  # Chỉ khi không có exception
finally:
    print("Cleanup")  # Luôn chạy

# enumerate — loop có index
fruits = ["apple", "banana", "cherry"]
for i, fruit in enumerate(fruits, start=1):
    print(f"{i}. {fruit}")

# zip — gộp nhiều iterables song song
names = ["Alice", "Bob"]
scores = [95, 87]
for name, score in zip(names, scores):
    print(f"{name}: {score}")

# map — transform mỗi element (lazy)
doubled = list(map(lambda x: x * 2, [1, 2, 3]))  # [2, 4, 6]
# Prefer list comprehension: [x*2 for x in [1,2,3]]

# filter — lọc elements (lazy)
evens = list(filter(lambda x: x % 2 == 0, range(10)))
# Prefer: [x for x in range(10) if x % 2 == 0]

# List — toàn bộ vào RAM
squares_list = [x**2 for x in range(1_000_000)]

# Generator — lazy, từng phần tử khi cần
def squares_gen(n):
    for x in range(n):
        yield x**2

gen = squares_gen(1_000_000)
print(next(gen))  # 0 — chỉ tính khi gọi next()

import functools, time

def timer(func):
    @functools.wraps(func)  # Giữ __name__, __doc__
    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

@timer
def slow():
    time.sleep(1)

from contextlib import asynccontextmanager

# Dùng @asynccontextmanager cho async code (không phải @contextmanager)
@asynccontextmanager
async def db_transaction(session):
    try:
        yield session
        await session.commit()
    except Exception:
        await session.rollback()
        raise

# Class-based (sync)
class Timer:
    def __enter__(self):
        import time; self.start = time.perf_counter(); return self
    def __exit__(self, *args):
        self.elapsed = time.perf_counter() - self.start

# Closure — nhớ state
def make_counter(start=0):
    count = [start]
    def counter():
        count[0] += 1
        return count[0]
    return counter

c = make_counter()
print(c(), c(), c())  # 1, 2, 3

# Lambda — function tức thời
double = lambda x: x * 2

from functools import lru_cache

@lru_cache(maxsize=128)
def fibonacci(n: int) -> int:
    if n < 2: return n
    return fibonacci(n-1) + fibonacci(n-2)

# Không cache (default: cached forever)
@lru_cache(maxsize=None)  # Cache không giới hạn
def expensive_lookup(key): ...

fibonacci.cache_info()   # hits, misses, maxsize, currsize
fibonacci.cache_clear()  # Xóa cache

class User:
    _count = 0

    def __init__(self, name):
        self.name = name
        User._count += 1

    def greet(self):            # Instance method
        return f"Hi, {self.name}"

    @classmethod
    def from_dict(cls, data):   # Factory — dùng cls thay vì User
        return cls(data['name'])

    @staticmethod
    def validate_name(name):    # Utility — không cần self/cls
        return len(name) >= 2

class Temperature:
    def __init__(self, celsius: float):
        self._celsius = celsius

    @property
    def celsius(self) -> float:
        return self._celsius

    @celsius.setter
    def celsius(self, value: float):
        if value < -273.15:
            raise ValueError("Below absolute zero!")
        self._celsius = value

    @property
    def fahrenheit(self) -> float:  # Read-only derived property
        return self._celsius * 9/5 + 32

t = Temperature(25)
t.celsius = 30      # Gọi setter tự động
print(t.fahrenheit) # 86.0

from abc import ABC, abstractmethod

class Repository(ABC):
    @abstractmethod
    async def get_by_id(self, id: int): ...

    @abstractmethod
    async def create(self, data: dict): ...

    def find_all(self):          # Concrete method — có thể kế thừa
        return self.get_all()

class UserRepository(Repository):
    async def get_by_id(self, id): ...   # Bắt buộc implement
    async def create(self, data): ...    # Bắt buộc implement

# Repository()  # TypeError — cannot instantiate abstract class

from dataclasses import dataclass, field
from typing import List

@dataclass
class User:
    name: str
    email: str
    age: int = 0
    tags: List[str] = field(default_factory=list)
    # __init__, __repr__, __eq__ tự động được generate!

@dataclass(frozen=True)  # Immutable như tuple
class Point:
    x: float
    y: float

p = Point(1.0, 2.0)
# p.x = 3.0  # FrozenInstanceError

import asyncio, httpx

async def fetch(url: str) -> dict:
    async with httpx.AsyncClient() as client:
        r = await client.get(url)  # Yield control tại đây
        return r.json()

async def main():
    # Chạy 3 requests đồng thời (concurrent), không phải song song
    results = await asyncio.gather(
        fetch("/users/1"),
        fetch("/users/2"),
        fetch("/users/3"),
    )

asyncio.run(main())

from fastapi import Depends

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session   # cleanup sau khi request xong

class Pagination:
    def __init__(self, page: int = 1, size: int = 20):
        self.offset = (page - 1) * size
        self.limit = min(size, 100)

@app.get("/users")
async def list_users(
    db = Depends(get_db),           # Injected automatically
    pg: Pagination = Depends(),     # Class-based DI
    user = Depends(get_current_user)
):
    ...

from pydantic import BaseModel, EmailStr, Field, field_validator

class UserCreate(BaseModel):
    name: str = Field(min_length=2, max_length=50)
    email: EmailStr
    age: int = Field(ge=0, le=150)

    @field_validator('name')
    @classmethod
    def name_alpha(cls, v: str) -> str:
        if not v.replace(' ', '').isalpha():
            raise ValueError('Only letters allowed')
        return v.strip().title()

    model_config = {"from_attributes": True}  # Pydantic v2

@app.post("/users", status_code=201)
async def create_user(user: UserCreate):  # Auto-validated!
    ...

import time, uuid
from fastapi import Request

@app.middleware("http")
async def logging_middleware(request: Request, call_next):
    request_id = str(uuid.uuid4())
    start = time.perf_counter()

    response = await call_next(request)

    duration = time.perf_counter() - start
    response.headers["X-Request-ID"] = request_id
    response.headers["X-Duration"] = f"{duration:.4f}"
    return response

from passlib.context import CryptContext
import jwt  # pip install PyJWT
from datetime import datetime, timedelta, timezone

pwd = CryptContext(schemes=["bcrypt"])
SECRET = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

def create_token(user_id: int) -> str:
    expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    return jwt.encode({"sub": str(user_id), "exp": expire}, SECRET, ALGORITHM)

async def get_current_user(token = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET, algorithms=[ALGORITHM])
        uid = payload.get("sub")
        if not uid: raise credentials_exception
    except jwt.PyJWTError:
        raise HTTPException(401, "Invalid token")
    return await get_user(int(uid))

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "http://localhost:5173",  # Vite dev server
        "https://yourapp.com",    # Production
    ],
    allow_credentials=True,  # Cần cho cookies/auth headers
    allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"],
    allow_headers=["Authorization", "Content-Type"],
)

# N+1 problem — 1 + N queries
users = await db.execute(select(User))
for user in users.scalars():
    posts = await db.execute(  # 1 query mỗi user!
        select(Post).where(Post.user_id == user.id)
    )

# Fix: selectinload — 2 queries total
from sqlalchemy.orm import selectinload

result = await db.execute(
    select(User).options(selectinload(User.posts))
)
# 1 query lấy tất cả users + 1 query lấy posts cho tất cả user_ids

# joinedload — 1 query với JOIN (tốt khi result set nhỏ)
from sqlalchemy.orm import joinedload

from sqlalchemy.ext.asyncio import (
    create_async_engine, AsyncSession, async_sessionmaker
)
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/db",
    pool_size=10, max_overflow=20,
)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

class Base(DeclarativeBase): pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(unique=True, index=True)
    name: Mapped[str] = mapped_column(nullable=False)

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

from fastapi.testclient import TestClient
import pytest

def override_get_db():
    yield TestingSessionLocal()

app.dependency_overrides[get_db] = override_get_db
client = TestClient(app)

def test_create_user():
    res = client.post("/users", json={
        "name": "Alice", "email": "alice@test.com"
    })
    assert res.status_code == 201
    assert res.json()["email"] == "alice@test.com"

# Async test
@pytest.mark.asyncio
async def test_async():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        res = await ac.get("/users/1")
    assert res.status_code == 200

@pytest.fixture(scope="session")
def db_engine():   # Tạo 1 lần cho toàn bộ session
    engine = create_engine(TEST_URL)
    Base.metadata.create_all(engine)
    yield engine
    Base.metadata.drop_all(engine)

@pytest.fixture      # scope="function" — mỗi test
def db(db_engine):
    connection = db_engine.connect()
    transaction = connection.begin()
    session = Session(bind=connection)
    yield session
    session.close()
    transaction.rollback()  # Rollback sau mỗi test — isolation

async def test_register_sends_email(mocker):
    # Mock email service
    mock_send = mocker.patch(
        "app.services.email_service.send",
        new_callable=AsyncMock
    )
    mock_send.return_value = True

    await register_user(email="test@test.com")

    mock_send.assert_called_once_with(
        "test@test.com", "Welcome!"
    )

def test_with_side_effect(mocker):
    mock_api = mocker.patch("app.services.external_api.call")
    mock_api.side_effect = [{"status": "ok"}, ConnectionError()]

    assert call_api() == {"status": "ok"}
    with pytest.raises(ConnectionError):
        call_api()

from pydantic_settings import BaseSettings, SettingsConfigDict
from functools import lru_cache

class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env", env_file_encoding="utf-8"
    )
    app_name: str = "My API"
    debug: bool = False
    database_url: str          # Required — không có default
    secret_key: str
    redis_url: str = "redis://localhost:6379"
    access_token_expire_minutes: int = 30

@lru_cache  # Singleton — đọc .env một lần
def get_settings() -> Settings:
    return Settings()

settings = get_settings()

def log(level, *args, **kwargs):
    print(f"[{level}]", *args)
    for k, v in kwargs.items():
        print(f"  {k}: {v}")

log("INFO", "Server started", port=8080, debug=True)
# [INFO] Server started
#   port: 8080
#   debug: True

# Unpacking khi gọi function
def create_user(name, email, age):
    ...

data = {"name": "Alice", "email": "a@b.com", "age": 25}
create_user(**data)  # Unpack dict thành kwargs

# Trước 3.8 — gọi hàm 2 lần hoặc dùng temp variable
data = get_data()
if data:
    process(data)

# Với walrus — assign và kiểm tra cùng lúc
if data := get_data():
    process(data)

# Đặc biệt hữu ích trong while loop
while chunk := file.read(8192):
    process(chunk)

# List comprehension với filter + transform
results = [y for x in data if (y := transform(x)) > 0]
# Tránh gọi transform() 2 lần (1 cho filter, 1 cho value)

from typing import Optional, Union, TypeVar, Generic, Protocol
from collections.abc import Sequence

# Optional[X] tương đương Union[X, None]
def find_user(user_id: int) -> Optional[User]:
    return db.get(user_id)  # Có thể None

# Union — nhiều types
def parse_id(value: Union[str, int]) -> int:
    return int(value)

# TypeVar — generic type variable
T = TypeVar('T')
def first(lst: Sequence[T]) -> Optional[T]:
    return lst[0] if lst else None

# Generic class
class Stack(Generic[T]):
    def __init__(self): self._items: list[T] = []
    def push(self, item: T) -> None: self._items.append(item)
    def pop(self) -> T: return self._items.pop()

# Protocol — structural subtyping (duck typing + type safety)
class Drawable(Protocol):
    def draw(self) -> None: ...

def render(obj: Drawable) -> None:
    obj.draw()  # Bất kỳ class nào có .draw() đều hợp lệ

# Class-based Iterator
class CountDown:
    def __init__(self, start): self.current = start

    def __iter__(self): return self  # Trả về chính nó

    def __next__(self):
        if self.current <= 0:
            raise StopIteration
        self.current -= 1
        return self.current + 1

list(CountDown(3))  # [3, 2, 1]

# Generator function — ngắn gọn hơn
def countdown(n):
    while n > 0:
        yield n
        n -= 1

# Generator expression — lazy (khác list comprehension)
gen = (x**2 for x in range(1_000_000))  # Không chiếm RAM

def accumulator():
    total = 0
    while True:
        value = yield total  # Nhận value từ .send()
        if value is None: break
        total += value

acc = accumulator()
next(acc)          # Khởi động generator
acc.send(10)       # 10
acc.send(20)       # 30

# v1 → v2 migration

# 1. validator → field_validator
# v1:
@validator('email')
def validate_email(cls, v): ...

# v2:
from pydantic import field_validator
@field_validator('email')
@classmethod
def validate_email(cls, v: str) -> str: ...

# 2. orm_mode → model_config
# v1: class Config: orm_mode = True
# v2:
from pydantic import ConfigDict
class UserSchema(BaseModel):
    model_config = ConfigDict(from_attributes=True)

# 3. .dict() → .model_dump()
user.model_dump()
user.model_dump(exclude={'password'})

# 4. .json() → .model_dump_json()
user.model_dump_json()

# 5. parse_obj → model_validate
UserSchema.model_validate({"id": 1, "name": "Alice"})
UserSchema.model_validate(orm_object, from_attributes=True)

# 6. BaseSettings bị tách ra pydantic-settings package
from pydantic_settings import BaseSettings

@app.get("/users/{id}")
async def get_user(id: int, db: AsyncSession = Depends(get_db)):
    user = await db.get(User, id)  # Async I/O
    return user

@app.get("/compute")
def cpu_task(data: list):
    return [x**2 for x in data]  # CPU-bound — sync OK

@app.get("/wrong")
async def bad_endpoint():
    time.sleep(1)           # BLOCKING! Sai!
    requests.get("...")     # BLOCKING! Sai!
    # Dùng asyncio.sleep() và httpx.AsyncClient() thay

import pytest

@pytest.mark.parametrize("email,is_valid", [
    ("user@example.com", True),
    ("invalid-email", False),
    ("@nodomain.com", False),
    ("user@.com", False),
    ("", False),
])
def test_validate_email(email: str, is_valid: bool):
    assert validate_email(email) == is_valid

# Multiple params
@pytest.mark.parametrize("a,b,expected", [
    (1, 2, 3),
    (0, 0, 0),
    (-1, 1, 0),
    (100, 200, 300),
])
def test_add(a, b, expected):
    assert add(a, b) == expected

# Nested parametrize — cartesian product
@pytest.mark.parametrize("base", [2, 10])
@pytest.mark.parametrize("exponent", [0, 1, 2])
def test_power(base, exponent):
    assert power(base, exponent) == base ** exponent
# 6 test cases: (2,0), (2,1), (2,2), (10,0), (10,1), (10,2)

# Với pytest.param cho custom IDs hoặc marks
@pytest.mark.parametrize("value", [
    pytest.param(None, id="null-value"),
    pytest.param([], marks=pytest.mark.xfail, id="empty-list"),
])
def test_process(value): ...

from unittest.mock import Mock, MagicMock, AsyncMock, patch

# Mock — basic mock object
mock = Mock()
mock.method()           # OK, trả về Mock
mock.method.assert_called_once()

# MagicMock — Mock + magic methods (__len__, __iter__, __enter__ etc.)
magic = MagicMock()
len(magic)              # OK, gọi __len__
with magic:             # OK, gọi __enter__/__exit__
    pass
list(magic)             # OK, gọi __iter__

# AsyncMock (Python 3.8+) — cho async functions
async_mock = AsyncMock(return_value={"data": []})
# await async_mock()  # OK

# Patch context manager
with patch('mymodule.requests.get') as mock_get:
    mock_get.return_value.json.return_value = {'id': 1}
    result = fetch_user(1)
    mock_get.assert_called_once_with('https://api.com/users/1')

# Patch cho async
with patch('mymodule.send_email', new_callable=AsyncMock) as mock_email:
    mock_email.return_value = True
    await register_user("test@test.com")
    mock_email.assert_awaited_once()

# side_effect — simulate exceptions hay dynamic return values
mock.method.side_effect = ValueError("DB connection failed")
mock.method.side_effect = [1, 2, 3]  # Trả về lần lượt

from dataclasses import dataclass
import sys

@dataclass
class UserNormal:
    id: int; name: str; email: str

@dataclass(slots=True)  # Python 3.10+
class UserSlots:
    id: int; name: str; email: str

print(sys.getsizeof(UserNormal(1, 'Alice', 'a@b.com')))  # ~256 bytes (có __dict__)
print(sys.getsizeof(UserSlots(1, 'Alice', 'a@b.com')))   # ~56 bytes (không có __dict__)

# frozen=True + slots=True — immutable và memory-efficient
@dataclass(frozen=True, slots=True)
class Point:
    x: float
    y: float

p = Point(1.0, 2.0)
# p.x = 3.0  # FrozenInstanceError