浅谈Python peewee 使用经验
本文使用案例是基于python2.7实现
以下内容均为个人使用peewee的经验和遇到的坑，不会涉及过多的基本操作。所以，没有使用过peewee，可以先阅读文档
正确性和覆盖面有待提高，如果遇到新的问题欢迎讨论。
一、介绍
Peewee是一个简单、轻巧的PythonORM。
- 简单、轻巧、富有表现力（原词expressive）的ORM
- 支持python版本2.6+和3.2+
- 支持数据库包括：sqlite，mysqlandpostgresql
- 包含一堆实用的扩展在playhouse模块中
总而言之，peewee可以完全可以应付个人或企业的中小型项目的Model层，上手容易，功能很强大。
二、基本使用方法
frompeeweeimport*
db=SqliteDatabase('people.db')
classBaseModel(Model):
classMeta:
database=db#Thismodelusesthe"people.db"database.
classPerson(BaseModel):
name=CharField()
birthday=DateField()
is_relative=BooleanField()
基本的使用方法，推荐阅读文档－－quickstart
三、推荐使用姿势
下面介绍一些我在使用过程的经验和遇到的坑，希望可以帮助大家更好的使用peewee。
3.1连接数据库
连接数据库时，推荐使用playhouse中的db_url模块。db_url的connect方法可以通过传入的URL字符串，生成数据库连接。
3.1.1connect(url,**connect_params)
通过传入的url字符串，创建一个数据库实例
url形如：
- mysql://user:passwd@ip:port/my_db将创建一个本地MySQL的my_db数据库的实例（willcreateaMySQLDatabaseinstance）
- mysql+pool://user:passwd@ip:port/my_db?charset=utf8&max_connections=20&stale_timeout=300将创建一个本地MySQL的my_db的连接池，最大连接数为20（Inamulti-threadedapplication,uptomax_connectionswillbeopened.Eachthread(or,ifusinggevent,greenlet)willhaveit'sownconnection.），超时时间为300秒（willcreateaPooledMySQLDatabaseinstance）
注意：charset默认为utf8。如需要支持emoji，charset设置为utf8mb4，同时保证创建数据库时的字符集设置正确CREATEDATABASEmydatabaseCHARACTERSETutf8mb4COLLATEutf8mb4_unicode_ci;。
支持的schemes：
- apsw:APSWDatabase
- mysql:MySQLDatabase
- mysql+pool:PooledMySQLDatabase
- postgres:PostgresqlDatabase
- postgres+pool:PooledPostgresqlDatabase
- postgresext:PostgresqlExtDatabase
- postgresext+pool:PooledPostgresqlExtDatabase
- sqlite:SqliteDatabase
- sqliteext:SqliteExtDatabase
- sqlite+pool:PooledSqliteDatabase
- sqliteext+pool:PooledSqliteExtDatabase
3.1.2推荐姿势
fromplayhouse.db_urlimportconnect
fromdock.commonimportconfig
＃url:mysql+pool://root:root@127.0.0.1:3306/appmanage?max_connections=300&stale_timeout=300
mysql_config_url=config_dict.get('config').get('mysql').get('url')
db=connect(url=mysql_config_url)
查看更多详情请移步官方文档：db-url
3.2连接池的使用
peewee的连接池，使用时需要显式的关闭连接。下面先说下为什么，最后会给出推荐的使用方法，避免进坑。
3.2.1为什么要显式的关闭连接
Connectionswillnotbeclosedexactlywhentheyexceedtheirstale_timeout.Instead,staleconnectionsareonlyclosedwhenanewconnectionisrequested.
这里引用官方文档的提示。大致说：“超时连接不会自动关闭，只会在有新的请求时是才会关闭”。这里的request是指‘web框架处理的请求'，peewee源码片段：
def_connect(self,*args,**kwargs):
whileTrue:
try:
#Removetheoldestconnectionfromtheheap.
ts,conn=heapq.heappop(self._connections)#_connections是连接实例的list(pool)
key=self.conn_key(conn)
exceptIndexError:
ts=conn=None
logger.debug('Noconnectionavailableinpool.')
break
else:
ifself._is_closed(key,conn):
#Thisconnectonwasclosed,butsinceitwasnotstale
#itgotaddedbacktothequeueofavailableconns.We
#thencloseditandmarkeditasexplicitlyclosed,so
#it'ssafetothrowitawaynow.
#(BecauseDatabase.close()callsDatabase._close()).
logger.debug('Connection%swasclosed.',key)
ts=conn=None
self._closed.discard(key)
elifself.stale_timeoutandself._is_stale(ts):
#Ifweareattemptingtocheckoutastaleconnection,
#thencloseit.Wedon'tneedtomarkitinthe"closed"
#set,becauseitisnotinthelistofavailableconns
#anymore.
logger.debug('Connection%swasstale,closing.',key)
self._close(conn,True)
self._closed.discard(key)
ts=conn=None
else:
break
ifconnisNone:
ifself.max_connectionsand(
len(self._in_use)>=self.max_connections):
raiseValueError('Exceededmaximumconnections.')
conn=super(PooledDatabase,self)._connect(*args,**kwargs)
ts=time.time()
key=self.conn_key(conn)
logger.debug('Creatednewconnection%s.',key)
self._in_use[key]=ts#使用中的数据库连接实例dict
returnconn
根据pool库中的_connect方法的代码可知：每次在建立数据库连接时，会检查连接实例是否超时。但是需要注意一点：使用中的数据库连接实例（_in_usedict中的数据库连接实例)，是不会在创建数据库连接时，检查是否超时的。
因为这段代码中，每次创建连接实例，都是在_connections(pool)取实例，如果有的话就判断是否超时；如果没有的话就新建。
然而，使用中的数据库连接并不在_connections中，所以每次创建数据库连接实例时，并没有检测使用中的数据库连接实例是否超时。
只有调用连接池实例的_close方法。执行这个方法后，才会把使用后的连接实例放回到_connections(pool)。
def_close(self,conn,close_conn=False):
key=self.conn_key(conn)
ifclose_conn:
self._closed.add(key)
super(PooledDatabase,self)._close(conn)#关闭数据库连接的方法
elifkeyinself._in_use:
ts=self._in_use[key]
delself._in_use[key]
ifself.stale_timeoutandself._is_stale(ts):#到这里才会判断_in_use中的连接实例是否超时
logger.debug('Closingstaleconnection%s.',key)
super(PooledDatabase,self)._close(conn)#超时的话，关闭数据库连接
else:
logger.debug('Returning%stopool.',key)
heapq.heappush(self._connections,(ts,conn))#没有超时的话，放回到pool中
3.2.2如果不显式的关闭连接，会出现的问题
如果不调用_close方法的话，使用后的数据库连接就一直不会关闭(两个含义：回到pool中和关闭数据库连接)，这样会造成两个问题：
1.每次都是新建数据库连接，因为pool中没有数据库连接实例。会导致稍微有一点并发量就会返回Exceededmaximumconnections.错误
2.MySQL也是有timeout的，如果一个连接长时间没有请求的话，MySQLServer就会关闭这个连接，但是，peewee的已建立（后面会解释为什么特指已建立的）的连接实例，并不知道MySQLServer已经关闭了，再去通过这个连接请求数据的话，就会返回Error2006:“MySQLserverhasgoneaway”错误，根据官方文档
3.2.3推荐姿势
所以，每次操作完数据库就关闭连接实例。
用法1：使用with
defsend_rule(): withdb.execution_context(): #Anewconnectionwillbeopenedor,ifusingaconnectionpool, #pulledfromthepoolofavailableconnections.Additionally,a #transactionwillbestarted. foruseringet_all_user(): user_id=user['id'] rule=Rule(user_id) rule_dict=rule.slack_rule(index) .....dosomething.....
用法2：使用Flaskhook
@app.before_request def_db_connect(): database.connect() # #Thishookensuresthattheconnectionisclosedwhenwe'vefinished #processingtherequest. @app.teardown_request def_db_close(exc): ifnotdatabase.is_closed(): database.close() # # #更优雅的用法： fromplayhouse.flask_utilsimportFlaskDB fromdock_fastgear.model.baseimportdb # app=Flask(__name__) FlaskDB(app,db)#这样就自动做了上面的事情（具体实现可查看http://docs.peewee-orm.com/en/latest/peewee/playhouse.html?highlight=Flask%20DB#flask-utils）
查看更多详情请移步官方文档：pool-apis
3.3处理查询结果
这里没有什么大坑，就是有两点需要注意：
首先，查询的结果都是该Model的object，注意不是dict。如果想让结果为dict，需要playhouse模块的工具方法进行转化：fromplayhouse.shortcutsimportmodel_to_dict
其次，get方法只会返回一条记录
3.3.1推荐姿势
fromplayhouse.shortcutsimportmodel_to_dict frommodelimportHelloGitHub defread_from_db(input_vol): content_list=[] category_object_list=HelloGitHub.select(HelloGitHub.category).where(HelloGitHub.vol==input_vol)\ .group_by(HelloGitHub.category).order_by(HelloGitHub.category) forfi_category_objectincategory_object_list: hellogithub=HelloGitHub.select()\ .where((HelloGitHub.vol==input_vol) &(HelloGitHub.category==fi_category_object.category))\ .order_by(HelloGitHub.create_time) forfi_hellogithubinhellogithub: content_list.append(model_to_dict(fi_hellogithub)) returncontent_list
四、常见错误及解决办法
4.1'buffer'objecthasnoattribute'translate'
- 错误信息："'buffer'objecthasnoattribute'translate'"
- 场景：BlobField字段存储zlibcompress压缩的数据
- 解决办法：需要指定pymysql的版本小于0.6.7否则会报错
- 参考
4.2Can'tconnecttoMySQLserverLostconnectiontoMySQLserverduringquery
- 错误信息：Can'tconnecttoMySQLserverLostconnectiontoMySQLserverduringquery
- 场景：向RDS中插入数据
- 解决办法：因为请求的连接数过多，达到了RDS设置的连接数，所以需要调高RDS连接数
- 参考
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持毛票票。