首页 > 代码库 > 关于db访问层的封装设计感想 dbpy项目的开发

关于db访问层的封装设计感想 dbpy项目的开发

2024-10-31 09:11:02 207人阅读

dbpy

dbpy是一个python写的数据库CURD人性化api库。借鉴了 webpy db 和 drupal database 的设计。如果喜欢 tornado db 或者 webpy db这类轻巧的db库，或者想发挥原生SQL优势，那么值得一试。

Featues

灵活简单
天马行空的SQL构建语法糖
线程安全的连接池
支持读写分离（当前限定只能是一主多副模式）
支持简单事务

Contents

Featues
Install
Development
Compatibility
DB API
- setup
- query
- execute
- select
  - select all
  - specific columns
  - execute
  - Condition
    - simple
    - in
    - between
    - multi condition
    - or condition
  - order by
  - distinct
  - group by
  - limit and offset
  - null condition
- complex conditions
- expr
- insert
- update
  - limit
- delete
- to_sql and str
- transaction
- simple orm
- Future
LICENSE

Install

从github上fork下来，终端执行下面命令:

cd dbpy # the path to the projectpython setup.py install

Note

安装前先安装 MySQLdb (MySQL-python) 依赖python库

Development

下载后终端执行:

cd dbpy # the path to the projectpython setup.py develop

Compatibility

在 Python 2.7.x 测试开发

DB API

先提醒下模块使用单例模式。所以api相对比较好使。

config = {        ‘passwd‘: ‘test‘,        ‘user‘: ‘test‘,        ‘host‘: ‘localhost‘,        ‘db‘: ‘test‘,        ‘max_idle‘ : 5*60    }db.setup(config，  minconn=5, maxconn=10,    adapter=‘mysql‘, key=‘defalut‘, slave=False)

setup

config:	是数据库连接参数，可以传入MySQLDB#connect接口中所有的可选参数。其中``max_idle`` 相对是mysql服务端 connect_timeout配置,默认10秒。
minconn:	为当前数据库连接池保持最小连接池，默认为5
maxconn:	为当前数据库连接池最大连接池，默认为10
adapter:	为适配器名，当前只支持 mysql
key:	是数据库的标识符,默认为 default
slave:	如果为true那么当前的数据库将会注册为读数据库。如果你没有做读写分离，只有一个数据库用来读写，那么setup一次就好，这样就可以读写。

config = {        ‘passwd‘: ‘test‘,        ‘user‘: ‘test‘,        ‘host‘: ‘localhost‘,        ‘db‘: ‘test‘,        ‘max_idle‘ : 5*60    }db.setup(config， key=‘test‘)config[‘host‘] = ‘test.slave‘# 这次setup将会把key标记为仅可写，就是在后面用api时，制定到当前key的数据库会做数据分离db.setup(config， key=‘test‘, slave=True)config[‘host‘] = ‘test.slave2‘# 再加入一个slave数据库db.setup(config， key=‘test‘, slave=True)config[‘host‘] = ‘host2‘config[‘db‘] = ‘social‘# 再加入一个数据库db.setup(config， key=‘social‘, slave=True)

query

query用于raw sql的查询语言。如果有更新数据请用execute.

query(sql, args=None, many=None, as_dict=False, key=‘default‘):

sql:	mysql的格式化raw sql
args:	可以为元组和list，是sql格式化预处理的输入
many:	如果指定为大于零的整数将会使用fetchmany语句，并返回对象将会是迭代器.否则api调用fetchall返回结果.
as_dict:	如果为 true将会返回字典行，否则返回元组行。
key:	用于指定使用那个数据库。

print db.query(‘SELECT 1‘)# > ((1L,),)# use social dbprint db.query(‘SELECT 1‘, key=‘social‘)# > ((1L,),)print db.query(‘SELECT * FROM users WHERE uid=%s and name=%s‘, (1, ‘user_1‘))# > ((1L, u‘user_1‘),)# Wanna return dict rowprint db.query(‘SELECT * FROM users WHERE uid=%s and name=%s‘,            (1, ‘user_1‘), as_dict=True)# > ({‘uid‘: 1L, ‘name‘: u‘user_1‘},)# Use fetchmany(many) then yeild, Return generatorres = db.query(‘SELECT * FROM users WHERE uid=%s and name=%s‘,                (1, ‘user_1‘), many=5, as_dict=True)print resprint res.next()# > <generator object _yield at 0x7f818f4b6820># > {‘uid‘: 1L, ‘name‘: u‘user_1‘}

