金环建设集团有限公司官方网站,网络营销做得好的公司,无锡培训网站建设,平面广告设计公司在一个Web App中#xff0c;所有数据#xff0c;包括用户信息、发布的日志、评论等#xff0c;都存储在数据库中。在awesome-python-app中#xff0c;我们选择MySQL作为数据库。
Web App里面有很多地方都要访问数据库。访问数据库需要创建数据库连接、游标对象#xff0c;…在一个Web App中所有数据包括用户信息、发布的日志、评论等都存储在数据库中。在awesome-python-app中我们选择MySQL作为数据库。
Web App里面有很多地方都要访问数据库。访问数据库需要创建数据库连接、游标对象然后执行SQL语句最后处理异常清理资源。这些访问数据库的代码如果分散到各个函数中势必无法维护也不利于代码复用。
此外在一个Web App中有多个用户会同时访问系统以多进程或多线程模式来处理每个用户的请求。假设以多线程为例每个线程在访问数据库时都必须创建仅属于自身的连接对别的线程不可见否则就会造成数据库操作混乱。
所以我们还要创建一个简单可靠的数据库访问模型在一个线程中能既安全又简单地操作数据库。
为什么不选择SQLAlchemySQLAlchemy太庞大过度地面向对象设计导致API太复杂。
所以我们决定自己设计一个封装基本的SELECT、INSERT、UPDATE和DELETE操作的db模块transwarp.db。
设计db接口
设计底层模块的原则是根据上层调用者设计简单易用的API接口然后实现模块内部代码。
假设transwarp.db模块已经编写完毕我们希望以这样的方式来调用它
首先初始化数据库连接信息通过create_engine()函数
1
2
from transwarpimport db
db.create_engine(userroot, passwordpassword, databasetest, host127.0.0.1, port3306)
然后就可以直接操作SQL了。
如果需要做一个查询可以直接调用select()方法返回的是list每一个元素是用dict表示的对应的行
1
2
3
4
5
6
7
users db.select(select * from user)
# users
# [
# { id: 1, name: Michael},
# { id: 2, name: Bob},
# { id: 3, name: Adam}
# ]
如果要执行INSERT、UPDATE或DELETE操作执行update()方法返回受影响的行数
1
n db.update(insert into user(id, name) values(?, ?),4,Jack)
update()函数签名为
1
update(sql,*args)
统一用?作为占位符并传入可变参数来绑定从根本上避免SQL注入攻击。
每个select()或update()调用都隐含地自动打开并关闭了数据库连接这样上层调用者就完全不必关心数据库底层连接。
但是如果要在一个数据库连接里执行多个SQL语句怎么办我们用一个with语句实现
1
2
3
4
with db.connection():
db.select(...)
db.update(...)
db.update(...)
如果要在一个数据库事务中执行多个SQL语句怎么办我们还是用一个with语句实现
1
2
3
4
with db.transaction():
db.select(...)
db.update(...)
db.update(...)
实现db模块
由于模块是全局对象模块变量是全局唯一变量所以有两个重要的模块变量
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
# db.py
# 数据库引擎对象:
class _Engine(object):
def __init__(self, connect):
self._connect connect
def connect(self):
return self._connect()
engine None
# 持有数据库连接的上下文对象:
class _DbCtx(threading.local):
def __init__(self):
self.connection None
self.transactions 0
def is_init(self):
return not self.connectionis None
def init(self):
self.connection _LasyConnection()
self.transactions 0
def cleanup(self):
self.connection.cleanup()
self.connection None
def cursor(self):
return self.connection.cursor()
_db_ctx _DbCtx()
由于_db_ctx是threadlocal对象所以它持有的数据库连接对于每个线程看到的都是不一样的。任何一个线程都无法访问到其他线程持有的数据库连接。
有了这两个全局变量我们继续实现数据库连接的上下文目的是自动获取和释放连接
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
class _ConnectionCtx(object):
def __enter__(self):
global _db_ctx
self.should_cleanup False
if not _db_ctx.is_init():
_db_ctx.init()
self.should_cleanup True
return self
def __exit__(self, exctype, excvalue, traceback):
global _db_ctx
if self.should_cleanup:
_db_ctx.cleanup()
def connection():
return _ConnectionCtx()
定义了__enter__()和__exit__()的对象可以用于with语句确保任何情况下__exit__()方法可以被调用。
把_ConnectionCtx的作用域作用到一个函数调用上可以这么写
1
2
with connection():
do_some_db_operation()
但是更简单的写法是写个decorator
1
2
3
with_connection
def do_some_db_operation():
pass
这样我们实现select()、update()方法就更简单了
1
2
3
4
5
6
7
with_connection
def select(sql,*args):
pass
with_connection
def update(sql,*args):
pass
注意到Connection对象是存储在_DbCtx这个threadlocal对象里的因此嵌套使用with connection()也没有问题。_DbCtx永远检测当前是否已存在Connection如果存在直接使用如果不存在则打开一个新的Connection。
对于transaction也是类似的with transaction()定义了一个数据库事务
1
2
3
4
with db.transaction():
db.select(...)
db.update(...)
db.update(...)
函数作用域的事务也有一个简化的decorator
1
2
3
with_transaction
def do_in_transaction():
pass
事务也可以嵌套内层事务会自动合并到外层事务中这种事务模型足够满足99%的需求。
事务嵌套比Connection嵌套复杂一点因为事务嵌套需要计数每遇到一层嵌套就1离开一层嵌套就-1最后到0时提交事务
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
class _TransactionCtx(object):
def __enter__(self):
global _db_ctx
self.should_close_conn False
if not _db_ctx.is_init():
_db_ctx.init()
self.should_close_conn True
_db_ctx.transactions _db_ctx.transactions 1
return self
def __exit__(self, exctype, excvalue, traceback):
global _db_ctx
_db_ctx.transactions _db_ctx.transactions- 1
try:
if _db_ctx.transactions0:
if exctypeis None:
self.commit()
else:
self.rollback()
finally:
if self.should_close_conn:
_db_ctx.cleanup()
def commit(self):
global _db_ctx
try:
_db_ctx.connection.commit()
except:
_db_ctx.connection.rollback()
raise
def rollback(self):
global _db_ctx
_db_ctx.connection.rollback()
最后把select()和update()方法实现了db模块就完成了。
