不使用元類的簡單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 實現更加強大和靈活。
總結
在這個系列的文章中,我們通過不使用元類的方式,實現了一個簡單的 Python ORM。我們定義了?Field
?類表示數據庫字段,Model
?類表示數據庫表,以及?Query
?類用于構建和執行查詢。通過這個實現,我們可以方便地進行數據操作,構建靈活的查詢條件,而不需要深入理解元類的概念。
然而,這個簡單的 ORM 仍然有一些局限性,例如不支持復雜的表關聯等功能。在實際項目中,選擇使用元類的 ORM 實現或其他成熟的 ORM 框架取決于項目的需求和團隊的技術選型。希望這個實現能夠為你提供一種不同的思路,促使更多的思考和探討。