Day 3 - 编写ORM

在一个Web App中,所有数据,包括用户信息、发布的日志、评论等,都存储在数据库中。在awesome-python3-webapp中,我们选择MySQL作为数据库。

Web App里面有很多地方都要访问数据库。访问数据库需要创建数据库连接、游标对象,然后执行SQL语句,最后处理异常,清理资源。这些访问数据库的代码如果分散到各个函数中,势必无法维护,也不利于代码复用。

所以,我们要首先把常用的SELECT、INSERT、UPDATE和DELETE操作用函数封装起来。

由于Web框架使用了基于asyncio的aiohttp,这是基于协程的异步模型。在协程中,不能调用普通的同步IO操作,因为所有用户都是由一个线程服务的,协程的执行速度必须非常快,才能处理大量用户的请求。而耗时的IO操作不能在协程中以同步的方式调用,否则,等待一个IO操作时,系统无法响应任何其他用户。

这就是异步编程的一个原则:一旦决定使用异步,则系统每一层都必须是异步,“开弓没有回头箭”。

幸运的是aiomysql为MySQL数据库提供了异步IO的驱动。

创建连接池

我们需要创建一个全局的连接池,每个HTTP请求都可以从连接池中直接获取数据库连接。使用连接池的好处是不必频繁地打开和关闭数据库连接,而是能复用就尽量复用。

连接池由全局变量__pool存储,缺省情况下将编码设置为utf8,自动提交事务:

  1. @asyncio.coroutine
  2. def create_pool(loop, **kw):
  3. logging.info('create database connection pool...')
  4. global __pool
  5. __pool = yield from aiomysql.create_pool(
  6. host=kw.get('host', 'localhost'),
  7. port=kw.get('port', 3306),
  8. user=kw['user'],
  9. password=kw['password'],
  10. db=kw['db'],
  11. charset=kw.get('charset', 'utf8'),
  12. autocommit=kw.get('autocommit', True),
  13. maxsize=kw.get('maxsize', 10),
  14. minsize=kw.get('minsize', 1),
  15. loop=loop
  16. )

Select

要执行SELECT语句,我们用select函数执行,需要传入SQL语句和SQL参数:

  1. @asyncio.coroutine
  2. def select(sql, args, size=None):
  3. log(sql, args)
  4. global __pool
  5. with (yield from __pool) as conn:
  6. cur = yield from conn.cursor(aiomysql.DictCursor)
  7. yield from cur.execute(sql.replace('?', '%s'), args or ())
  8. if size:
  9. rs = yield from cur.fetchmany(size)
  10. else:
  11. rs = yield from cur.fetchall()
  12. yield from cur.close()
  13. logging.info('rows returned: %s' % len(rs))
  14. return rs

SQL语句的占位符是?,而MySQL的占位符是%sselect()函数在内部自动替换。注意要始终坚持使用带参数的SQL,而不是自己拼接SQL字符串,这样可以防止SQL注入攻击。

注意到yield from将调用一个子协程(也就是在一个协程中调用另一个协程)并直接获得子协程的返回结果。

如果传入size参数,就通过fetchmany()获取最多指定数量的记录,否则,通过fetchall()获取所有记录。

Insert, Update, Delete

要执行INSERT、UPDATE、DELETE语句,可以定义一个通用的execute()函数,因为这3种SQL的执行都需要相同的参数,以及返回一个整数表示影响的行数:

  1. @asyncio.coroutine
  2. def execute(sql, args):
  3. log(sql)
  4. with (yield from __pool) as conn:
  5. try:
  6. cur = yield from conn.cursor()
  7. yield from cur.execute(sql.replace('?', '%s'), args)
  8. affected = cur.rowcount
  9. yield from cur.close()
  10. except BaseException as e:
  11. raise
  12. return affected

execute()函数和select()函数所不同的是,cursor对象不返回结果集,而是通过rowcount返回结果数。

ORM

有了基本的select()execute()函数,我们就可以开始编写一个简单的ORM了。

设计ORM需要从上层调用者角度来设计。

我们先考虑如何定义一个User对象,然后把数据库表users和它关联起来。

  1. from orm import Model, StringField, IntegerField
  2. class User(Model):
  3. __table__ = 'users'
  4. id = IntegerField(primary_key=True)
  5. name = StringField()

注意到定义在User类中的tableidname是类的属性,不是实例的属性。所以,在类级别上定义的属性用来描述User对象和表的映射关系,而实例属性必须通过init()方法去初始化,所以两者互不干扰:

  1. # 创建实例:
  2. user = User(id=123, name='Michael')
  3. # 存入数据库:
  4. user.insert()
  5. # 查询所有User对象:
  6. users = User.findAll()

定义Model

首先要定义的是所有ORM映射的基类Model

  1. class Model(dict, metaclass=ModelMetaclass):
  2. def __init__(self, **kw):
  3. super(Model, self).__init__(**kw)
  4. def __getattr__(self, key):
  5. try:
  6. return self[key]
  7. except KeyError:
  8. raise AttributeError(r"'Model' object has no attribute '%s'" % key)
  9. def __setattr__(self, key, value):
  10. self[key] = value
  11. def getValue(self, key):
  12. return getattr(self, key, None)
  13. def getValueOrDefault(self, key):
  14. value = getattr(self, key, None)
  15. if value is None:
  16. field = self.__mappings__[key]
  17. if field.default is not None:
  18. value = field.default() if callable(field.default) else field.default
  19. logging.debug('using default value for %s: %s' % (key, str(value)))
  20. setattr(self, key, value)
  21. return value

