Python 描述符与 property：ORM 框架的核心秘密

1
class MyDescriptor:
2
    def __get__(self, obj, objtype=None):
3
        print(f"__get__ 被调用: obj={obj}, objtype={objtype}")
4
        return "descriptor value"
5

6
    def __set__(self, obj, value):
7
        print(f"__set__ 被调用: obj={obj}, value={value}")
8

9
class MyClass:
10
    attr = MyDescriptor()  # 类属性是描述符实例
11

12
obj = MyClass()
13
print(obj.attr)   # 触发 __get__
14
obj.attr = 42     # 触发 __set__
15
# 输出:
16
# __get__ 被调用: obj=<__main__.MyClass object at 0x...>, objtype=<class '__main__.MyClass'>
17
# descriptor value
18
# __set__ 被调用: obj=<__main__.MyClass object at 0x...>, value=42

1
class DataDesc:
2
    def __get__(self, obj, objtype=None):
3
        return "from descriptor"
4
    def __set__(self, obj, value):
5
        obj.__dict__["_stored"] = value
6

7
class NonDataDesc:
8
    def __get__(self, obj, objtype=None):
9
        return "from non-data descriptor"
10

11
class MyClass:
12
    data_desc = DataDesc()         # 数据描述符
13
    non_data_desc = NonDataDesc()  # 非数据描述符
14

15
obj = MyClass()
16

17
# 数据描述符：实例 __dict__ 无法覆盖
18
obj.__dict__["data_desc"] = "hacked"
19
print(obj.data_desc)  # 仍然是 "from descriptor"（数据描述符优先）
20

21
# 非数据描述符：实例 __dict__ 可以覆盖
22
obj.__dict__["non_data_desc"] = "hacked"
23
print(obj.non_data_desc)  # "hacked"（实例 __dict__ 优先）

1
# ❌ 无限递归的典型案例
2
class Bad:
3
    @property
4
    def value(self):
5
        return self.value  # 触发 property 的 __get__ → 又调用 self.value → 无限递归
6

7
    @value.setter
8
    def value(self, v):
9
        self.value = v  # 触发 property 的 __set__ → 又调用 self.value = v → 无限递归
10

11
# ✅ 正确做法
12
class Good:
13
    @property
14
    def value(self):
15
        return self._value  # 读取底层存储
16

17
    @value.setter
18
    def value(self, v):
19
        self._value = v  # 写入底层存储

1
class Circle:
2
    def __init__(self, radius):
3
        self._radius = radius
4

5
    @property
6
    def radius(self):
7
        return self._radius
8

9
    @property
10
    def area(self):
11
        """只读计算属性"""
12
        import math
13
        return math.pi * self._radius ** 2
14

15
c = Circle(5)
16
print(c.area)  # 78.53981633974483
17
c.area = 100   # AttributeError: can't set attribute 'area'

1
import json
2

3
class Config:
4
    def __init__(self, filepath):
5
        self._filepath = filepath
6
        self._data = None  # 初始为 None
7

8
    @property
9
    def data(self):
10
        """首次访问时才读取文件"""
11
        if self._data is None:
12
            print("读取配置文件...")
13
            with open(self._filepath) as f:
14
                self._data = json.load(f)
15
        return self._data
16

17
# 使用
18
config = Config("config.json")
19
print("尚未访问 data")
20
# 此时文件未被读取
21
print(config.data)  # 触发读取
22
print(config.data)  # 使用缓存，不再读取

1
class TypedField:
2
    """类型校验描述符"""
3

4
    def __init__(self, field_type):
5
        self.field_type = field_type
6
        self.storage_name = None  # 后续由 __set_name__ 填充
7

8
    def __set_name__(self, owner, name):
9
        """Python 3.6+：类创建时自动调用，告知描述符自己的属性名"""
10
        self.storage_name = f"_field_{name}"
11

12
    def __get__(self, obj, objtype=None):
13
        if obj is None:
14
            return self  # 通过类访问时返回描述符本身
15
        return getattr(obj, self.storage_name, None)
16

17
    def __set__(self, obj, value):
18
        if not isinstance(value, self.field_type):
19
            raise TypeError(
20
                f"{self.storage_name} 必须是 {self.field_type.__name__}，"
21
                f"实际是 {type(value).__name__}"
22
            )
23
        setattr(obj, self.storage_name, value)
24

25
class User:
26
    name = TypedField(str)
27
    age = TypedField(int)
28
    score = TypedField(float)
29

30
user = User()
31
user.name = "Alice"    # ✅
32
user.age = 30          # ✅
33
user.score = 95.5      # ✅
34
user.age = "thirty"    # ❌ TypeError

