1
from pprint import pprint
2

3
class User:
4
  id = 0
5
  def __init__(self, name, age):
6
    self.name = name
7
    self.age = age
8
    User.id += 1
9
    self.id = User.id
10
    self._info = f'{self.name}, {self.age}, {self.id}'
11

12
  @classmethod
13
  def get_name(cls): # cls 是实际实例化对象对应的类, 例如 Admin 继承了 User, Admin.get_name() 这里的 cls 就是 Admin
14
    return 'User'
15

16
  @staticmethod
17
  def print():
18
    print('Hello :)', User.id)
19
  def say_hello(self):
20
    print(f'hello, {self.name}')
21

22
  # getter and setter
23
  @property
24
  def info(self) -> str:
25
    return self._info
26
  @info.setter
27
  def info(self, value: str) -> None:
28
    self._info = value
29

30
user = User('kuidi', 18)
31
user2 = User('kuidi', 18)
32
user.say_hello()
33
User.print()
34

35
print(user.info)
36
pprint(vars(user)) # 打印对象的所有属性

1
point = (100, 200)
2

3
x, y = point # 解构
4
point[0] = 150 # ❌ 不可修改

1
from enum import Enum, auto
2

3
class Status(Enum):
4
  PENDING = 1
5
  RUNNING = 2
6
  SUCCESS = 3
7
  FAILED = 4
8
  # 默认情况下, 枚举允许有相同的值, 可通过增加 @unique 装饰器强制保持唯一
9
  # CANCELLED = 4
10
  # auto() 会自动分配下一个可用的整数
11
  CANCELLED = auto()

1
class Status(IntEnum):
2
  PENDING = 1
3
  RUNNING = 2
4
  SUCCESS = 3
5

6
print(Status.PENDING == 1) # True
7

8
# 其他 API
9
print(Status.PENDING.name) # PENDING
10
print(Status.PENDING.value) # 1
11

12
request_params = {
13
  'status': Status.PENDING.name,
14
  'code': 1,
15
}
16

17
print(Status(request_params['code'])) # 通过值获取枚举
18
print(Status[request_params['status']]) # 通过键获取枚举

1
def handle_event(event):
2
  match event:
3
    # 1. 匹配固定的字符串
4
    case "quit":
5
      print("Action: Quit")
6
    # 2. 匹配列表/元组，并解构其中的值
7
    case ["click", x, y]:
8
      print(f"Mouse clicked at x={x}, y={y}")
9
    # 3. 匹配字典（对象），并提取其中的键值
10
    case {"type": "keypress", "key": key_name}:
11
      print(f"Key pressed: {key_name}")
12
    # 4. 兜底匹配（相当于 switch 的 default）
13
    case _:
14
      print("Unknown event")

1
try:
2
  # 可能会抛出异常的代码
3
except ValueError as e:
4
  # ...
5
except Exception as e:
6
  # 处理异常
7
  print(e)
8
  # 继续执行后续代码
9
finally:
10
  # 无论是否抛出异常，都会执行
11
  pass

1
# 使用 * 解包 list / tuple
2
user_info = ['xxx', 18, 'male']
3
print(*user_info)
4

5
# 使用 ** 解包 dict
6
user_info = {'name': 'xxx', 'age': 18, 'gender': 'male'}
7
print(**user_info)

1
def make_counter():
2
  count = 0
3
  def counter():
4
    nonlocal count
5
    count += 1
6
    return count
7
  return counter
8

9
my_counter = make_counter()
10
print(my_counter()) # > 1
11
print(my_counter()) # > 2
12
print(my_counter()) # > 3

1
double = lambda x: x * 2

1
from json import dumps, loads
2

3
json_str = '{"name": "倪碟", "age": 18}'
4

5
user = loads(json_str)
6
print(user) # 转换为字典 > {'name': '倪碟', 'age': 18}
7
# 转为字符串, 保证中文不会被转义, 使用 2 个空格缩进
8
user_str = dumps(user, ensure_ascii=False, indent=2)
9
print(user_str)
10
'''
11
{
12
  "name": "倪碟",
13
  "age": 18
14
}
15
'''

