创立ORM--廖雪峰python笔记

写在前面

本文是根据廖雪峰Day 3 - 编写ORM实践后整理的学习笔记,主要记录实践过程中遇到的问题,以及对其所涉及到的知识进行提炼与补充。

必备知识

mysql数据库

数据库(database)即按照数据结构来组织、存储与管理数据的仓库。
mysql数据库是一种关系型数据库(RDBMS),建立在关系模型基础上的数据库,其特点是:

  • 数据以表(table)的形式存储在数据库中(database)
  • 表的每列为记录名称所对应的数据域【可以理解为划分类】
  • 表的每行为记录名称【可以理解为具体属性】

一种常见的数据表
| name | score | place |
| —– | —– | —– |
| Mike | 100 | China |
| Jane | 88 | US |
| Mille | 68 | UK |

RDBMS术语

(这里只罗列一些本文可能涉及到的)

  • 主键(key):主键是唯一的,用来查询数据。即主键的属性值能够唯一定位到一组数据【类似书籍的页码】。
  • 索引:索引可以不唯一,使用索引可以快速访问数据库表中的特定信息【类似书籍的目录】

    注意

    1
    2
    1.本文所用到的mysql数据库可以去官网下载,但是该数据库只支持python 3.4版本,若要通过python连接数据库,需要下载pymysql模块。
    2.本文需要使用到异步aiomysql模块,该模块可能与pymysql模块存在版本不兼容问题。妥善处理方式是,更新aiomysql版本为0.0.7,pymysql版本为0.6.7

sql语法

篇幅有限,这里只解释本项目用到的schema.sql相关语句,具体语法可参考w3school SQL教程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
--schema.sql

--如果存在awesome数据库,则删除该数据库(drop)
drop database if exists awesome;

--创建awesome数据库(create database)
create database awesome;

--选择awesome数据库(use)
use awesome;

--分配权限给特定用户(grant 权限 on 数据库名.表名 to 用户名@登陆方式 identified by 'password')
grant select, insert, update, delete on awesome.* to 'www-data'@'localhost' identified by 'www-data';

--创建users表并设置具体列属性
create table users (
`id` varchar(50) not null,
`email` varchar(50) not null,
`passwd` varchar(50) not null,
`admin` bool not null,
`name` varchar(50) not null,
`image` varchar(500) not null,
`created_at` real not null,

key `idx_created_at` (`created_at`),
primary key (`id`)
) engine=innodb default charset=utf8;
--括号内最后两句分别为设置主键,设置索引

执行下列命令即可在mysql数据库中创建相应的数据表。

1
$ mysql -u root -q < shemal.sql

ORM

说了这么多,这一节是要干嘛?ORM又是什么玩意?

ORM即Object Relational Mapping,全称对象关系映射
当我们需要对数据库进行操作时,势必需要通过连接数据、调用sql语句、执行sql语句等操作,ORM将数据库中的表,字段,行与我们面向对象编程的类及其方法,属性等一一对应,即将该部分操作封装起来,程序猿不需懂得sql语句即可完成对数据库的操作。

编写ORM模块

创建连接池

