4种Python中基于字段的不使用元类的ORM实现方法

本文分享自华为云社区《Python中基于字段的不使用元类的ORM实现》，作者： 柠檬味拥抱 。
不使用元类的简单ORM实现

在 Python 中，ORM（Object-Relational Mapping）是一种将对象和数据库之间的映射关系进行转换的技术，使得通过面向对象的方式来操作数据库更加方便。通常，我们使用元类（metaclass）来实现ORM，但是本文将介绍一种不使用元类的简单ORM实现方式。
Field类

首先，我们定义一个Field类，用于表示数据库表中的字段。这个类包含字段的名称和类型等信息，并且支持一些比较操作，以便后续构建查询条件。

1. class Field:
2.     def __init__(self, **kwargs):
3.         self.name = kwargs.get('name')
4.         self.column_type = kwargs.get('column_type')
6.     def __eq__(self, other):
7.         return Compare(self, '=', other)
9.     # 其他比较操作略...

复制代码

Compare类

为了构建查询条件，我们引入了一个Compare类，用于表示字段之间的比较关系。它可以支持链式操作，构建复杂的查询条件。

1. class Compare:
2.     def __init__(self, left: Field, operation: str, right: Any):
3.         self.condition = f'`{left.name}` {operation} "{right}"'
5.     def __or__(self, other: "Compare"):
6.         self.condition = f'({self.condition}) OR ({other.condition})'
7.         return self
9.     def __and__(self, other: "Compare"):
10.         self.condition = f'({self.condition}) AND ({other.condition})'
11.         return self

复制代码

Model类

接下来，我们定义Model类，表示数据库中的表。该类通过Field类的实例来定义表的字段，并提供了插入数据的方法。

1. class Model:
2.     def __init__(self, **kwargs):
3.         _meta = self.get_class_meta()
5.         for k, v in kwargs.items():
6.             if k in _meta:
7.                 self.__dict__[k] = v
9.     @classmethod
10.     def get_class_meta(cls) -> Dict:
11.         if hasattr(cls, '_meta'):
12.             return cls.__dict__['_meta']
13.         _meta = {}
15.         for k, v in cls.__dict__.items():
16.             if isinstance(v, Field):
17.                 if v.name is None:
18.                     v.name = k
19.                 name = v.name
20.                 _meta[k] = (name, v)
22.         table = cls.__dict__.get('__table__')
23.         table = cls.__name__ if table is None else table
24.         _meta['__table__'] = table
26.         setattr(cls, '_meta', _meta)
28.         return _meta
30.     def insert(self):
31.         _meta = self.get_class_meta()
32.         column_li = []
33.         val_li = []
35.         for k, v in self.__dict__.items():
36.             field_tuple = _meta.get(k)
37.             if field_tuple:
38.                 column, field = field_tuple
39.                 column_li.append(column)
40.                 val = str(v) if field.column_type == 'INT' else f'"{str(v)}"'
41.                 val_li.append(val)
43.         sql = f'INSERT INTO {_meta["__table__"]} ({",".join(column_li)}) VALUES ({",".join(val_li)});'
44.         print(sql)

复制代码

Query类

最后，我们实现了Query类，用于构建数据库查询。这个类支持链式调用，可以设置查询条件、排序等。

