侧边栏壁纸
博主头像
Zeeland

全栈算法工程师 | 大模型创业 | 开源项目分享 | Python开发者 | @Promptulate Founder | @SparkLab cofounder | @LangChainAI Top Contributor | @CogitLab core member

  • 累计撰写 61 篇文章
  • 累计创建 47 个标签
  • 累计收到 7 条评论

目 录CONTENT

文章目录

【ORM】Python如何自己手写一个ORM框架?

Zeeland
2023-06-07 / 0 评论 / 0 点赞 / 287 阅读 / 4,492 字

简介

前段时间因为一个开源项目cushy-storage需要更新新特性,因此着手研究了一下ORM,相关pr代码已经合并入cushy-storage的最新版本中。

cushy-storage是一个基于磁盘缓存的Python框架,让你无需花费精力在如何制订一套数据存储规范上,字典般的操作可以减少很多开发的成本。如果你有对本地文件数据操作的需求,使用本框架可以轻松的进行数据的本地存储。

但是有的时候,我们需要对自定义对象进行增删改查,如何可以像查询数据库一下对本地文件存储的数据进行快速查询呢?因此对于cushy-storage来说有这样一个需求,我们对cushy-storage封装一个ORM框架,让用户使用ORM更加轻松地对数据进行交互,因此有了本文。

本文将会介绍什么是ORM,并且介绍笔者如何把ORM嵌入到Cushy-storage中,基于ORM进行方便的本地文件数据读取。

什么是ORM?

ORM框架是对象关系映射框架的简称,是一种可以让程序员使用面向对象的方式来操作数据库的工具,它将数据库中的表和行转换成对象和属性,使得程序员可以通过面向对象的方式来进行数据库操作,这样可以避免了直接使用SQL语言对数据库进行操作时的复杂性和繁琐性。

在Python中,使用ORM框架可以轻松地实现对数据库的增删改查操作。下面是一个使用ORM框架进行CRUD操作的示例代码:

首先,安装ORM框架的库。在Python中非常流行的ORM框架有多种,最受欢迎的是SQLAlchemy,可以通过以下命令来安装:

pip install sqlalchemy

其次,创建一个数据库表格,这里我们以一个学生表为例,假设表名为Student,包含以下字段:

字段名称 字段类型
id int
name varchar
age int
sex varchar
grade varchar

代码实现如下所示:

from sqlalchemy import create_engine, Column, Integer, String, MetaData, Table

engine = create_engine('mysql://user:password@host/database')

metadata = MetaData()

Student = Table(
    'Student',
    metadata,
    Column('id', Integer, primary_key=True, autoincrement=True),
    Column('name', String(50), nullable=False),
    Column('age', Integer),
    Column('sex', String(10)),
    Column('grade', String(10))
)

metadata.create_all(engine)

其中,create_engine方法用于创建一个数据库引擎,MetaData用于定义某个数据库的元数据信息,Table用于定义某个表的结构。

接着,进行增删改查操作。示例代码如下:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import Student

engine = create_engine('mysql://user:password@host/database')

Session = sessionmaker(bind=engine)
session = Session()

# 增加数据
student_1 = Student(name='张三', age=18, sex='男', grade='高一')
session.add(student_1)
session.commit()

# 查询数据
students = session.query(Student).filter(Student.age >= 18).all()
for student in students:
    print(student.name)

# 修改数据
student_2 = session.query(Student).filter(Student.name == '张三').first()
student_2.grade = '高三'
session.commit()

# 删除数据
student_3 = session.query(Student).filter(Student.name == '李四').first()
session.delete(student_3)
session.commit()

以上代码通过session对象实现对数据库的增、删、查、改。其中,add方法用于添加数据,查询方法使用了ORM内置的一些查询方法,比如queryfilter等,delete方法用于删除数据。

如何自己写一个ORM框架

编写一个完整的ORM框架是一项复杂的任务,需要对数据库、Python编程以及设计模式有深刻的理解。以下是实现ORM框架的一般步骤:

  1. 以一种统一的方式处理数据库连接,例如使用Python中内置的sqlite3模块或第三方库如psycopg2来管理连接。在本文中,数据库就是本地文件,因此没有数据库连接等操作,因此可以做的很轻量。

  2. 定义一个基本的模型类,用于描述一个数据库表,包括表名(tableName)、列名(columnName)和数据类型(dataType)等属性,还可以包括一些CRUD操作方法和过滤器等。在本文中,一个类就是一个表名,而表名可以作为文件名,如你有一个User对象,那么你可以将User作为文件名进行存储数据。

  3. 编写查询语句,将查询结果存储在模型类中。

  4. 实现对模型类的修改和删除操作。

  5. 添加数据验证和错误处理机制,确保ORM框架操作数据库的数据的完整性。

  6. 扩展ORM框架,使其支持多种数据库系统(例如PostgreSQL,MySQL等)。

