from typing import Literal

def move(direction: Literal["up", "down", "left", "right"]) -> None:
    ...

move("up")      # OK
move("forward") # 类型检查器会报错

from pydantic import BaseModel, Field

class User(BaseModel):
    name: str = Field(
        ...,
        description="用户的真实姓名",
        min_length=1,
        max_length=50,
    )
    age: int = Field(
        default=18,
        ge=0,
        le=150,
        description="用户年龄",
    )

user = User(name="Alice", age=25)
print(user.model_dump())  # {'name': 'Alice', 'age': 25}

author = "Alice"
to = "Bob"
result = f"你好，{author}！这封邮件是发送给 {to} 的。"

lista = list(range(1, 7, 2))          # [1, 3, 5]
listb = sorted(lista, reverse=True)   # [5, 3, 1]
listb[-1]                             # 1，倒数第一个
listb[0:1]                            # [5]，切片左闭右开

a = {"x": 1, "y": 2}
b = {"y": 3, "z": 4}
c = {**a, **b}  # {'x': 1, 'y': 3, 'z': 4}
del c["x"]      # 删除键

def sum_all(*numbers: int) -> int:
    return sum(numbers)

def build_user(**kwargs: str) -> dict:
    return kwargs

sum_all(1, 2, 3)              # 6
build_user(name="Alice", age="25")  # {'name': 'Alice', 'age': '25'}

# ppp.py 中定义了 def abc(): ...
from ppp import abc

class Demo:
    def __init__(self, x):
        self.x = x

    def __init__(self, x, y):  # 上面的定义被覆盖
        self.x = x
        self.y = y

import json

with open("data.json", "r", encoding="utf-8") as f:
    data = json.load(f)

with open("out.json", "w", encoding="utf-8") as f:
    json.dump(data, f, indent=2, ensure_ascii=False)

if __name__ == "__main__":
    unittest.main()

import matplotlib.pyplot as plt

ss = [1, 4, 9, 16, 25, 36, 49, 64, 81, 100, 121, 144]
fig, ax = plt.subplots()
ax.plot(ss)
ax.set_title("Square Numbers")
plt.show()

import csv

with open("data.csv", newline="", encoding="utf-8") as f:
    reader = csv.DictReader(f)
    for row in reader:
        print(row["name"], row["age"])

python -m venv ll_env
source ll_env/bin/activate  # Linux/macOS
ll_env\Scripts\activate     # Windows

import sys
print(sys.version_info)

b = b"hello"        # bytes
s = "hello"         # str

# 显式编解码
b.decode("utf-8")   # -> str
s.encode("utf-8")   # -> bytes

name = "Alice"
age = 30

# 不推荐
print("Name: %s, Age: %d" % (name, age))
print("Name: {}, Age: {}".format(name, age))

# 推荐
print(f"Name: {name}, Age: {age}")
print(f"Name: {name!r}, Age: {age:.2f}")  # !r 用 repr，:.2f 格式化数字

# 不推荐
result = (x + y) * 2 if x > 0 and y > 0 else (x - y) * 2

# 推荐
def compute_adjustment(x: int, y: int) -> int:
    if x > 0 and y > 0:
        return (x + y) * 2
    return (x - y) * 2

result = compute_adjustment(x, y)

pair = ("Alice", 30)

# 不推荐
name = pair[0]
age = pair[1]

# 推荐
name, age = pair

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

# 不推荐
for i in range(len(fruits)):
    print(i, fruits[i])

# 推荐
for i, fruit in enumerate(fruits):
    print(i, fruit)

# 还可以指定起始索引
for i, fruit in enumerate(fruits, start=1):
    print(i, fruit)

names = ["Alice", "Bob", "Charlie"]
ages = [25, 30, 35]

for name, age in zip(names, ages):
    print(f"{name}: {age}")

# 容易误读
for x in items:
    if x < 0:
        break
else:
    print("all positive")

# 更清晰的写法
found_negative = False
for x in items:
    if x < 0:
        found_negative = True
        break
if not found_negative:
    print("all positive")

# 不推荐
count = get_count()
if count > 0:
    process(count)

# 推荐
if (count := get_count()) > 0:
    process(count)

a = [1, 2, 3, 4, 5]

a[:3]    # [1, 2, 3]
a[2:]    # [3, 4, 5]
a[::2]   # [1, 3, 5]
a[::-1]  # [5, 4, 3, 2, 1]
a[1:100] # [2, 3, 4, 5]，不报错