1
class ValidatedField:
2
    """支持范围校验的描述符"""
3

4
    def __init__(self, validator=None, default=None):
5
        self.validator = validator
6
        self.default = default
7
        self.storage_name = None
8

9
    def __set_name__(self, owner, name):
10
        self.storage_name = f"_validated_{name}"
11

12
    def __get__(self, obj, objtype=None):
13
        if obj is None:
14
            return self
15
        return getattr(obj, self.storage_name, self.default)
16

17
    def __set__(self, obj, value):
18
        if self.validator and not self.validator(value):
19
            raise ValueError(f"{self.storage_name} 的值 {value!r} 未通过校验")
20
        setattr(obj, self.storage_name, value)
21

22
# 使用
23
class Product:
24
    price = ValidatedField(
25
        validator=lambda v: v > 0,
26
        default=0.0
27
    )
28
    quantity = ValidatedField(
29
        validator=lambda v: 0 <= v <= 10000,
30
        default=0
31
    )
32

33
p = Product()
34
p.price = 99.9      # ✅
35
p.quantity = 500     # ✅
36
p.price = -10        # ❌ ValueError
37
p.quantity = 99999   # ❌ ValueError

1
class StringField(TypedField):
2
    def __init__(self, max_length=255):
3
        super().__init__(str)
4
        self.max_length = max_length
5

6
    def __set__(self, obj, value):
7
        super().__set__(obj, value)
8
        if len(value) > self.max_length:
9
            raise ValueError(
10
                f"{self.storage_name} 长度不能超过 {self.max_length}"
11
            )
12

13
class IntegerField(TypedField):
14
    def __init__(self, min_value=None, max_value=None):
15
        super().__init__(int)
16
        self.min_value = min_value
17
        self.max_value = max_value
18

19
    def __set__(self, obj, value):
20
        super().__set__(obj, value)
21
        if self.min_value is not None and value < self.min_value:
22
            raise ValueError(f"{self.storage_name} 不能小于 {self.min_value}")
23
        if self.max_value is not None and value > self.max_value:
24
            raise ValueError(f"{self.storage_name} 不能大于 {self.max_value}")
25

26
class Article:
27
    title = StringField(max_length=200)
28
    content = StringField(max_length=10000)
29
    view_count = IntegerField(min_value=0, max_value=1000000000)
30
    is_published = TypedField(bool)
31

32
article = Article()
33
article.title = "Python 描述符详解"
34
article.content = "这是一篇关于..." * 50
35
article.view_count = 0
36
article.is_published = False
37

38
article.view_count = -1  # ❌ ValueError
39
article.title = "x" * 300  # ❌ ValueError

1
import sqlite3
2

3
class Field:
4
    """字段描述符"""
5

6
    def __init__(self, db_type, primary_key=False):
7
        self.db_type = db_type
8
        self.primary_key = primary_key
9
        self.name = ""
10

11
    def __set_name__(self, owner, name):
12
        self.name = name
13

14
    def __get__(self, obj, objtype=None):
15
        if obj is None:
16
            return self
17
        return getattr(obj, f"_value_{self.name}", None)
18

19
    def __set__(self, obj, value):
20
        setattr(obj, f"_value_{self.name}", value)
21
        if hasattr(obj, "_dirty"):
22
            obj._dirty.add(self.name)
23

24

25
class ModelMeta(type):
26
    """元类：收集所有 Field 描述符"""
27

28
    def __new__(mcs, name, bases, namespace):
29
        cls = super().__new__(mcs, name, bases, namespace)
30
        fields = {}
31
        for attr_name, attr_value in namespace.items():
32
            if isinstance(attr_value, Field):
33
                fields[attr_name] = attr_value
34
        cls._fields = fields
35
        return cls
36

37

38
class Model(metaclass=ModelMeta):
39
    """模型基类"""
40

41
    _table_name = ""
42

43
    def __init__(self, **kwargs):
44
        self._dirty = set()
45
        for key, value in kwargs.items():
46
            setattr(self, key, value)
47
        self._dirty.clear()  # 初始赋值不标记为脏
48

49
    @classmethod
50
    def create_table(cls, conn):
51
        """生成 CREATE TABLE 语句"""
52
        columns = []
53
        for name, field in cls._fields.items():
54
            col_def = f"{name} {field.db_type}"
55
            if field.primary_key:
56
                col_def += " PRIMARY KEY AUTOINCREMENT"
57
            columns.append(col_def)
58
        sql = f"CREATE TABLE IF NOT EXISTS {cls._table_name} ({', '.join(columns)})"
59
        conn.execute(sql)
60

61
    def save(self, conn):
62
        """INSERT 或 UPDATE"""