Modeldict继承,所以具备所有dict的功能,同时又实现了特殊方法getattr()setattr(),因此又可以像引用普通字段那样写:

  1. >>> user['id']
  2. 123
  3. >>> user.id
  4. 123

以及Field和各种Field子类:

  1. class Field(object):
  2. def __init__(self, name, column_type, primary_key, default):
  3. self.name = name
  4. self.column_type = column_type
  5. self.primary_key = primary_key
  6. self.default = default
  7. def __str__(self):
  8. return '<%s, %s:%s>' % (self.__class__.__name__, self.column_type, self.name)

映射varcharStringField

  1. class StringField(Field):
  2. def __init__(self, name=None, primary_key=False, default=None, ddl='varchar(100)'):
  3. super().__init__(name, ddl, primary_key, default)

注意到Model只是一个基类,如何将具体的子类如User的映射信息读取出来呢?答案就是通过metaclass:ModelMetaclass

  1. class ModelMetaclass(type):
  2. def __new__(cls, name, bases, attrs):
  3. # 排除Model类本身:
  4. if name=='Model':
  5. return type.__new__(cls, name, bases, attrs)
  6. # 获取table名称:
  7. tableName = attrs.get('__table__', None) or name
  8. logging.info('found model: %s (table: %s)' % (name, tableName))
  9. # 获取所有的Field和主键名:
  10. mappings = dict()
  11. fields = []
  12. primaryKey = None
  13. for k, v in attrs.items():
  14. if isinstance(v, Field):
  15. logging.info(' found mapping: %s ==> %s' % (k, v))
  16. mappings[k] = v
  17. if v.primary_key:
  18. # 找到主键:
  19. if primaryKey:
  20. raise RuntimeError('Duplicate primary key for field: %s' % k)
  21. primaryKey = k
  22. else:
  23. fields.append(k)
  24. if not primaryKey:
  25. raise RuntimeError('Primary key not found.')
  26. for k in mappings.keys():
  27. attrs.pop(k)
  28. escaped_fields = list(map(lambda f: '`%s`' % f, fields))
  29. attrs['__mappings__'] = mappings # 保存属性和列的映射关系
  30. attrs['__table__'] = tableName
  31. attrs['__primary_key__'] = primaryKey # 主键属性名
  32. attrs['__fields__'] = fields # 除主键外的属性名
  33. # 构造默认的SELECT, INSERT, UPDATE和DELETE语句:
  34. attrs['__select__'] = 'select `%s`, %s from `%s`' % (primaryKey, ', '.join(escaped_fields), tableName)
  35. attrs['__insert__'] = 'insert into `%s` (%s, `%s`) values (%s)' % (tableName, ', '.join(escaped_fields), primaryKey, create_args_string(len(escaped_fields) + 1))
  36. attrs['__update__'] = 'update `%s` set %s where `%s`=?' % (tableName, ', '.join(map(lambda f: '`%s`=?' % (mappings.get(f).name or f), fields)), primaryKey)
  37. attrs['__delete__'] = 'delete from `%s` where `%s`=?' % (tableName, primaryKey)
  38. return type.__new__(cls, name, bases, attrs)

这样,任何继承自Model的类(比如User),会自动通过ModelMetaclass扫描映射关系,并存储到自身的类属性如tablemappings中。

然后,我们往Model类添加class方法,就可以让所有子类调用class方法:

  1. class Model(dict):
  2. ...
  3. @classmethod
  4. @asyncio.coroutine
  5. def find(cls, pk):
  6. ' find object by primary key. '
  7. rs = yield from select('%s where `%s`=?' % (cls.__select__, cls.__primary_key__), [pk], 1)
  8. if len(rs) == 0:
  9. return None
  10. return cls(**rs[0])

User类现在就可以通过类方法实现主键查找:

  1. user = yield from User.find('123')

往Model类添加实例方法,就可以让所有子类调用实例方法:

  1. class Model(dict):
  2. ...
  3. @asyncio.coroutine
  4. def save(self):
  5. args = list(map(self.getValueOrDefault, self.__fields__))
  6. args.append(self.getValueOrDefault(self.__primary_key__))
  7. rows = yield from execute(self.__insert__, args)
  8. if rows != 1:
  9. logging.warn('failed to insert record: affected rows: %s' % rows)

这样,就可以把一个User实例存入数据库:

  1. user = User(id=123, name='Michael')
  2. yield from user.save()

最后一步是完善ORM,对于查找,我们可以实现以下方法:

  • findAll() - 根据WHERE条件查找;

  • findNumber() - 根据WHERE条件查找,但返回的是整数,适用于select count(*)类型的SQL。

以及update()remove()方法。

所有这些方法都必须用@asyncio.coroutine装饰,变成一个协程。

调用时需要特别注意:

  1. user.save()

没有任何效果,因为调用save()仅仅是创建了一个协程,并没有执行它。一定要用:

  1. yield from user.save()

才真正执行了INSERT操作。

最后看看我们实现的ORM模块一共多少行代码?累计不到300多行。用Python写一个ORM是不是很容易呢?

参考源码

day-03

原文: https://wizardforcel.gitbooks.io/liaoxuefeng/content/py3/109.html