编写ORM框架需要很多技能和经验。如果你想了解更多细节,建议你参考一些优秀的开源ORM框架的源代码,例如Django、SQLAlchemy、Peewee等,并参考相关文档。

下面我将会结合cushy-storage的能力快速构建一个ORM框架。

cushy-storage介绍

cushy-storage是一个基于磁盘缓存的Python库,可以将Python对象序列化后缓存到磁盘中,以便下次使用时直接读取,从而提高程序的执行效率。另一方面,cushy-storage让你无需花费精力在如何制订一套数据存储规范上,字典般的操作可以减少很多开发的成本。

下面我将会快速上手cushy-storage带你体验一下cushy-storage的方便之处。

  1. 第三方库下载
pip install cushy-storage --upgrade 

为了进行数据存储与读取,我们需要用到CushyDict类,你可以像操作字典一样操作CushyDict;其增加了对值进行序列化和反序列化的功能,可以存储任意类型的数据。 此外,CushyDict支持多种序列化算法 (pickle和json)和压缩算法(zlib和lzma),可以根据需要选择不同的算法进行数据压缩和序列化,下面是一些简单的使用教程。

  • 存储Python基本数据类型
from cushy_storage import CushyDict

# 初始化cache,保存在./data文件夹下
cache = CushyDict('./data')

cache['key'] = {'value': 42}
print(cache['key'])

cache['a'] = 1
print(cache['a'])

cache['b'] = "hello world"
print(cache['b'])

cache['arr'] = [1, 2, 3, 4, 5]
print(cache['arr'])

以cache[‘arr’] = [1, 2, 3, 4, 5]为例,在指令这段代码之后,CushyDict会将数据存储到指令文件夹下。

  • 存储自定义数据类型

from cushy_storage import CushyDict


class User:
    def __init__(self, name, age):
        self.name = name
        self.age = age


def main():
    cache = CushyDict(serialize='pickle')
    user = User("Jack", 18)
    cache['user'] = user

    user = cache['user']
    print(type(user))
    print(cache['user'].name)
    print(cache['user'].age)


if __name__ == '__main__':
    main()

需要说明的是,如果你有定义复杂数据的需求,如List里面存json;或者你没有去文件下看原数据的需求,则推荐使用pickle的方式来进行数据存储。

  • 如果在初始化的时候不传入参数,则默认保存在./cache文件夹下
from cushy_storage import CushyDict

cache = CushyDict()
  • 判断key是否存在(和字典操作同理)
from cushy_storage import CushyDict

cache = CushyDict()
if 'key' in cache:
    print("key exist")
else:
    print("key not exist")

将ORM框架嵌入cushy-storage

在简单介绍了cushy-storage的使用后,我们可以尝试对cushy-storage封装ORM框架,首先我们可以看到,使用其进行存储自定义数据类型是很方便的,如下所示。


from cushy_storage import CushyDict


class User:
    def __init__(self, name, age):
        self.name = name
        self.age = age


def main():
    cache = CushyDict(serialize='pickle')
    user = User("Jack", 18)
    cache['user'] = user

    user = cache['user']
    print(type(user))
    print(cache['user'].name)
    print(cache['user'].age)


if __name__ == '__main__':
    main()

需要实现的功能

因此,我们需要在这个基础上做改造,首先对于User,'user'这个key中需要存的是所有的user信息,这个时候我们可以认为"User"就是数据库的表名(即本项目的文件名),在这个文件内会存储所有用户信息,而我们需要实现:

  • 在文件末尾添加新用户
  • 删除指令某一用户
  • 查询某一指定用户或者满足条件的所有用户
  • 修改某个用户的数据

通过实现上面四个目标,从而实现ORM的增删改查功能。

BaseORMModel的构建

另一方面,我们不能只针对User这一个类做增删改查的适配,所有想要用cushy-storage存储的自定义类我们都要提供这样的功能,怎么做呢?很简单,构建一个BaseModel让这些自定义类去继承就好了,如下:

import uuid
from abc import ABC

class BaseORMModel(ABC):

    def __init__(self):
        self.__name__ = type(self).__name__
        self._unique_id = str(uuid.uuid4())

BaseORMModel__name__到后面就可以用作我们的文件名,如果存储User,则文件名为"User",如果存储Node,则文件名为"Node",以此类推。_unique_id为每个对象的唯一标识符,用uuid进行生成,在查询的时候可以确保数据的唯一性。

事实上,我们在关系型数据库中,对于每一个表,我们也需要构建一个主键作为唯一不可重复字段,用于检索。

