写点什么

简洁灵活:Python 中基于字段的不使用元类的 ORM 实现

作者:申公豹
  • 2023-12-24
    内蒙古
  • 本文字数:3271 字

    阅读完需:约 11 分钟

不使用元类的简单 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 self
def __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
@classmethod def 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 = k name = v.name _meta[k] = (name, v)
table = cls.__dict__.get('__table__') table = cls.__name__ if table is None else table _meta['__table__'] = table
setattr(cls, '_meta', _meta)
return _meta
def 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_tuple column_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 = cls        self._order_columns = None        self._desc = ''        self._meta = self._model.get_class_meta()        self._compare = None        self.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 sql
def 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 = sql print(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 self
def where(self, compare: "Compare") -> "Query": self._compare = compare return 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 self
def 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 self
def 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 实现更加强大和灵活。

总结

在这个系列的文章中,我们通过不使用元类的方式,实现了一个简单的 Python ORM。我们定义了 Field 类表示数据库字段,Model 类表示数据库表,以及 Query 类用于构建和执行查询。通过这个实现,我们可以方便地进行数据操作,构建灵活的查询条件,而不需要深入理解元类的概念。


然而,这个简单的 ORM 仍然有一些局限性,例如不支持复杂的表关联等功能。在实际项目中,选择使用元类的 ORM 实现或其他成熟的 ORM 框架取决于项目的需求和团队的技术选型。希望这个实现能够为你提供一种不同的思路,促使更多的思考和探讨。



发布于: 11 分钟前阅读数: 5
用户头像

申公豹

关注

申公豹本豹-这是新号,原号已废弃 2023-06-05 加入

InfoQ签约作者

评论

发布
暂无评论
简洁灵活:Python中基于字段的不使用元类的ORM实现_Python_申公豹_InfoQ写作社区