利用’aiomysql.create_pool()`创建协程连接池连接数据库。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
async def create_pool(loop, **kw):
'''创建连接池

'''
logging.info('create database connection pool...')
global __pool
__pool = await aiomysql.create_pool(
host=kw.get('host', 'localhost'),
port=kw.get('port', 3306),
user=kw['user'],
password=kw['password'],
db=kw['database'],
charset=kw.get('charset', 'utf8'),
autocommit=kw.get('autocommit', True),
maxsize=kw.get('maxsize', 10),
minsize=kw.get('minsize', 1),
loop=loop
)

注意

1
数据库的连接打开后一定要及时关闭

关闭连接池

1
2
3
4
5
6
async def close_pool():
'''异步关闭连接池'''
logging.info('close database connection pool...')
global __pool
__pool.close()
await __pool.wait_closed()

封装数据库操作指令

数据库操作主要包括select, execute(update, insert, delete)。除了select需要返回查询内容,其他命令只需返回一个影响行数,故可以封装为一个execute方法。

select

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
async def select(sql, args, size=None):
'''此处为选取数据库相关数据操作

'''
log(sql, args)
global __pool
async with __pool.get() as conn:#从连接池获取一个connect
async with conn.cursor(aiomysql.DictCursor) as cur:#获取游标cursor
await cur.execute(sql.replace('?', '%s'), args or ())#将输入的sql语句中的'?'替换为具体参数args
if size:
rs = await cur.fetchmany(size)
else:
rs = await cur.fetchall()
logging.info('rows returned: %s' % len(rs))
return rs

execute(update, insert, delete)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
async def execute(sql, args, autocommit=True):
'''此处执行数据库删减、增添等修改该操作

'''
log(sql)
async with __pool.get() as conn:
if not autocommit:
await conn.begin()
try:
async with conn.cursor(aiomysql.DictCursor) as cur:
await cur.execute(sql.replace('?', '%s'), args)
affected = cur.rowcount
if not autocommit:
await conn.commit()
except BaseException as e:
if not autocommit:
await conn.rollback()
raise
return affected#返回修改行

自顶向下的设计方案

当没有思路时,设想如果有一个现成的ORM框架,该去如何使用呢?

1
2
3
4
5
6
7
8
9
10
11
12
class Model(object):
async def save(self, **kw):
pass

class User(Model):
__table__ = 'users' #设定操作数据库表
name = StringField(...) #设定列属性
score = FloatField(...) #设定列属性
pass

u = User(name='Mike', score=98.23)
u.save()

也就说当操作某数据库的一个数据库表时,只需创立一个类,并利用继承的方式,从父类中继承相关属性与方法,这样就可以直接完成对数据库的操作。

字段类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
class Field(object):
'''用于标识model每个成员变量的类

name:表名称, column_type:值类型, primary_key:是否主键'''
def __init__(self, name, column_type, primary_key, default):
self.name = name
self.column_type = column_type
self.primary_key = primary_key
self.default = default

def __str__(self):
return '<%s, %s:%s>' % (self.__class__.__name__, self.column_type, self.name)

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

class IntegerField(Field):
pass

...

Model类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
class ModelMetaclass(type):
def __new__(cls, name, bases, attrs):
if name == 'Model':
return type.__new__(cls, name, bases, attrs)
tableName = attrs.get('__table__', None) or name
logging.info('found model: %s (table: %s)' %(name, tableName))
mappings = dict()
fields = []#可以理解为列名称
primaryKey = None
for k, v in attrs.items():
if isinstance(v, Field):
logging.info(' found mapping: %s ==> %s' %(k, v))
mappings[k] = v
if v.primary_key:#判断主键并记录
if primaryKey:
raise RuntimeError('Duplicate primary key for field: %s' %k)
primaryKey = k#记录主键
else:
fields.append(k)
if not primaryKey:
raise RuntimeError('Primary key not found.')
for k in mappings.keys():
attrs.pop(k)#删除attrs里属性,防止与实例属性冲突
escaped_fields = list(map(lambda f: ' %s ' %f, fields))
attrs['__mappings__'] = mappings # 保存属性和列的映射关系
attrs['__table__'] = tableName
attrs['__primary_key__'] = primaryKey # 主键属性名
attrs['__fields__'] = fields # 除主键外的属性名
attrs['__select__'] = 'select `%s`, %s from `%s`' % (primaryKey, ', '.join(escaped_fields), tableName)
attrs['__insert__'] = 'insert into `%s` (%s, `%s`) values (%s)' % (
tableName, ', '.join(escaped_fields), primaryKey, create_args_string(len(escaped_fields) + 1))
attrs['__update__'] = 'update `%s` set %s where `%s`=?' % (
tableName, ', '.join(map(lambda f: '`%s`=?' % (mappings.get(f).name or f), fields)), primaryKey)
attrs['__delete__'] = 'delete from `%s` where `%s`=?' % (tableName, primaryKey)
return type.__new__(cls, name, bases, attrs)

#------------------------------------------------------------------
class Model(dict, metaclass=ModelMetaclass):
def __init__(self, **kw):
super(Model, self).__init__(**kw)

def __getattr__(self, key):
try:
return self[key]
except KeyError:
raise AttributeError(r"'Model' object has no attribute '%s'" % key)

def __setattr__(self, key, value):
self[key] = value

def getValue(self, key):
return getattr(self, key, None)

def getValueOrDefault(self, key):
value = getattr(self, key, None)
if value is None:
field = self.__mappings__[key]
if field.default is not None:
value = field.default() if callable(field.default) else field.default#callable(obj)判断对象是否可调用
logging.debug('using default value for %s: %s' %(key, str(value)))
setattr(self, key, value)
return value

@classmethod
async def findAll(cls, where=None, args=None, **kw):
'find objects by where clause'
sql = [cls.__select__]
if where:
sql.append('where')
sql.append(where)
if args is None:
args = []
orderBy = kw.get('orderBy', None)
if orderBy:
sql.append('order by')
sql.append(orderBy)
limit = kw.get('limit', None)
if limit is not None:
sql.append('limit')
if isinstance(limit, int):
sql.append('?')
args.append(limit)
elif isinstance(limit, tuple) and len(limit) == 2:
sql.append('?, ?')
args.extend(limit)
else:
raise ValueError('Invalid limit value: %s' % str(limit))
rs = await select(' '.join(sql), args)
return [cls(**r) for r in rs]

@classmethod
async def findNumber(cls, selectField, where=None, args=None):
'find number by select and where'
sql = ['select %s _num_ from `%s`' %(selectField, cls.__table__)]
if where:
sql.append('where')
sql.append(where)
rs = await select(' '.join(sql), args, 1)
if len(rs) == 0:
return None
return rs[0]['_num_']

@classmethod
async def find(cls, pk):
'find object by primary key'
rs = await select('%s where `%s`=?' %(cls.__select__, cls.__primary_key__), [pk], 1)
if len(rs) == 0:
return None
return cls(**rs[0])

async def save(self):
args = list(map(self.getValueOrDefault, self.__fields__))
args.append(self.getValueOrDefault(self.__primary_key__))
rows = await execute(self.__insert__, args)
if rows != 1:
logging.warn('failed to insert record: affected rows: %s' % rows)

async def update(self):
args = list(map(self.getValue, self.__fields__))
args.append(self.getValue(self.__primary_key__))
rows = await execute(self.__update__, args)
if rows != 1:
logging.warn('failed to update by primary key: affected rows: %s' %rows)

async def remove(self):
args = [self.getValue(self.__primary_key__)]
rows = await execute(self.__delete__, args)
if rows != 1:
logging.warn('failed to remove by primary key: affected rows: %s' % rows)

关于元类

暂时没有理解透彻,之后再做补充。

本文标题:创立ORM--廖雪峰python笔记

文章作者:Lumo Wang

发布时间:2018年03月09日 - 11:03

最后更新:2018年11月07日 - 15:11

原始链接:https://luameows.github.io/2018/03/09/学习笔记--廖雪峰python笔记-创立ORM/

许可协议: 署名-非商业性使用-禁止演绎 4.0 国际 转载请保留原文链接及作者。

请我喝杯咖啡,我会继续熬夜的~