Python不使用元类的ORM实现

不使用元类的简单ORM实现

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

Field类

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

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

Compare类

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

class Compare:def __init__(self, left: Field, operation: str, right: Any):self.condition = f'`{left.name}` {operation} "{right}"'def __or__(self, other: "Compare"):self.condition = f'({self.condition}) OR ({other.condition})'return selfdef __and__(self, other: "Compare"):self.condition = f'({self.condition}) AND ({other.condition})'return self

Model类

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

class Model:def __init__(self, **kwargs):_meta = self.get_class_meta()for k, v in kwargs.items():if k in _meta:self.__dict__[k] = v@classmethoddef get_class_meta(cls) -> Dict:if hasattr(cls, '_meta'):return cls.__dict__['_meta']_meta = {}for k, v in cls.__dict__.items():if isinstance(v, Field):if v.name is None:v.name = kname = v.name_meta[k] = (name, v)table = cls.__dict__.get('__table__')table = cls.__name__ if table is None else table_meta['__table__'] = tablesetattr(cls, '_meta', _meta)return _metadef insert(self):_meta = self.get_class_meta()column_li = []val_li = []for k, v in self.__dict__.items():field_tuple = _meta.get(k)if field_tuple:column, field = field_tuplecolumn_li.append(column)val = str(v) if field.column_type == 'INT' else f'"{str(v)}"'val_li.append(val)sql = f'INSERT INTO {_meta["__table__"]} ({",".join(column_li)}) VALUES ({",".join(val_li)});'print(sql)

Query类

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

class Query:def __init__(self, cls: Model):self._model = clsself._order_columns = Noneself._desc = ''self._meta = self._model.get_class_meta()self._compare = Noneself.sql = ''def _get(self) -> str:sql = ''if self._compare:sql += f' WHERE {self._compare.condition}'if self._order_columns:sql += f' ORDER BY {self._order_columns}'sql += f' {self._desc}'return sqldef get(self, *args: Field) -> List[Model]:sql = self._get()table = self._meta['__table__']column_li = []if len(args) > 0:for field in args:column_li.append(f'`{field.name}`')else:for v in self._meta.values():if type(v) == tuple and isinstance(v[1], Field):column_li.append(f'`{v[0]}`')columns = ",".join(column_li)sql = f'SELECT {columns} FROM {table} {sql}'self.sql = sqlprint(self.sql)def order_by(self, columns: Union[List, str], desc: bool = False) -> "Query":if isinstance(columns, str):self._order_columns = f'`{columns}`'elif isinstance(columns, list):self._order_columns = ','.join([f'`{x}`' for x in columns])self._desc = 'DESC' if desc else ''return selfdef where(self, compare: "Compare") -> "Query":self._compare = comparereturn self

示例使用

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

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

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

扩展查询功能

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

支持 LIMIT 和 OFFSET

class Query:# ...def limit(self, num: int) -> "Query":self.sql += f' LIMIT {num}'return selfdef offset(self, num: int) -> "Query":self.sql += f' OFFSET {num}'return self

支持 GROUP BY 和 HAVING

class Query:# ...def group_by(self, columns: Union[List, str]) -> "Query":if isinstance(columns, str):columns = [columns]self.sql += f' GROUP BY {",".join([f"`{x}`" for x in columns])}'return selfdef having(self, condition: Compare) -> "Query":self.sql += f' HAVING {condition.condition}'return self

示例用法

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

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