Python实现ORM

ORM全称Object-Relational Mapping，简单的理解就是通过代码的方式操作数据库。ORM 的出现让我们不用关心 SQL 而用自己熟悉的编程语言来操作数据库。在Python开发中，ORM框架也是非常常见的，比如Django自带的ORM，SQLAlchemy等。

ORM的基本概念

建立连接

在使用ORM之前，需要首先建立与数据库的连接。可以使用Python的标准库sqlite3进行测试。

import sqlite3

# 连接数据库
conn = sqlite3.connect('test.db')
print('Opened database successfully')

定义模型

在使用ORM时，需要先定义模型。模型是一个面向对象的类，对应数据库中的一张表。类中的属性对应着表中的列。很多ORM框架都支持简便的语法来定义模型，比如Django的models.py文件。

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

操作数据库

ORM提供了常用的CRUD操作（Create、Read、Update、Delete）。

创建表

需要为模型创建对应的数据库表。

c.execute('''CREATE TABLE USER
       (ID INT PRIMARY KEY     NOT NULL,
       NAME           TEXT    NOT NULL,
       AGE            INT     NOT NULL);''')

插入数据

可以通过ORM向表中插入数据。

user = User(1, 'Alice', 18)
conn.execute("INSERT INTO USER (ID,NAME,AGE) \
      VALUES (?, ?, ?)", (user.id, user.name, user.age))

查询数据

可以通过属性来查询数据。

cursor = conn.execute("SELECT id, name, age from USER")
for row in cursor:
  user = User(row[0], row[1], row[2])
  print(f"ID = {user.id}, NAME = {user.name}, AGE = {user.age}")

更新数据

可以修改已有的数据。

user.age = 20
conn.execute("UPDATE USER set AGE = ? where ID = ?", (user.age, user.id))

删除数据

可以删除指定的数据。

conn.execute("DELETE from USER where ID = ?", (user.id,))

实现ORM的基本框架

ORM的实现思路：

1. 将每个类定义为python类，类中的属性对应着表中的列
2. 定义一个基础的Model类，所有的类继承这个类
3. 在基础的Model类中为类动态的添加增、删、改、查四个方法

下面我们来实现这个框架。

类的映射

我们可以为每一个表定义对应的类。

class User(Model):
    __table__ = 'USER'

    id = IntegerField('ID', primary_key=True)
    name = StringField('NAME')
    age = IntegerField('AGE')

其中，__table__表示对应的表名，__primary_key__表示主键，IntegerField和StringField是两个常用的属性类型。

创建连接

import sqlite3

conn = sqlite3.connect('test.db')

Field类

我们可以将每个字段抽象成一个类，包括字符型、数值型等不同的类型。

class Field:
    pass

class StringField(Field):
    def __init__(self, name):
        self.name = name

class IntegerField(Field):
    def __init__(self, name):
        self.name = name

Model类 & 增删改查

在基础的Model类中为类动态的添加新增、删除、查询和更新四个方法。

class Model:
    @classmethod
    def create_table(cls):
        fields = []
        for k, v in cls.__dict__.items():
            if isinstance(v, Field) and v.primary_key is True:
                field = k + " " + v.field_type + " PRIMARY KEY"
                fields.append(field)
            elif isinstance(v, Field):
                field = k + " " + v.field_type
                fields.append(field)
        field_str = ", ".join(fields)
        table_name = cls.__table__
        create_sql = f"CREATE TABLE {table_name} ({field_str})"
        conn.execute(create_sql)

    def save(self):
        fields = []
        values = []
        for k, v in self.__dict__.items():
            if isinstance(v, Field) and v.primary_key is not True:
                fields.append(k)
                values.append(v)
        field_str = ", ".join(fields)
        value_str = ", ".join(["?" for _ in range(len(fields))])
        table_name = self.__table__
        insert_sql = f"INSERT INTO {table_name} ({field_str}) VALUES ({value_str})"
        conn.execute(insert_sql, values)
        conn.commit()

    @classmethod
    def get_one(cls, pk):
        table_name = cls.__table__
        pk_field = None
        for k, v in cls.__dict__.items():
            if isinstance(v, Field) and v.primary_key is True:
                pk_field = k
        select_sql = f"SELECT * FROM {table_name} WHERE {pk_field}=?"
        cursor = conn.execute(select_sql, (pk, ))
        row = cursor.fetchone()
        if row is None:
            return None
        values = [row[i] for i in range(len(row))]
        obj = cls()
        for k, v in obj.__dict__.items():
            if isinstance(v, Field) and v.primary_key is True:
                setattr(obj, k, values.pop(0))
            elif isinstance(v, Field):
                setattr(obj, k, values.pop(0))
        return obj

    def update(self):
        fields = []
        values = []
        pk_field = None
        for k, v in self.__dict__.items():
            if isinstance(v, Field) and v.primary_key is not True:
                fields.append(k + "=?")
                values.append(v)
            elif isinstance(v, Field) and v.primary_key is True:
                pk_field = k
        values.append(getattr(self, pk_field))
        field_str = ", ".join(fields)
        table_name = self.__table__
        update_sql = f"UPDATE {table_name} SET {field_str} WHERE {pk_field}=?"
        conn.execute(update_sql, values)
        conn.commit()

    def delete(self):
        pk_field_name = None
        for k, v in self.__dict__.items():
            if isinstance(v, Field) and v.primary_key is True:
                pk_field_name = k
        pk_val = getattr(self, pk_field_name)
        table_name = self.__table__
        delete_sql = f"DELETE FROM {table_name} WHERE {pk_field_name}=?"
        conn.execute(delete_sql, (pk_val, ))
        conn.commit()

在使用的时候，可以通过继承Model类来实现增删改查的功能。

class User(Model):
    __table__ = 'USER'
    id = IntegerField('ID', primary_key=True)
    name = StringField('NAME')
    age = IntegerField('AGE')

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

User.create_table()

user1 = User(1, 'Alice', 18)
user2 = User(2, 'Bob', 20)
user3 = User(3, 'Cindy', 22)

# 插入
user1.save()
user2.save()
user3.save()

# 查询
user = User.get_one(1)
print(f"id: {user.id}, name: {user.name}, age: {user.age}")

# 更新
user.age = 20
user.update()

# 删除
user.delete()

总结

ORM是非常常用的技术，比如在Django、Flask、SQLAlchemy等框架中都有非常完善的ORM。实现一个简单的ORM可以帮助我们更好的理解ORM的底层实现原理，从而更好地应对一些较为复杂的业务场景。

阅读剩余 81%