1
import json
2

3
# 1. 直接将 Python 对象写入 json 文件
4
config = {"server": "127.0.0.1", "port": 8000, "debug": True}
5

6
with open("config.json", "w", encoding="utf-8") as f:
7
    json.dump(config, f, indent=4, ensure_ascii=False)
8

9
# 2. 直接从 json 文件读取 Python 对象
10
with open("config.json", "r", encoding="utf-8") as f:
11
    loaded_config = json.load(f)
12

13
print(loaded_config["port"])  # 8000

1
import functools
2

3
def logger(func):
4
  @functools.wraps(func)
5
  def wrapper(*args, **kwargs):
6
    print('start')
7
    func(*args, **kwargs)
8
    print('end')
9
  return wrapper
10

11
@logger
12
def add(a, b):
13
  return a + b
14

15
add(1, 2)

1
def logger(func):
2
  def wrapper(*args, **kwargs):
3
    print('start')
4
    func(*args, **kwargs)
5
    print('end')
6
  return wrapper
7

8
def add(a, b):
9
  return a + b
10

11
add_fn = logger(add)
12
add_fn(1, 2)

1
import logging
2
import sys
3
from logging.handlers import RotatingFileHandler
4

5
def setup_logger(name: str = "app_logger", log_file: str = "app.log") -> logging.Logger:
6
    """配置并返回一个高级 Logger"""
7

8
    # 1. 创建 Logger 对象
9
    logger = logging.getLogger(name)
10
    logger.setLevel(logging.DEBUG)  # 允许捕获最低 DEBUG 级别的日志
11

12
    # 防止日志重复打印（如果该 logger 继承了父 logger 的 handler）
13
    logger.propagate = False
14

15
    # 2. 定义 Formatter（格式化器）
16
    formatter = logging.Formatter(
17
        fmt='[%(asctime)s] [%(levelname)s] [%(name)s] (%(filename)s:%(lineno)d) - %(message)s',
18
        datefmt='%Y-%m-%d %H:%M:%S'
19
    )
20

21
    # 3. 创建 Handler 1: 控制台处理器（只打印 INFO 及以上级别）
22
    console_handler = logging.StreamHandler(sys.stdout)
23
    console_handler.setLevel(logging.INFO)
24
    console_handler.setFormatter(formatter)
25

26
    # 4. 创建 Handler 2: 文件处理器（滚动记录所有 DEBUG 及以上级别，避免单文件过大）
27
    # maxBytes=5*1024*1024 (5MB), backupCount=5 (保留5个备份文件)
28
    file_handler = RotatingFileHandler(
29
        log_file, maxBytes=5_242_880, backupCount=5, encoding="utf-8"
30
    )
31
    file_handler.setLevel(logging.DEBUG)
32
    file_handler.setFormatter(formatter)
33

34
    # 5. 将 Handler 绑定到 Logger 上
35
    logger.addHandler(console_handler)
36
    logger.addHandler(file_handler)
37

38
    return logger
39

40
# --- 测试使用 ---
41
if __name__ == "__main__":
42
    log = setup_logger("MainModule")
43

44
    log.debug("这条信息只会进入文件，不会出现在控制台。")
45
    log.info("这条信息既在控制台显示，也会写入文件。")
46

47
    try:
48
        1 / 0
49
    except ZeroDivisionError as e:
50
        # exc_info=True 会自动把完整的错误堆栈（Traceback）记录进日志
51
        log.error("程序运行抛出异常！", exc_info=True)

1
from concurrent.futures import ThreadPoolExecutor
2

3
import requests
4

5
urls = [
6
  'https://httpbingo.org/delay/1',
7
  'https://httpbingo.org/delay/2',
8
  'https://httpbingo.org/delay/3',
9
]
10

11
def fetch_url(url):
12
  print(f'fetch url: {url}')
13
  response = requests.get(url)
14
  return f'{url} status code: {response.status_code}'
15