# 不推荐：stride 和 start/end 同时出现，可读性差
x = a[1:-1:2]

# 推荐：拆成两步
b = a[::2]
x = b[1:-1]

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

a, *b = [1]  # a=1, b=[]

students = [
    {"name": "Alice", "age": 20},
    {"name": "Bob", "age": 18},
    {"name": "Charlie", "age": 20},
]

# age 降序，name 升序
sorted_students = sorted(students, key=lambda x: (-x["age"], x["name"]))

from collections import OrderedDict

# 当顺序是业务逻辑的一部分时，显式表达
ordered = OrderedDict([("a", 1), ("b", 2)])

# 不推荐
counts = {}
if "apple" in counts:
    counts["apple"] += 1
else:
    counts["apple"] = 1

# 推荐
counts["apple"] = counts.get("apple", 0) + 1

from collections import defaultdict

# 不推荐
visits = {}
visits.setdefault("China", set()).add("Beijing")

# 推荐
visits = defaultdict(set)
visits["China"].add("Beijing")

class MyDict(dict):
    def __missing__(self, key):
        value = self[key] = f"computed-{key}"
        return value

d = MyDict()
print(d["foo"])  # computed-foo
print(d)         # {'foo': 'computed-foo'}

from typing import NamedTuple

# 不推荐
def get_user():
    return "Alice", 25, "Beijing"

name, age, city = get_user()  # 顺序一旦变化就崩

# 推荐
class User(NamedTuple):
    name: str
    age: int
    city: str

def get_user() -> User:
    return User("Alice", 25, "Beijing")

u = get_user()
print(u.name, u.age)

# 不推荐：None 容易与 0、False、空字符串混淆
def divide(a, b):
    if b == 0:
        return None
    return a / b

# 推荐
def divide(a, b):
    if b == 0:
        raise ValueError("b cannot be zero")
    return a / b

def make_counter():
    count = 0

    def counter():
        nonlocal count
        count += 1
        return count

    return counter

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

def log(message, *args):
    print(message.format(*args))

log("Hello, {}", "Alice")
log("{} + {} = {}", 1, 2, 3)

def connect(host, port=3306, timeout=30):
    ...

# 调用意图清晰
connect("localhost", timeout=10)

from datetime import datetime

# 错误：默认值只在定义时计算一次
def log_event(msg, when=datetime.now()):
    ...

# 正确
def log_event(msg, when=None):
    """Log a message with optional timestamp.

    Args:
        msg: Message to log.
        when: Timestamp; defaults to current time.
    """
    if when is None:
        when = datetime.now()
    print(f"{when}: {msg}")

def greet(name, /, greeting="Hello", *, excited=False):
    """
    name: 只能位置传参
    greeting: 位置或关键字均可
    excited: 只能关键字传参
    """
    tail = "!" if excited else "."
    print(f"{greeting}, {name}{tail}")

greet("Alice")                      # OK
greet("Alice", "Hi")                # OK
greet("Alice", excited=True)        # OK
# greet(name="Alice")               # 错误：name 是位置-only

from functools import wraps