execute

execute用于raw sql的更新语言。 execute(sql, args=None, key=‘default‘):

sql:	mysql的格式化raw sql
args:	可以为元组和list，是sql格式化预处理的输入.如下面例子insert语句values有多个插入时，调用 `executemany`
key:	用于指定使用那个数据库。

返回规范:

对于insert 将会返回 last_insert_id, 其他更新语句返回rowcount

db.execute(‘DROP TABLE IF EXISTS `users`‘)db.execute("""CREATE TABLE `users` (         `uid` int(10) unsigned NOT NULL AUTO_INCREMENT,        `name` varchar(20) NOT NULL,        PRIMARY KEY (`uid`))""")# insert语句插入多个value，注意这样写将会调用executemany，你懂的，就是封装了多条execute的玩意db.execute(‘INSERT INTO users VALUES(%s, %s)‘, [(10, ‘execute_test‘), (9, ‘execute_test‘)])# > 9db.execute(‘DELETE FROM users WHERE name=%s‘, (‘execute_test‘,))# > 2# use social dbdb.execute(‘delete from events where created_at<%s‘, (expired, ), key=‘social‘)# > 10

select

select用于构建select 查询语言。

select(table, key=‘default‘):

table:	选定表
key:	用于指定使用那个数据库。

select all

db.select(‘users‘)# > SELECT * FROM `users`

specific columns

db.select(‘users‘).fields(‘uid‘, ‘name‘)# > SELECT `uid`, `name` FROM `users`

execute

在构建好查询条语句后使用execute api可以返回结果。

execute(many=None, as_dict=False):

many:	如果指定为大于零的整数将会使用fetchmany语句，并返回对象将会是迭代器.否则api调用fetchall返回结果.
as_dict:	如果为 true将会返回字典行，否则返回元组行。

q = db.select(‘users‘).fields(‘uid‘, ‘name‘)res = q.execute()print res# > ((1L, u‘user_1‘), (2L, u‘user_2‘), (3L, u‘user_3‘), (4L, u‘user_4‘), (5L, None))res = q.execute(many=2, as_dict=True)print resprint res.next()# > <generator object _yield at 0x7f835825e820># > {‘uid‘: 1L, ‘name‘: u‘user_1‘}

Condition

上面已经学会如何做简单的查询，那么如何组件条件查询。这里将会重点讲述condition方法如何构建各种查询条件。