16
with ThreadPoolExecutor(max_workers=3) as executor:
17
  results = executor.map(fetch_url, urls) # 类似于 Promise.all()
18
  for result in results:
19
    print(result)

1
import time
2
from concurrent.futures import ProcessPoolExecutor
3

4
# 模拟一个耗时的 CPU 计算任务（比如图像像素矩阵处理）
5
def cpu_bound_task(n):
6
    print(f"进程开始计算: {n}")
7
    count = 0
8
    for i in range(n):
9
        count += i
10
    return count
11

12
if __name__ == "__main__":  # Windows/MacOS 下多进程必须加这句，防止无限死循环创建进程
13
    numbers = [50000000, 60000000, 70000000]
14

15
    start_time = time.time()
16

17
    # ProcessPoolExecutor 会自动根据 CPU 核心数创建多个独立进程
18
    with ProcessPoolExecutor() as executor:
19
        results = executor.map(cpu_bound_task, numbers)
20

21
        for result in results:
22
            print(f"计算结果: {result}")
23

24
    print(f"总耗时: {time.time() - start_time:.2f} 秒")

1
import asyncio
2
import httpx
3

4
async def fetch_api(item_id):
5
  print(f'协程启动, id: {item_id}')
6
  async with httpx.AsyncClient() as client:
7
    response = await client.get(f'https://httpbingo.org/delay/{item_id}')
8
    print(response.status_code)
9
    print(response.text)
10
    return response.text
11

12
async def main():
13
  task = [fetch_api(item_id) for item_id in range(1, 4)]
14
  results = await asyncio.gather(*task) # 解包
15
  print(results)
16

17
asyncio.run(main())

1
import asyncio
2
import httpx
3

4
async def fetch_data():
5
  async with httpx.AsyncClient() as client:
6
    response = await client.get('https://httpbingo.org/delay/1')
7
    print(response.status_code)
8
    print(response.text)
9

10
asyncio.run(fetch_data())

1
import sqlite3
2

3
# 1. 连接到数据库（如果文件不存在，会自动创建）
4
# 也可以使用 ':memory:' 创建内存数据库，程序关闭后数据丢失
5
conn = sqlite3.connect('my_database.db')
6

7
# 2. 创建一个游标对象（通过游标执行 SQL 语句）
8
cursor = conn.cursor()
9

10
# 3. 执行 SQL 语句创建表
11
cursor.execute('''
12
    CREATE TABLE IF NOT EXISTS users (
13
        id INTEGER PRIMARY KEY AUTOINCREMENT,
14
        name TEXT NOT NULL,
15
        age INTEGER,
16
        email TEXT UNIQUE
17
    )
18
''')
19

20
# 4. 提交更改（对数据库结构或数据的增删改必须 commit）
21
conn.commit()
22

23
# 5. 关闭游标和连接
24
cursor.close()
25
conn.close()

1
from datetime import datetime, date, timedelta
2
from time import time
3

4
print(date.today()) # 当前日期 > 2026-06-11
5
print(datetime.now()) # 当前时间 > 2026-06-11 14:47:52.840429
6

7
# 日期格式化
8
print(datetime.strptime('2022-01-01 12:12:12', '%Y-%m-%d %H:%M:%S')) # 2022-01-01 12:12:12
9
print(date.today().strftime('%Y-%m-%d')) # 2026-06-11
10

11
# 时间戳转换
12
ts = time()
13
print(ts) # 时间戳(float s) > 1723422473.278943
14
print(int(ts)) # 时间戳(int s) > 1723422473
15
print(int(ts * 1000)) # 时间戳(int ms) > 1723422473233
16

17
# 时间运算
18
now = datetime.now()
19
print('三天后: ', now + timedelta(days=3))

1
with open('file.txt', 'r') as f:
2
  content = f.read()
3
  print(content)

1
from pathlib import Path
2

3
file = Path('test.txt')
4

5
content = file.read_text()
6
print(content)

1
from pathlib import Path
2

3
file = Path('test.txt')
4

5
print(file.exists()) # 文件是否存在