def my_decorator(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        print("before")
        return func(*args, **kwargs)
    return wrapper

@my_decorator
def add(a, b):
    """Return sum of a and b."""
    return a + b

print(add.__name__)   # add
print(add.__doc__)    # Return sum of a and b.

a = [1, 2, 3, 4, 5]

# 不推荐
squares = list(map(lambda x: x**2, filter(lambda x: x % 2 == 0, a)))

# 推荐
squares = [x**2 for x in a if x % 2 == 0]

# 不推荐：难以理解
matrix = [[...]]
flat = [x for row in matrix for cell in row for x in cell if x > 0]

# 推荐
flat = []
for row in matrix:
    for cell in row:
        for x in cell:
            if x > 0:
                flat.append(x)

def get_batches(count, size):
    return count // size

stocks = {"apple": 10, "banana": 3}
order = ["apple", "banana", "orange"]

found = {
    name: batches
    for name in order
    if (batches := get_batches(stocks.get(name, 0), 4))
}
# {'apple': 2}

def read_large_file(path):
    with open(path, "r") as f:
        for line in f:
            yield line.strip()

# 内存友好，每次只处理一行
for line in read_large_file("huge.log"):
    process(line)

def normalize(numbers):
    total = sum(numbers)
    for n in numbers:  # 如果 numbers 是迭代器，这里已经空了
        yield 100 * n / total

it = iter([1, 2, 3])
print(list(normalize(it)))  # []

def normalize(numbers):
    numbers = list(numbers)  # 显式物化
    total = sum(numbers)
    for n in numbers:
        yield 100 * n / total

# 列表推导：一次性占用大量内存
lengths = [len(x) for x in open("file.txt")]

# 生成器表达式：惰性求值
lengths = (len(x) for x in open("file.txt"))
print(next(lengths))

def gen_a():
    yield 1
    yield 2

def gen_b():
    yield 3
    yield 4

def gen_all():
    yield from gen_a()
    yield from gen_b()

print(list(gen_all()))  # [1, 2, 3, 4]

from itertools import chain, cycle, islice, repeat, tee

list(chain([1, 2], [3, 4]))          # [1, 2, 3, 4]
list(repeat("x", 3))                 # ['x', 'x', 'x']
list(islice(cycle([1, 2]), 5))       # [1, 2, 1, 2, 1]

it = iter([1, 2, 3])
a, b = tee(it, 2)

# 不推荐
def process(data: dict):
    name = data["name"]
    address = data["address"]["city"]

# 推荐
from dataclasses import dataclass

@dataclass
class Address:
    city: str

@dataclass
class Person:
    name: str
    address: Address

def process(person: Person):
    print(person.address.city)

names = ["Alice", "Bob", "Charlie"]
names.sort(key=len)  # 传入函数即可

class InputData:
    def read(self):
        raise NotImplementedError

class PathInputData(InputData):
    def __init__(self, path):
        self.path = path

    def read(self):
        with open(self.path) as f:
            return f.read()

class Worker:
    def __init__(self, input_data):
        self.input_data = input_data

    @classmethod
    def create_workers(cls, input_class, config):
        workers = []
        for item in config:
            workers.append(cls(input_class(item)))
        return workers

class Base:
    def __init__(self, value):
        self.value = value

class Derived(Base):
    def __init__(self, value, extra):
        super().__init__(value)
        self.extra = extra

class JSONSerializableMixin:
    def to_json(self):
        import json
        return json.dumps(self.__dict__)

class User(JSONSerializableMixin):
    def __init__(self, name):
        self.name = name

u = User("Alice")
print(u.to_json())  # {"name": "Alice"}

class Demo:
    def __init__(self):
        self._internal = 1
        self.__mangled = 2

d = Demo()
print(d._internal)       # 可以访问
print(d._Demo__mangled)  # 也可以访问

from collections.abc import Sequence

class MyList(Sequence):
    def __init__(self, items):
        self._items = list(items)

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

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

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

    @property
    def celsius(self):
        return self._celsius

    @celsius.setter
    def celsius(self, value):
        if value < -273.15:
            raise ValueError("Too cold")
        self._celsius = value

class Validated:
    def __set_name__(self, owner, name):
        self.name = name
        self.private_name = f"_{name}"

    def __get__(self, obj, objtype=None):
        if obj is None:
            return self
        return getattr(obj, self.private_name, None)

    def __set__(self, obj, value):
        self.validate(value)
        setattr(obj, self.private_name, value)

    def validate(self, value):
        raise NotImplementedError

class NonNegative(Validated):
    def validate(self, value):
        if value < 0:
            raise ValueError(f"{self.name} must be >= 0")

class Person:
    age = NonNegative()

p = Person()
p.age = 30
# p.age = -1  # ValueError

class LazyRecord:
    def __init__(self):
        self.exists = 5

    def __getattr__(self, name):
        value = f"Value for {name}"
        setattr(self, name, value)
        return value

data = LazyRecord()
print(data.exists)    # 5，普通属性
print(data.foo)       # Value for foo，触发 __getattr__
print(data.foo)       # Value for foo，已缓存

class Base:
    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)
        if not hasattr(cls, "required_method"):
            raise TypeError(f"{cls.__name__} must define required_method")

class Good(Base):
    def required_method(self):
        pass

# class Bad(Base): pass  # TypeError

class PluginBase:
    plugins = []

    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)
        cls.plugins.append(cls)

class PluginA(PluginBase):
    pass

class PluginB(PluginBase):
    pass