构建完BaseORMModel之后,想要使用cushy-storage的类只需要继承这个类就好了,如下所示:

class User(BaseORMModel):

    def __init__(self, name, age):
        super().__init__()
        self.name = name
        self.age = age

CushyOrmCache与ORMMixin的构建

接下来,我们需要构建一个类,这个类需要具有CushyDict的所有特性,又需要具有ORM框架的功能,在这里,我使用了多继承的思想进行设计,因为我们需要隔离直接嵌入CushyDict类中所引发了一些不必要的冲突。

CushyOrmCache定义如下所示, ORM框架的功能通过ORMMixin进行引入,这种设计模式与Django Rest FrameworkModelViewSet的设计一致。

class CushyOrmCache(CushyDict, ORMMixin):

    def __init__(
            self,
            path: str = get_default_cache_path(),
            compress: Union[str, Tuple[Callable, Callable], None] = None,
    ):
        super().__init__(path, compress, "pickle")

这里可以看到,关于序列化定死为pickle,因为json不可以进行python类的数据存储。

接下来,我们需要构建ORMMixin,ORMMixin中需要有增删改查的功能,一个大概的框架如下所示。

class ORMMixin(ABC):

    def _get_original_data_from_cache(self, class_name_or_obj: Union[str, type(BaseORMModel)]) -> List[BaseORMModel]:
        pass

    def query(self, class_name_or_obj: Union[str, type(BaseORMModel)]):
        """query all objects by class name"""
        pass

    def add(self, obj: Union[BaseORMModel, List[BaseORMModel]]) -> QuerySet:
        pass

    def delete(self, obj: BaseORMModel):
        """delete obj by obj._unique_id"""
        pass

    def update_obj(self, obj: BaseORMModel):
        pass

QuerySet的构建

如果想要更方便地对数据进行查询,我们还需要将查询出的数据通过QuerySet进行包装,让其具有条件筛选、返回所有数据、返回第一个数据等功能。

class QuerySet:

    def __init__(self, obj: Union[List[BaseORMModel], BaseORMModel], name: Optional[str] = None):
        self._data: List[BaseORMModel] = obj
        if isinstance(obj, BaseORMModel):
            self._data = [obj]
        self.__name__ = name if name else self._data[0].__name__

    @classmethod
    def _from_filter(cls, obj: Union[List[BaseORMModel], BaseORMModel]):
        """generate a new queryset from filter"""
        return cls(obj)

    def filter(self, **kwargs):
        """
        filter by specified parameter
        Args:
            **kwargs: The property of the object you want to query

        Returns: return a new QuerySet object

        Examples:
            class User(BaseORMModel):
                def __init__(self, name, age):
                    super().__init__()
                    self.name = name
                    self.age = age
            # get all user, you will get a List[User] type data.
            # Actually, it will get two users named "jack" and "jasmine".
            orm_cache.query("User").filter(age=18).all()
            # get first in queryset, you will get a User type data
            orm_cache.query("User").filter(name="jack").first()
            # filter by multiple parameters
            orm_cache.query("User").filter(name="jack", age=18).first()
        """
        result: List[BaseORMModel] = []
        for item in self._data:
            for query_key in kwargs.keys():
                if item.__dict__[query_key] != kwargs[query_key]:
                    continue
                result.append(item)

        return self._from_filter(result)

    def all(self) -> Optional[List[BaseORMModel]]:
        return self._data

    def first(self) -> Optional[BaseORMModel]:
        if len(self._data) == 0:
            return None
        return self._data[0]

    def print_all(self):
        for item in self._data:
            print(f"[cushy-storage orm] {item.__dict__}")

最后我们完善一下ORMMixin

class ORMMixin(ABC):

    def _get_original_data_from_cache(self, class_name_or_obj: Union[str, type(BaseORMModel)]) -> List[BaseORMModel]:
        class_name = _get_class_name(class_name_or_obj)
        if class_name not in self:
            self.__setitem__(class_name, [])
        return self.__getitem__(class_name)

    def query(self, class_name_or_obj: Union[str, type(BaseORMModel)]):
        """query all objects by class name"""
        original_result = self._get_original_data_from_cache(class_name_or_obj)
        if len(original_result) == 0:
            return QuerySet(original_result, name=_get_class_name(class_name_or_obj))
        return QuerySet(original_result)

    def add(self, obj: Union[BaseORMModel, QuerySet, List[BaseORMModel]]) -> QuerySet:
        obj_name = obj.__name__ if not isinstance(obj, list) else obj[0].__name__
        original_result = self._get_original_data_from_cache(obj_name)

        if isinstance(obj, BaseORMModel):
            original_result.append(obj)
        elif isinstance(obj, QuerySet):
            original_result += obj.all()
        else:
            original_result += obj

        self[obj_name] = original_result
        return QuerySet(self[obj_name])

    def delete(self, obj: BaseORMModel):
        """delete obj by obj._unique_id"""
        original_result: List[BaseORMModel] = self._get_original_data_from_cache(obj.__name__)
        copy_result: List[BaseORMModel] = original_result.copy()

        for item in copy_result:
            if item._unique_id == obj._unique_id:
                copy_result.remove(item)
                return self.__setitem__(obj.__name__, copy_result)

        raise ValueError(f"can not found object: {obj}")

    def update_obj(self, obj: BaseORMModel):
        original_result: List[BaseORMModel] = self._get_original_data_from_cache(obj.__name__)
        copy_result: List[BaseORMModel] = original_result.copy()

        for i in range(len(copy_result)):
            if copy_result[i]._unique_id == obj._unique_id:
                copy_result[i] = obj
                return self.__setitem__(obj.__name__, copy_result)

        raise ValueError(f"can not found object: {obj}")

