SQLModel
一、安装 SQLModel
1. 基础安装
pip install sqlmodel
- 依赖说明
- **Python 3.7+**：需确保 Python 版本符合要求。
- 数据库驱动：根据使用的数据库类型安装对应驱动：
- SQLite：内置无需额外安装。
- PostgreSQL：
pip install asyncpg - MySQL：
pip install pymysql
二、定义数据模型
1. 基础模型定义
from sqlmodel import SQLModel, Field, create_engine
from typing import Optional
class User(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True) # 主键
username: str = Field(index=True, unique=True) # 唯一索引
email: str = Field(index=True) # 普通索引
age: Optional[int] = None # 可选字段
created_at: datetime = Field(default_factory=datetime.utcnow) # 默认值
- 关键参数：
table=True：标记该类为数据库表。Field：定义字段约束（如唯一性、索引、默认值等）。
三、连接数据库与创建表
1. 创建数据库引擎
# SQLite 示例
engine = create_engine("sqlite:///./test.db", echo=True) # echo=True 显示SQL日志
# PostgreSQL 示例
# engine = create_engine("postgresql+asyncpg://user:password@localhost/dbname")
2. 创建表
SQLModel.metadata.create_all(engine) # 根据模型生成数据库表
四、CRUD 操作示例
1. 创建数据（Create）
from sqlmodel import Session
with Session(engine) as session:
new_user = User(username="alice", email="alice@example.com", age=25)
session.add(new_user) # 添加记录
session.commit() # 提交事务
session.refresh(new_user) # 刷新对象获取数据库生成的数据（如自增ID）
2. 查询数据（Read）
# 查询单条记录
user = session.get(User, 1) # 通过主键查询
# 条件查询
user = session.exec(
select(User).where(User.username == "alice")
).first()
# 查询所有记录
users = session.exec(select(User)).all()
3. 更新数据（Update）
with Session(engine) as session:
user = session.get(User, 1)
if user:
user.email = "new_email@example.com"
session.commit()
4. 删除数据（Delete）
with Session(engine) as session:
user = session.get(User, 1)
if user:
session.delete(user)
session.commit()
五、高级功能示例
1. 数据库关系（一对多）
class Post(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
title: str
content: str
author_id: Optional[int] = Field(default=None, foreign_key="user.id") # 外键
author: User = Relationship(back_populates="posts") # 关联关系
class User(SQLModel, table=True):
# ... 其他字段
posts: list[Post] = Relationship(back_populates="author") # 反向关联
2. 动态查询（模糊匹配）
from sqlmodel import select
form_data = {"username": "alice", "email": None} # 假设表单输入
query = select(User)
if form_data["username"]:
query = query.where(User.username.like(f"%{form_data['username']}%"))
if form_data["email"]:
query = query.where(User.email == form_data["email"])
results = session.exec(query).all()
六、与 FastAPI 集成
1. 定义 API 路由
from fastapi import FastAPI, Depends
from sqlmodel import Session
app = FastAPI()
# 依赖注入数据库会话
def get_db():
with Session(engine) as session:
yield session
@app.post("/users/")
def create_user(user: User, db: Session = Depends(get_db)):
db.add(user)
db.commit()
return user
@app.get("/users/{user_id}")
def get_user(user_id: int, db: Session = Depends(get_db)):
return db.get(User, user_id)
七、常见问题与解决
- 数据库连接失败
- 检查数据库驱动是否安装（如 PostgreSQL 需
asyncpg）。 - 确认数据库 URL 格式正确（如
postgresql+asyncpg://user:pass@host/db）。
- 检查数据库驱动是否安装（如 PostgreSQL 需
- 字段类型不匹配
- 确保模型字段类型与数据库列类型一致（如
DateTime对应数据库的datetime类型）。
- 确保模型字段类型与数据库列类型一致（如
- 会话未提交导致数据丢失
- 操作后务必调用
session.commit()，或使用上下文管理器自动提交。
- 操作后务必调用
namedtuple
创建带字段名的元组，提升代码可读性。
一、基础用法
1. 创建与实例化
from collections import namedtuple
# 定义命名元组类
Point = namedtuple("Point", ["x", "y"])
# 或简写字段名（用空格/逗号分隔）
# Point = namedtuple("Point", "x y")
# 实例化
p = Point(3, 4)
print(p) # 输出: Point(x=3, y=4)
2. 字段访问
print(p.x) # 通过字段名访问 → 3
print(p[0]) # 通过索引访问 → 3
3. 不可变性
try:
p.x = 10 # 尝试修改字段值 → 报错
except AttributeError as e:
print(f"Error: {e}") # AttributeError: can't set attribute
二、高级功能
1. 字段名与默认值
# 定义带默认值的命名元组
Person = namedtuple("Person", ["name", "age", "email"], defaults=[None, "unknown"])
p = Person(name="Alice")
print(p) # Person(name='Alice', age=None, email='unknown')
2. 转换为字典
data = p._asdict() # → {'name': 'Alice', 'age': None, 'email': 'unknown'}
3. 从可迭代对象创建实例
data = [10, 20]
p = Point._make(data) # → Point(x=10, y=20)
4. 替换字段值
p2 = p._replace(x=100) # → Point(x=100, y=4)
三、实战场景示例
1. 处理结构化数据（如 CSV）
from collections import namedtuple
# 定义 CSV 数据结构
User = namedtuple("User", ["id", "name", "email"])
# 模拟 CSV 数据解析
csv_line = "1,alice,alice@example.com"
user = User._make(csv_line.split(","))
print(user.name) # alice
print(user._asdict()) # {'id': '1', 'name': 'alice', 'email': 'alice@example.com'}
2. 替代字典提升可读性
# 传统字典方式
user_dict = {"id": 1, "name": "Bob", "email": "bob@example.com"}
print(user_dict["name"]) # 需记忆键名
# 使用命名元组
User = namedtuple("User", user_dict.keys())
user = User(*user_dict.values())
print(user.name) # 直接通过字段名访问，代码更清晰
3. 扩展功能（继承）
from collections import namedtuple
from math import sqrt
# 基础坐标点
Point = namedtuple("Point", ["x", "y"])
# 扩展为颜色点，添加计算距离方法
class ColorPoint(Point):
def distance_to_origin(self):
return sqrt(self.x**2 + self.y**2)
p = ColorPoint(3, 4)
print(p.distance_to_origin()) # 5.0
四、与字典、类的对比
| 特性 | namedtuple |
字典（dict） |
普通类（class） |
|---|---|---|---|
| 内存占用 | 低（轻量级） | 较高（存储键哈希表） | 较高（含__dict__） |
| 不可变性 | 是 | 否 | 可自定义 |
| 字段访问 | 通过名称或索引 | 仅通过键 | 仅通过属性名 |
| 适用场景 | 固定结构的小型数据 | 动态键值对 | 复杂行为与状态管理 |
五、类型注解（Python 3.6+）
from typing import NamedTuple
class Product(NamedTuple):
name: str
price: float
stock: int
p = Product(name="iPhone", price=999.99, stock=50)
print(p.price) # 类型提示更清晰
六、应用场景总结
- 替代简单类：当仅需存储固定字段且无需方法时，比自定义类更简洁。
- 提升代码可读性：通过字段名明确数据含义，避免魔法数字/字符串。
- 函数式数据处理：与
map、filter结合，处理结构化数据流。 - API 响应解析：将 JSON 数据转换为命名元组，增强类型安全性。
__init__.py:
1：包初始化
2：管理包接口
3：包信息
抽象基类（ABC）
一、抽象基类的核心概念
- 定义
抽象基类（Abstract Base Class, ABC）通过abc模块实现，用于定义接口规范，强制子类实现特定方法。- 核心作用：
- 强制子类实现抽象方法（否则无法实例化）。
- 提供类型检查支持（如
isinstance(obj, MyABC)）。 - 定义统一的接口契约，增强代码可维护性。
- 核心作用：
- 关键组件
ABC类：所有抽象基类的基类，需通过继承使用。@abstractmethod装饰器：标记抽象方法。ABCMeta元类：用于自定义抽象基类的元逻辑。
二、基础使用示例
1. 定义抽象基类
from abc import ABC, abstractmethod
class Shape(ABC):
@abstractmethod
def area(self) -> float:
"""计算面积"""
pass
@abstractmethod
def draw(self) -> None:
"""绘制图形"""
pass
- 规则：
- 抽象基类必须继承自
ABC。 - 抽象方法需使用
@abstractmethod装饰器，且不能有具体实现。
- 抽象基类必须继承自
2. 子类实现抽象基类
class Circle(Shape):
def __init__(self, radius: float):
self.radius = radius
def area(self) -> float:
return 3.14 * self.radius ** 2
def draw(self) -> None:
print(f"Drawing a circle with radius {self.radius}")
class Rectangle(Shape):
def __init__(self, width: float, height: float):
self.width = width
self.height = height
def area(self) -> float:
return self.width * self.height
def draw(self) -> None:
print(f"Drawing a rectangle ({self.width}x{self.height})")
验证强制实现：
若子类未实现所有抽象方法，实例化时会抛出TypeError
class InvalidShape(Shape): pass # 以下代码会报错： # InvalidShape() → TypeError: Can't instantiate abstract class InvalidShape with abstract methods area, draw
三、高级功能示例
1. 抽象属性（Abstract Properties）
class Animal(ABC):
@property
@abstractmethod
def species(self) -> str:
"""物种属性"""
pass
class Dog(Animal):
@property
def species(self) -> str:
return "Canine"
# 错误示例：未实现抽象属性
class Bird(Animal):
pass # 报错：Can't instantiate abstract class Bird with abstract methods species
2. 注册虚拟子类（Virtual Subclass）
即使不直接继承，也可通过 register() 声明符合接口的类：
class Square(Shape):
def __init__(self, side: float):
self.side = side
def area(self) -> float:
return self.side ** 2
def draw(self) -> None:
print(f"Drawing a square with side {self.side}")
# 注册为虚拟子类
Shape.register(Square)
# 类型检查通过
print(isinstance(Square(5), Shape)) # → True
3. 自定义子类检查逻辑
通过重写 __subclasshook__ 方法，动态判断类是否符合接口：
class IterableContainer(ABC):
@classmethod
def __subclasshook__(cls, subclass):
return hasattr(subclass, '__iter__') and hasattr(subclass, '__len__')
# 任何实现 __iter__ 和 __len__ 的类都被视为子类
class MyList:
def __init__(self, data):
self.data = data
def __iter__(self):
return iter(self.data)
def __len__(self):
return len(self.data)
print(issubclass(MyList, IterableContainer)) # → True
四、实战场景：图形绘制系统
1. 定义接口
from abc import ABC, abstractmethod
class Drawable(ABC):
@abstractmethod
def draw(self) -> None:
"""绘制图形"""
pass
@abstractmethod
def get_color(self) -> str:
"""获取颜色"""
pass
2. 实现具体图形
class Circle(Drawable):
def __init__(self, color: str, radius: float):
self.color = color
self.radius = radius
def draw(self) -> None:
print(f"Drawing a {self.color} circle with radius {self.radius}")
def get_color(self) -> str:
return self.color
class Triangle(Drawable):
def __init__(self, color: str, sides: int):
self.color = color
self.sides = sides
def draw(self) -> None:
print(f"Drawing a {self.color} triangle with {self.sides} sides")
def get_color(self) -> str:
return self.color
3. 使用多态
shapes = [Circle("red", 5), Triangle("blue", 3)]
for shape in shapes:
shape.draw()
print(f"Color: {shape.get_color()}")
输出：
Drawing a red circle with radius 5
Color: red
Drawing a blue triangle with 3 sides
Color: blue
五、与普通类的对比
| 特性 | 抽象基类 | 普通类 |
|---|---|---|
| 实例化 | 不可直接实例化 | 可直接实例化 |
| 方法实现 | 必须包含抽象方法（无实现） | 可包含完整实现 |
| 子类约束 | 强制实现所有抽象方法 | 无强制约束 |
| 类型检查 | 支持 isinstance 检查 |
依赖显式继承关系 |
六、最佳实践
接口与实现分离
抽象基类仅定义接口，具体实现由子类完成，提升代码模块化。结合工厂模式
使用工厂方法根据抽象基类创建具体对象：def create_shape(shape_type: str, **kwargs) -> Drawable: if shape_type == "circle": return Circle(**kwargs) elif shape_type == "triangle": return Triangle(**kwargs) else: raise ValueError("Invalid shape type")文档与注释
在抽象方法中添加详细文档，明确子类实现要求：class Shape(ABC): @abstractmethod def area(self) -> float: """计算图形的面积，单位：平方米。 返回: float: 面积值 """ pass
七、常见错误与解决
- 未实现所有抽象方法
- 错误：子类遗漏抽象方法 →
TypeError。 - 解决：检查子类是否覆盖所有基类抽象方法。
- 错误：子类遗漏抽象方法 →
- 抽象方法被覆盖但未重写
- 错误：子类声明了同名方法但未正确实现 → 运行时错误。
- 解决：确保子类方法签名与抽象方法一致。
- 抽象属性未定义
- 错误：抽象属性未在子类中通过
@property实现 →TypeError。 - 解决：使用
@property和@<attr>.setter完整定义属性。
- 错误：抽象属性未在子类中通过
mock
一、基础概念
unittest.mock 提供 Mock 对象 和 Patch 装饰器，用于在测试中替换真实对象，隔离依赖项，控制方法行为（如返回值、异常抛出等），并验证调用细节。
二、核心使用场景与示例
1. 模拟外部接口调用
场景：测试依赖第三方 API 的函数，避免真实请求。
from unittest import mock
import unittest
import requests
def fetch_weather(city):
response = requests.get(f"https://api.weather.com/{city}")
return response.json()["temperature"]
class TestWeather(unittest.TestCase):
@mock.patch('requests.get') # 替换 requests.get 为 Mock 对象
def test_fetch_weather(self, mock_get):
# 设置 Mock 返回值
mock_response = mock.Mock()
mock_response.json.return_value = {"temperature": 25}
mock_get.return_value = mock_response
result = fetch_weather("beijing")
self.assertEqual(result, 25)
mock_get.assert_called_once_with("https://api.weather.com/beijing")
if __name__ == '__main__':
unittest.main()
说明：
- @mock.patch装饰器替换requests.get为 Mock 对象
- 通过
return_value设置模拟响应数据。 - 使用assert_called_once_with验证调用参数
2. 模拟类方法调用
场景：测试依赖其他类方法的函数，隔离被测单元。
from unittest import mock
import unittest
class Database:
def get_user(self, user_id):
# 实际数据库查询逻辑
pass
def get_user_name(user_id):
db = Database()
user = db.get_user(user_id)
return user.name
class TestGetName(unittest.TestCase):
def test_get_user_name(self):
mock_db = mock.Mock(spec=Database) # 创建模拟对象，限定方法范围
mock_db.get_user.return_value = mock.Mock(name="Alice") # 设置返回值
# 替换函数中的 Database 实例
with mock.patch('__main__.Database', return_value=mock_db):
result = get_user_name(1)
self.assertEqual(result, "Alice")
mock_db.get_user.assert_called_once_with(1)
if __name__ == '__main__':
unittest.main()
说明：spec=Database确保 Mock 对象仅暴露Database的真实方法
patch作为上下文管理器临时替换类
3. 模拟异常场景
场景：测试错误处理逻辑。
from unittest import mock
import unittest
def divide(a, b):
return a / b
class TestDivide(unittest.TestCase):
@mock.patch('__main__.divide', side_effect=ZeroDivisionError)
def test_divide_by_zero(self, mock_divide):
with self.assertRaises(ZeroDivisionError):
divide(10, 0)
mock_divide.assert_called_once_with(10, 0)
if __name__ == '__main__':
unittest.main()
说明：
side_effect设置为异常类或可调用对象，模拟错误触发
4. 动态返回值（动态参数依赖）
场景：根据输入参数返回不同结果。
from unittest import mock
import unittest
def get_discount(price, user_type):
# 实际逻辑根据 user_type 计算折扣
pass
class TestDiscount(unittest.TestCase):
def test_discount(self):
mock_discount = mock.Mock()
mock_discount.side_effect = lambda p, ut: 0.9 if ut == "VIP" else 0.95
with mock.patch("__main__.get_discount", mock_discount):
self.assertEqual(get_discount(100, "VIP"), 90)
self.assertEqual(get_discount(100, "guest"), 95)
if __name__ == '__main__':
unittest.main()
说明：
side_effect可以是函数，根据参数动态返回值
5. 验证调用次数与参数
场景：确保方法被正确调用。
from unittest import mock
import unittest
class Logger:
def info(self, message):
pass
def process_data(data, logger):
if data:
logger.info("Processing data")
else:
logger.info("No data")
class TestLogger(unittest.TestCase):
def test_logger_calls(self):
mock_logger = mock.Mock(spec=Logger)
process_data(None, mock_logger)
mock_logger.info.assert_called_once_with("No data")
if __name__ == '__main__':
unittest.main()
说明：
assert_called_once_with验证调用次数和参数
三、高级功能
1. Patch 多个对象
@mock.patch('module.ClassA.method1')
@mock.patch('module.ClassB.method2')
def test_multiple_patches(mock_method2, mock_method1):
# 参数顺序：后装饰的先注入
mock_method1.return_value = "Mocked A"
mock_method2.return_value = "Mocked B"
# ... 测试逻辑
2. Mock 类属性
@mock.patch('module.MyClass.static_attr', new=100)
def test_static_attr(mock_attr):
from module import MyClass
self.assertEqual(MyClass.static_attr, 100)
3. 使用 MagicMock 自动创建魔术方法
mock = mock.MagicMock()
mock.__len__.return_value = 5
self.assertEqual(len(mock), 5) # 自动支持 len()
四、最佳实践
最小化 Mock 范围：仅替换必要的依赖，避免过度 Mock。
使用
autospec=True：自动同步真实对象的签名，避免参数错误
结合上下文管理器：临时替换对象，自动恢复原状。
断言覆盖：验证调用次数、参数、副作用（如异常）。
五、总结
unittest.mock 的核心价值在于：
- 隔离测试：避免外部依赖干扰。
- 精准控制：模拟复杂行为（如网络延迟、数据库事务）。
- 高效验证：快速定位代码交互问题。
collections
一、核心容器类
1. deque（双端队列）
用途：高效实现队列（FIFO）和栈（LIFO），支持两端快速插入/删除。
示例：
from collections import deque q = deque([1, 2, 3]) q.append(4) # 队尾添加 → deque([1,2,3,4]) q.appendleft(0) # 队头添加 → deque([0,1,2,3,4]) q.popleft() # 队头删除 → 0 q.pop() # 队尾删除 → 4
2. Counter（计数器）
用途：统计可哈希对象的出现次数，支持频率分析。
示例：
from collections import Counter c = Counter("abracadabra") print(c) # Counter({'a': 5, 'b': 2, 'r': 2, 'c': 1, 'd': 1}) print(c.most_common(2))# [('a', 5), ('b', 2)]
3. defaultdict（默认字典）
用途：为字典提供默认值，避免
KeyError。示例：
from collections import defaultdict dd = defaultdict(list) dd["fruits"].append("apple") dd["fruits"].append("banana") print(dd) # defaultdict(<class 'list'>, {'fruits': ['apple', 'banana']})
4. namedtuple（命名元组）
用途：创建带字段名的元组，提升代码可读性。
示例
from collections import namedtuple Point = namedtuple("Point", ["x", "y"]) p = Point(3, 4) print(p.x, p.y) # 3 4
5. OrderedDict（有序字典）
用途：保持键值对的插入顺序（Python 3.7+ 普通字典已支持，此功能保留兼容性）。
示例
from collections import OrderedDict od = OrderedDict([("a", 1), ("b", 2)]) od.move_to_end("a") # 将键"a"移动到末尾 print(od) # OrderedDict([('b', 2), ('a', 1)])
6. ChainMap（多字典组合）
用途：将多个字典逻辑合并为一个视图，优先按顺序查找。
from collections import ChainMap dict1 = {"a": 1, "b": 2} dict2 = {"b": 3, "c": 4} cm = ChainMap(dict1, dict2) print(cm["b"]) # 2（优先从dict1查找）
二、辅助工具类
1. default_factory（默认工厂函数）
用途：为
defaultdict提供动态默认值生成逻辑。dd = defaultdict(lambda: "N/A") print(dd["key"]) # "N/A"
2. MappingProxyType（只读字典代理）
用途：创建不可变字典视图，防止意外修改。
示例
from collections import MappingProxyType original = {"a": 1} proxy = MappingProxyType(original) print(proxy["a"]) # 1 proxy["a"] = 2 # 报错：TypeError: 'mappingproxy' object does not support item assignment
3. deque.rotate(n)（队列旋转）
用途：将队列元素向右旋转
n步（正数右移，负数左移）。q = deque([1, 2, 3, 4, 5]) q.rotate(2) # 右移两位 → deque([4, 5, 1, 2, 3])
三、高级用法示例
1. deque 实现滑动窗口
from collections import deque
window = deque(maxlen=3)
data = [1, 3, -1, -3, 5, 3, 6, 7]
for num in data:
window.append(num)
print(window) # 输出滑动窗口的当前状态
2. Counter 统计文件词频
from collections import Counter
with open("text.txt") as f:
words = f.read().split()
freq = Counter(words)
print(freq.most_common(10)) # 输出出现频率最高的前10个词
3. defaultdict 构建树形结构
from collections import defaultdict
tree = lambda: defaultdict(tree)
root = tree()
root["a"]["b"]["c"] = 1
print(root) # defaultdict(<function <lambda> at 0x...>, {'a': defaultdict(<function <lambda> at 0x...>, {'b': defaultdict(<function <lambda> at 0x...>, {'c': 1})})})
四、工具类对比
| 类/函数 | 核心功能 | 典型场景 |
|---|---|---|
deque |
高效队列/栈操作 | 任务调度、广度优先搜索（BFS） |
Counter |
元素频率统计 | 文本分析、数据聚合 |
defaultdict |
带默认值的字典 | 分组统计、避免键缺失错误 |
namedtuple |
命名元组 | 替代简单类、提升代码可读性 |
OrderedDict |
保持插入顺序 | 配置管理、有序数据存储 |
ChainMap |
多字典逻辑合并 | 配置层叠、多命名空间查询 |
五、总结
collections 模块通过高效的数据结构和工具函数，解决了以下常见问题：
- 结构化数据存储：
namedtuple、deque、Counter。 - 默认值处理：
defaultdict。 - 顺序控制：
OrderedDict。 - 多数据源整合：
ChainMap。 - 不可变视图：
MappingProxyType。