print(PluginBase.plugins)  # [<class PluginA>, <class PluginB>]

def add_repr(cls):
    def __repr__(self):
        attrs = ", ".join(f"{k}={v!r}" for k, v in self.__dict__.items())
        return f"{cls.__name__}({attrs})"
    cls.__repr__ = __repr__
    return cls

@add_repr
class Person:
    def __init__(self, name):
        self.name = name

p = Person("Alice")
print(p)  # Person(name='Alice')

import subprocess

# 同步等待
result = subprocess.run(["ls", "-l"], capture_output=True, text=True)
print(result.stdout)

# 异步启动
proc = subprocess.Popen(["sleep", "5"])
print("doing other things...")
proc.communicate()  # 等待完成

import threading

# 这个例子不会因为多线程而加速
def count(n):
    while n > 0:
        n -= 1

threads = [threading.Thread(target=count, args=(100_000_000,)) for _ in range(2)]
for t in threads:
    t.start()
for t in threads:
    t.join()

from threading import Lock, Thread

counter = 0
lock = Lock()

def increment():
    global counter
    for _ in range(100_000):
        with lock:
            counter += 1

threads = [Thread(target=increment) for _ in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print(counter)  # 1_000_000

from queue import Queue
from threading import Thread

q = Queue(maxsize=1)  # 容量为 1，形成背压

def producer():
    for i in range(5):
        q.put(i)
        print(f"produced {i}")
    q.put(None)  # 结束信号

def consumer():
    while True:
        item = q.get()
        if item is None:
            break
        print(f"consumed {item}")

Thread(target=producer).start()
Thread(target=consumer).start()

from concurrent.futures import ThreadPoolExecutor, as_completed

def fetch(url):
    import urllib.request
    with urllib.request.urlopen(url, timeout=5) as resp:
        return len(resp.read())

urls = ["https://example.com"] * 5
with ThreadPoolExecutor(max_workers=4) as executor:
    futures = {executor.submit(fetch, url): url for url in urls}
    for future in as_completed(futures):
        print(future.result())

import asyncio

async def say_after(delay, what):
    await asyncio.sleep(delay)
    print(what)

async def main():
    await asyncio.gather(
        say_after(1, "hello"),
        say_after(2, "world"),
    )

asyncio.run(main())

import asyncio
from concurrent.futures import ThreadPoolExecutor

def blocking_io():
    import time
    time.sleep(1)
    return "done"

async def main():
    loop = asyncio.get_running_loop()
    with ThreadPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, blocking_io)
        print(result)

asyncio.run(main())

from concurrent.futures import ProcessPoolExecutor

def fib(n):
    if n < 2:
        return n
    return fib(n - 1) + fib(n - 2)

with ProcessPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(fib, [30, 31, 32, 33]))
    print(results)

def read_file(path):
    f = open(path, "r")
    try:
        data = f.read()
    except UnicodeDecodeError as e:
        print(f"decode error: {e}")
        raise
    else:
        print(f"read {len(data)} bytes")
        return data
    finally:
        f.close()

from contextlib import contextmanager

@contextmanager
def managed_resource(name):
    print(f"acquiring {name}")
    yield name
    print(f"releasing {name}")

with managed_resource("db") as r:
    print(f"using {r}")

from datetime import datetime, timezone

# 不推荐：依赖系统时区的 naive datetime
now = datetime.now()

# 推荐：显式带时区
utc_now = datetime.now(timezone.utc)
print(utc_now)

import copyreg
import pickle

class GameState:
    def __init__(self, level, health):
        self.level = level
        self.health = health

def pickle_game_state(state):
    kwargs = {"level": state.level, "health": state.health}
    return unpickle_game_state, (kwargs,)

def unpickle_game_state(kwargs):
    return GameState(**kwargs)

copyreg.pickle(GameState, pickle_game_state)

state = GameState(1, 100)
data = pickle.dumps(state)
restored = pickle.loads(data)

from decimal import Decimal, ROUND_HALF_UP

# 浮点数的精度问题
print(0.1 + 0.2)  # 0.30000000000000004

# Decimal 精确控制
price = Decimal("0.1") + Decimal("0.2")
print(price)  # 0.3

# 控制舍入
fee = Decimal("10.005").quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
print(fee)  # 10.01

python -m cProfile -s cumtime my_script.py

from collections import deque