condition(field, value=http://www.mamicode.com/None, operator=None):

field:	是条件限制的表字段
value:	是字段的条件值，如果炸路额， oprator都不指定就是 "field is null"
operator:	默认可能是等于操作符号，可选的操作符号有 BETWEEN, IN, NOT IN, EXISTS, NOT EXISTS, IS NULL, IS NOT NULL, LIKE, NOT LIKE, =, <, >, >=, <=, <>等

在所有的select，update, delete查询中多个默认的condition将会是and条件组合。

simple

db.select(‘users‘).condition(‘uid‘, 1) # condition(‘uid‘, 1, ‘=‘)# > SELECT * FROM `users`# > WHERE  `uid` = %s

in

db.select(‘users‘).condition(‘uid‘, (1, 3)) # condition(‘uid‘, [1, 3]) 一样# > SELECT * FROM `users`# > WHERE  `uid` IN  (%s, %s)

between

db.select(‘users‘).condition(‘uid‘, (1, 3), ‘between‘)# > SELECT * FROM `users`# > WHERE  `uid` BETWEEN %s AND %s

multi condition

db.select(‘users‘).condition(‘uid‘, (1, 3), ‘between‘)# > SELECT * FROM `users`# > WHERE  `uid` BETWEEN %s AND %s

or condition

or_cond = db.or_().condition(‘uid‘, 1).condition(‘name‘, ‘blabla‘)db.select(‘users‘).condition(or_cond).condition(‘uid‘, 1, ‘<>‘)# > SELECT * FROM `users`# > WHERE  ( `uid` = %s OR `name` = %s ) AND `uid` <> %s

order by

db.select(‘users‘).order_by(‘name‘)# > SELECT * FROM `users`# > ORDER BY `name`db.select(‘users‘).order_by(‘name‘, ‘DESC‘)# > SELECT * FROM `users`# > ORDER BY `name` DESCdb.select(‘users‘).order_by(‘name‘, ‘DESC‘).order_by(‘uid‘)# > SELECT * FROM `users`# > ORDER BY `name` DESC, `uid`

distinct

db.select(‘users‘).order_by(‘name‘)# > SELECT * FROM `users`# > ORDER BY `name`db.select(‘users‘).order_by(‘name‘, ‘DESC‘)# > SELECT * FROM `users`# > ORDER BY `name` DESCdb.select(‘users‘).order_by(‘name‘, ‘DESC‘).order_by(‘uid‘)# > SELECT * FROM `users`# > ORDER BY `name` DESC, `uid`

group by

db.select(‘users‘).group_by(‘name‘, ‘uid‘)# > SELECT * FROM `users`# > GROUP BY `name`, `uid`

limit and offset

db.select(‘users‘).limit(2).offset(5)# > SELECT * FROM `users`# > LIMIT 2 OFFSET 5

null condition

db.select(‘users‘).is_null(‘name‘).condition(‘uid‘, 5)# > SELECT * FROM `users`# > WHERE  `name` IS NULL  AND `uid` = %sdb.select(‘users‘).is_not_null(‘name‘).condition(‘uid‘, 5)# > SELECT * FROM `users`# > WHERE  `name` IS NOT NULL  AND `uid` = %sdb.select(‘users‘).condition(‘name‘, None)# > SELECT * FROM `users`# > WHERE  `name` IS NULL

complex conditions

使用 db.and_(), db.or_() 可以构建and或or粘合的条件组合。

or_cond = db.or_().condition(‘field1‘, 1).condition(‘field2‘, ‘blabla‘)and_cond = db.and_().condition(‘field3‘, ‘what‘).condition(‘field4‘, ‘then?‘)print db.select(‘table_name‘).condition(or_cond).condition(and_cond)# > SELECT * FROM `table_name`# > WHERE  ( `field1` = %s OR `field2` = %s ) AND ( `field3` = %s AND `field4` = %s )

expr

如果你需要使用 count sum之类的集聚函数，那么使用 Expr构建字段吧。

from  db import exprdb.select(‘users‘).fields(expr(‘count(*)‘))# > SELECT count(*) FROM `users`db.select(‘users‘).fields(expr(‘count(uid)‘, ‘total‘))# > SELECT count(uid) AS `total` FROM `users`

insert

insert用于构建insert into的sql语句。

insert(table, key=‘default‘):

table:	选定表
key:	用于指定使用那个数据库。

q = db.insert(‘users‘).values((10, ‘test_insert‘))# > INSERT INTO `users` VALUES(%s, %s)print q._values# > [(10, ‘test_insert‘)]q = db.insert(‘users‘).fields(‘name‘).values({‘name‘: ‘insert_1‘}).values((‘insert_2‘,))# > INSERT INTO `users` (`name`) VALUES(%s)print q._values# > [(‘insert_1‘,), (‘insert_2‘,)]

构建好执行execute会执行数据库插入,execute返回的是last insert id：

print q.execute()# > 2

update

update用于构建update的sql语句

update(table, key=‘default‘):

table:	选定表
key:	用于指定使用那个数据库。

update 主要可用的方法是mset和set， mset：

mset:	传入的是字典，用于一次set多个表属性
set(column, value):	只能设置一个属性，可以多次使用

构建条件codition前面已经讲述了。请参考 select

db.update(‘users‘).mset({‘name‘:None, ‘uid‘ : 12}).condition(‘name‘,‘user_1‘)# > UPDATE `users` SET `name` = %s, `uid` = %s WHERE  `name` = %sq = (db.update(‘users‘).set(‘name‘, ‘update_test‘).set(‘uid‘, 12)    .condition(‘name‘, ‘user_2‘).condition(‘uid‘, 2)) # .execute()print q.to_sql()# > UPDATE `users` SET `name` = %s, `uid` = %s WHERE  `name` = %s AND `uid` = %s

构建好执行execute会执行数据库插入,execute返回的是更新的 rowcount：

print q.execute()# > 2

limit

因为你可能希望限制更新几条。那么可以使用limit

db.update(‘users‘).mset({‘name‘:None, ‘uid‘ : 12}).condition(‘name‘,‘user_1‘).limit(5)# > UPDATE `users` SET `name` = %s, `uid` = %s WHERE  `name` = %s  LIMIT 5

delete

delete 用于构建delete from的sql语句。

delete(table, key=‘default‘):

table:	选定表
key:	用于指定使用那个数据库。

构建条件codition前面已经讲述了。请参考 select

db.delete(‘users‘).condition(‘name‘,‘user_1‘)# > DELETE FROM `users` WHERE  `name` = %s

构建好执行execute会执行数据库插入,execute返回的是删除的 rowcount：

print q.execute()# > 2

to_sql and str

db.insert, db.update, db.delete 返回的对象都可以使用 to_sql 或者__str__ 来查看构建成的sql语句。

q = (db.update(‘users‘).set(‘name‘, ‘update_test‘).set(‘uid‘, 12)        .condition(‘name‘, ‘user_2‘).condition(‘uid‘, 2))print q.to_sql()print q# > UPDATE `users` SET `name` = %s, `uid` = %s WHERE  `name` = %s AND `uid` = %s

transaction

transaction(table, key=‘default‘):

table:	选定表
key:	用于指定使用那个数据库。

对于事务，这里比较简单的实现。要么全部执行，要么全部不做，没有做保存点。

# with contextwith db.transaction() as t:    t.delete(‘users‘).condition(‘uid‘, 1).execute()    (t.update(‘users‘).mset({‘name‘:None, ‘uid‘ : 12})        .condition(‘name‘,‘user_1‘).execute())# 普通用法t = db.transaction()t.begin()t.delete(‘users‘).condition(‘uid‘, 1).execute()(t.update(‘users‘).mset({‘name‘:None, ‘uid‘ : 12})    .condition(‘name‘,‘user_1‘).execute())#这里将会提交，如果失败将会rollbackt.commit()

Note

使用 begin一定要结合commit方法，不然可能连接不会返还连接池。建议用 with 语句。

simple orm

这里将会讲述最简单的orm构建技巧，详细参考 samples

import modelfrom orm import Backendimport dbdb.setup({ ‘host‘: ‘localhost‘, ‘user‘: ‘test‘, ‘passwd‘: ‘test‘, ‘db‘: ‘blog‘})user = Backend(‘user‘).find_by_username(‘username‘)if user and user.check(‘password‘):    print ‘auth‘user = model.User(‘username‘, ‘email‘, ‘real_name‘, ‘password‘,        ‘bio‘, ‘status‘, ‘role‘)if Backend(‘user‘).create(user):    print ‘fine‘user = Backend(‘user‘).find(12)user.real_name = ‘blablabla....‘if Backend(‘user‘).save(user):    print ‘user saved‘if Backend(‘user‘).delete(user):    print ‘delete user failed‘post = model.Post(‘title‘, ‘slug‘, ‘description‘, ‘html‘, ‘css‘, ‘js‘,        ‘category‘, ‘status‘, ‘comments‘, ‘author‘)if not Backend(‘post‘).create(post):    print ‘created failed‘

Future

当前只支持mysql适配驱动，因为个人并不熟悉其他关联数据库，dbpy的设计比较灵活，所以如果有高手可以尝试写写其他数据库适配，仿照 db/mysql目录如果写pgsql的适配应该不会多余800行代码。

对于构建orm框架方面，从个人来讲，更喜欢原生SQL，也不打算再造一个orm轮子。从设计和实现来说，dbpy是为了更好的发挥原生SQL优势和简单灵活。

个人一些想法：

为select加入join构建方法糖。
尝试完成schema类，用于创建表，修改表结构等。
加入一些mysql特有的sql方法糖，比如replace， on dup更新等。
优化改进连接池，比如加入固定数量连接的连接池。

关于db访问层的封装设计感想 dbpy项目的开发

声明：以上内容来自用户投稿及互联网公开渠道收集整理发布，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任，若内容有误或涉及侵权可进行投诉：投诉/举报工作人员会在5个工作日内联系你，一经查实，本站将立刻删除涉嫌侵权内容。

联系
我们

首页 > 代码库 > 关于db访问层的封装设计感想 dbpy项目的开发

关于db访问层的封装设计感想 dbpy项目的开发

dbpy

Featues

Install

Development

Compatibility

DB API

setup

query

execute

select

select all

specific columns

execute

Condition

simple

in

between

or condition

order by

distinct

group by

limit and offset

null condition

complex conditions

expr

insert

update

limit

to_sql and str

transaction

simple orm

Future

看完仍有疑问？有类似问题直接问程序猿