63
        if not hasattr(self, "_value_id") or self._value_id is None:
64
            # INSERT
65
            fields = [(n, f) for n, f in self._fields.items() if not f.primary_key]
66
            names = ", ".join(n for n, _ in fields)
67
            placeholders = ", ".join("?" for _ in fields)
68
            values = [getattr(self, n) for n, _ in fields]
69
            sql = f"INSERT INTO {self._table_name} ({names}) VALUES ({placeholders})"
70
            cursor = conn.execute(sql, values)
71
            if self._fields.get("id"):
72
                setattr(self, "id", cursor.lastrowid)
73
        else:
74
            # UPDATE
75
            if not self._dirty:
76
                return
77
            sets = ", ".join(f"{n} = ?" for n in self._dirty)
78
            values = [getattr(self, n) for n in self._dirty]
79
            values.append(getattr(self, "id"))
80
            sql = f"UPDATE {self._table_name} SET {sets} WHERE id = ?"
81
            conn.execute(sql, values)
82
            self._dirty.clear()
83

84
    @classmethod
85
    def get(cls, conn, obj_id):
86
        """根据 ID 查询"""
87
        cursor = conn.execute(
88
            f"SELECT * FROM {cls._table_name} WHERE id = ?", (obj_id,)
89
        )
90
        row = cursor.fetchone()
91
        if row is None:
92
            return None
93
        columns = [desc[0] for desc in cursor.description]
94
        kwargs = dict(zip(columns, row))
95
        return cls(**kwargs)
96

97

98
# === 使用示例 ===
99
class Person(Model):
100
    _table_name = "persons"
101
    id = Field("INTEGER", primary_key=True)
102
    name = Field("TEXT")
103
    age = Field("INTEGER")
104

105
# 创建数据库
106
conn = sqlite3.connect(":memory:")
107
Person.create_table(conn)
108

109
# 插入
110
p = Person(name="张三", age=28)
111
p.save(conn)
112
print(f"插入后 ID: {p.id}")  # 1
113

114
# 查询
115
p2 = Person.get(conn, 1)
116
print(f"查询: {p2.name}, {p2.age}")  # 张三, 28
117

118
# 更新
119
p2.name = "李四"
120
p2.age = 30
121
p2.save(conn)
122

123
# 验证更新
124
p3 = Person.get(conn, 1)
125
print(f"更新后: {p3.name}, {p3.age}")  # 李四, 30

1
import timeit
2

3
class WithDescriptor:
4
    def __init__(self):
5
        self._x = 0
6

7
    @property
8
    def x(self):
9
        return self._x
10

11
class WithoutDescriptor:
12
    def __init__(self):
13
        self.x = 0
14

15
wd = WithDescriptor()
16
wod = WithoutDescriptor()
17

18
t1 = timeit.timeit(lambda: wd.x, number=1_000_000)
19
t2 = timeit.timeit(lambda: wod.x, number=1_000_000)
20
print(f"描述符: {t1:.3f}s, 直接访问: {t2:.3f}s, 倍数: {t1/t2:.1f}x")
21
# 典型结果：描述符 0.080s, 直接访问 0.020s, 倍数: 4.0x

1
class SharedStateDesc:
2
    counter = 0  # 类级别共享（危险！）
3

4
    def __set_name__(self, owner, name):
5
        SharedStateDesc.counter += 1
6
        print(f"__set_name__ for {owner.__name__}.{name}")
7

8
class A:
9
    x = SharedStateDesc()
10

11
class B:
12
    y = SharedStateDesc()
13

14
# __set_name__ for A.x
15
# __set_name__ for B.y
16
# SharedStateDesc.counter == 2

1
class Parent:
2
    def __init__(self):
3
        self._value = 10
4

5
    @property
6
    def value(self):
7
        return self._value
8

9
class Child(Parent):
10
    @property
11
    def value(self):
12
        return super().value * 2
13

14
c = Child()
15
print(c.value)  # 20 ✅ super() 正确处理了 property

1
class Parent:
2
    def __init__(self):
3
        self._value = 10
4

5
    @property
6
    def value(self):
7
        return self._value
8

9
    @value.setter
10
    def value(self, v):
11
        if v < 0:
12
            raise ValueError("不能为负")
13
        self._value = v
14

15
class Child(Parent):
16
    @Parent.value.setter
17
    def value(self, v):
18
        if v > 1000:
19
            raise ValueError("不能超过 1000")
20
        # 调用父类 setter
21
        Parent.value.fset(self, v)
22

23
c = Child()
24
c.value = 500    # ✅
25
c.value = -1     # ❌ 父类校验
26
c.value = 2000   # ❌ 子类校验