d = deque()
d.append(1)
d.appendleft(0)
d.pop()       # 1
d.popleft()   # 0

import bisect

values = [1, 3, 5, 7, 9]
index = bisect.bisect_left(values, 4)
print(index)  # 2，4 应该插入的位置

import heapq
from functools import total_ordering

@total_ordering
class Task:
    def __init__(self, priority, name):
        self.priority = priority
        self.name = name

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

    def __eq__(self, other):
        return self.priority == other.priority

heap = []
heapq.heappush(heap, Task(2, "write tests"))
heapq.heappush(heap, Task(1, "fix bug"))

while heap:
    task = heapq.heappop(heap)
    print(task.priority, task.name)

data = bytearray(b"hello world")
mv = memoryview(data)
sub = mv[6:11]
sub[0] = ord("W")  # 修改会影响原始 data
print(data)  # bytearray(b'hello World')

x = {"a": 1, "b": [2, 3]}
print(repr(x))  # {'a': 1, 'b': [2, 3]}

# 自定义类也应实现 __repr__
class Point:
    def __init__(self, x, y):
        self.x = x
        self.y = y

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

p = Point(1, 2)
print(repr(p))  # Point(1, 2)

import unittest

class TestStringMethods(unittest.TestCase):
    def test_upper(self):
        self.assertEqual("foo".upper(), "FOO")

    def test_isupper(self):
        self.assertTrue("FOO".isupper())
        self.assertFalse("Foo".isupper())

if __name__ == "__main__":
    unittest.main()

class TestDatabase(unittest.TestCase):
    def setUp(self):
        self.conn = create_test_db()

    def tearDown(self):
        self.conn.close()
        cleanup_test_db()

    def test_insert(self):
        self.conn.execute("INSERT INTO users VALUES ('Alice')")
        row = self.conn.execute("SELECT * FROM users").fetchone()
        self.assertEqual(row[0], "Alice")

from unittest.mock import Mock, patch

def get_user_name(user_id):
    import requests
    resp = requests.get(f"/users/{user_id}")
    return resp.json()["name"]

with patch("requests.get") as mock_get:
    mock_get.return_value.json.return_value = {"name": "Alice"}
    assert get_user_name(1) == "Alice"

from abc import ABC, abstractmethod

class UserRepository(ABC):
    @abstractmethod
    def get(self, user_id):
        ...

class HttpUserRepository(UserRepository):
    def get(self, user_id):
        ...  # 真实 HTTP 调用

class FakeUserRepository(UserRepository):
    def __init__(self, users):
        self.users = users

    def get(self, user_id):
        return self.users.get(user_id)

# 测试时使用 FakeUserRepository

def buggy_func(x):
    import pdb; pdb.set_trace()
    return x / 0

buggy_func(10)

import tracemalloc

tracemalloc.start()

# 运行你的代码
large = [bytearray(1024 * 1024) for _ in range(100)]

snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics("lineno")[:10]
for stat in top_stats:
    print(stat)

# 生成依赖
pip freeze > requirements.txt

# 重建环境
python -m venv venv
source venv/bin/activate
pip install -r requirements.txt

def calculate_area(radius):
    """Calculate the area of a circle.

    Args:
        radius: The radius of the circle.

    Returns:
        The area as a float.

    Raises:
        ValueError: If radius is negative.
    """
    if radius < 0:
        raise ValueError("radius must be non-negative")
    return 3.14159 * radius ** 2

myproject/
    __init__.py
    core.py
    utils.py

# __init__.py
from .core import process
from .utils import helper

__all__ = ["process", "helper"]

import os

ENV = os.getenv("APP_ENV", "development")

if ENV == "production":
    DATABASE_URL = os.getenv("DATABASE_URL")
else:
    DATABASE_URL = "sqlite:///dev.db"

class MyLibError(Exception):
    """Base exception for mylib."""

class ConfigError(MyLibError):
    """Raised when configuration is invalid."""

# 调用方可以统一捕获 MyLibError
try:
    do_something()
except MyLibError as e:
    logger.error(f"mylib error: {e}")

# 不推荐：A -> B -> A

# 推荐：A -> common <- B

def do_something():
    from module_b import helper
    helper()

import warnings

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

from typing import Optional, List

def find_user(user_id: int) -> Optional[dict]:
    ...

def list_users(ids: List[int]) -> List[dict]:
    ...