1. class Query:
2.     def __init__(self, cls: Model):
3.         self._model = cls
4.         self._order_columns = None
5.         self._desc = ''
6.         self._meta = self._model.get_class_meta()
7.         self._compare = None
8.         self.sql = ''
10.     def _get(self) -> str:
11.         sql = ''
13.         if self._compare:
14.             sql += f' WHERE {self._compare.condition}'
16.         if self._order_columns:
17.             sql += f' ORDER BY {self._order_columns}'
19.         sql += f' {self._desc}'
20.         return sql
22.     def get(self, *args: Field) -> List[Model]:
23.         sql = self._get()
24.         table = self._meta['__table__']
26.         column_li = []
28.         if len(args) > 0:
29.             for field in args:
30.                 column_li.append(f'`{field.name}`')
31.         else:
32.             for v in self._meta.values():
33.                 if type(v) == tuple and isinstance(v[1], Field):
34.                     column_li.append(f'`{v[0]}`')
36.         columns = ",".join(column_li)
37.         sql = f'SELECT {columns} FROM {table} {sql}'
38.         self.sql = sql
39.         print(self.sql)
41.     def order_by(self, columns: Union[List, str], desc: bool = False) -> "Query":
42.         if isinstance(columns, str):
43.             self._order_columns = f'`{columns}`'
44.         elif isinstance(columns, list):
45.             self._order_columns = ','.join([f'`{x}`' for x in columns])
47.         self._desc = 'DESC' if desc else ''
48.         return self
50.     def where(self, compare: "Compare") -> "Query":
51.         self._compare = compare
52.         return self

复制代码

示例使用


现在，我们可以定义一个模型类，并使用这个简单的ORM实现进行数据操作。

1. class User(Model):
2.     name = Field()
3.     age = Field()
5. # 插入数据
6. user = User(name='Tom', age=24)
7. user.insert()
9. # 构建查询条件并查询数据
10. User.query().where((User.name == 'Tom') & (User.age >= 20)).order_by('age').get()

复制代码

这样，我们就完成了一个不使用元类的简单ORM实现。尽管相较于使用元类的方式，代码结构更为简单，但在实际应用中，根据项目需求和团队的约定，选择合适的实现方式是很重要的。
我们已经介绍了一个基于 Python 的简单 ORM 实现，它不依赖于元类。在这一部分，我们将继续探讨这个实现，深入了解查询构建和更复杂的用法。
扩展查询功能

我们的查询功能还比较简单，为了更好地支持复杂查询，我们可以添加更多的查询方法和条件。
支持 LIMIT 和 OFFSET

1. class Query:
2.     # ...
4.     def limit(self, num: int) -> "Query":
5.         self.sql += f' LIMIT {num}'
6.         return self
8.     def offset(self, num: int) -> "Query":
9.         self.sql += f' OFFSET {num}'
10.         return self

复制代码

支持 GROUP BY 和 HAVING

1. class Query:
2.     # ...
4.     def group_by(self, columns: Union[List, str]) -> "Query":
5.         if isinstance(columns, str):
6.             columns = [columns]
7.         self.sql += f' GROUP BY {",".join([f"`{x}`" for x in columns])}'
8.         return self
10.     def having(self, condition: Compare) -> "Query":
11.         self.sql += f' HAVING {condition.condition}'
12.         return self

复制代码

示例用法

1. class User(Model):
2.     name = Field()
3.     age = Field()
5. # 插入数据
6. user = User(name='Tom', age=24)
7. user.insert()
9. # 构建查询条件并查询数据
10. query = User.query().where((User.name == 'Tom') & (User.age >= 20)).order_by('age').limit(1).offset(0)
11. query.get(User.name, User.age)  # 仅查询指定字段
13. # 更复杂的查询
14. query = User.query().group_by('age').having((User.age > 20) & (User.age < 30)).order_by('age').limit(10).offset(0)
15. query.get(User.age, User.count(User.name))  # 查询年龄在20到30之间的用户数量

复制代码

通过引入额外的查询功能，我们使得这个简单的 ORM 实现更加强大和灵活。
总结

在这个系列的文章中，我们通过不使用元类的方式，实现了一个简单的 Python ORM。我们定义了 Field 类表示数据库字段，Model 类表示数据库表，以及 Query 类用于构建和执行查询。通过这个实现，我们可以方便地进行数据操作，构建灵活的查询条件，而不需要深入理解元类的概念。
然而，这个简单的 ORM 仍然有一些局限性，例如不支持复杂的表关联等功能。在实际项目中，选择使用元类的 ORM 实现或其他成熟的 ORM 框架取决于项目的需求和团队的技术选型。希望这个实现能够为你提供一种不同的思路，促使更多的思考和探讨。
点击关注，第一时间了解华为云新鲜技术~
 

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！