具体实现细节可以看代码,个人认为实现不是很难。但事实上,这只是一个简单的ORM框架,有如果要实现更加复杂的功能,如事务、复杂条件查询(大于小于、模糊查询)等功能,则大家可以自己探索一下,也欢迎大家往https://github.com/Undertone0809/cushy-storage/项目中增加新的功能。

CushyORMCache的使用

CushyOrmCache是一个基于ORM框架的对象存储,可以十分方便的对对象级数据进行增删改查,下面,我们将会用一些简单的场景介绍其使用方法。

现在我们需要构建一个简单的用户系统,用户系统的数据我们直接保存在本地文件中(当前对象级数据只支持pickle序列化的形式存储),用户的字段简单就好,
只需要一个name和一个age,则我们可以构建如下的操作。

from cushy_storage.orm import BaseORMModel, CushyOrmCache

class User(BaseORMModel):

    def __init__(self, name, age):
        super().__init__()
        self.name = name
        self.age = age

这个示例中,我们实现了一个User类,并且继承了BaseORMModel,在cushy-storage中,如果你想让你的类可以进行ORM操作,就必须要继承这个类。
接着,我们需要初始化CushyOrmCache

orm_cache = CushyOrmCache()

接着,你就可以直接进行User的增删改查操作了。

"""add user"""
user = User("jack", 18)
orm_cache.add(user)
user = User("jasmine", 18)
orm_cache.add(user)

"""query all user"""
users = orm_cache.query(User).all()
orm_cache.query(User).print_all()

"""query by filter"""
# get all user, you will get a List[User] type data.
# Actually, it will get two users named "jack" and "jasmine".
orm_cache.query("User").filter(age=18).all()
# get first in queryset, you will get a User type data
orm_cache.query("User").filter(name="jack").first()
# filter by multiple parameters
orm_cache.query("User").filter(name="jack", age=18).first()

"""update"""
user = orm_cache.query("User").filter(name='jack').first()
user.age = 18
orm_cache.update_obj(user)

"""delete"""
user = orm_cache.query("User").filter(name="jack").first()
orm_cache.delete(user)
orm_cache.query(User).print_all()

完整代码如下:

from cushy_storage.orm import BaseORMModel, CushyOrmCache


class User(BaseORMModel):

    def __init__(self, name, age):
        super().__init__()
        self.name = name
        self.age = age


orm_cache = CushyOrmCache()

"""add user"""
user = User("jack", 18)
orm_cache.add(user)
user = User("jasmine", 18)
orm_cache.add(user)

"""query all user"""
users = orm_cache.query(User).all()
orm_cache.query(User).print_all()

"""query by filter"""
# get all user, you will get a List[User] type data.
# Actually, it will get two users named "jack" and "jasmine".
orm_cache.query("User").filter(age=18).all()
# get first in queryset, you will get a User type data
orm_cache.query("User").filter(name="jack").first()
# filter by multiple parameters
orm_cache.query("User").filter(name="jack", age=18).first()

"""update"""
user = orm_cache.query("User").filter(name='jack').first()
user.age = 18
orm_cache.update_obj(user)

"""delete"""
user = orm_cache.query("User").filter(name="jack").first()
orm_cache.delete(user)
orm_cache.query(User).print_all()

需要注意的是,你可以通过在query()中传入User对象来进行数据的查询,也可以直接传入"User"字符串进行数据的查询(这里的设计思路和数据库的表是一样的
,User是表名)

总结

本文介绍了ORM是什么,并且介绍了如何构建一个属于自己的ORM框架的,在本文中,以cushy-storage磁盘缓存框架为依托,构建了一个基于cushy-storage的ORM框架,可以轻松地进行对象的增删改查操作,并且支持复杂条件查询、批量用户返回等功能。

0

评论区