peewee非常适合处理一些简单的crud任务。如果是复杂的查询语句,我更喜欢用rawsql来处理。以连接查询为例:models.pyfrompeeweeimport*importsettingshost=settings.DATABASE.hostport=settings。DATABASE.portusername=settings.DATABASE.usernamepassword=settings.DATABASE.passworddatabase_name=settings.DATABASE.databasedb=MySQLDatabase(database=database_name,host=host,port=port,user=username,password=password)classUser(模型):id=AutoField()name=CharField(max_length=255,null=False)created_at=DateTimeField(null=False,constraints=[SQL('DEFAULTCURRENT_TIMESTAMP')],help_text='使用数据库时间')updated_at=DateTimeField(null=False,constraints=[SQL('DEFAULTCURRENT_TIMESTAMP'),SQL('ONUPDATECURRENT_TIMESTAMP'),])classMeta:database=dbtable_name='account'classOrder(Model):"""订单信息表"""id=AutoField(主_key=True)user_id=IntegerField(null=False,unique=False,index=True)product_id=IntegerField(null=False,unique=False)activity_id=IntegerField(null=False,unique=False)is_paid=BooleanField(null=False,unique=False,default=False)created_at=DateTimeField(null=False,constraints=[SQL('DEFAULTCURRENT_TIMESTAMP')])updated_at=DateTimeField(null=False,约束=[SQL('DEFAULTCURRENT_TIMESTAMP'),SQL('ONUPDATECURRENT_TIMESTAMP'),])classMeta:database=dbtable_name='order'#http://docs.peewee-orm.com/en/latest/peewee/models.html?highlight=table%20generation#multi-column-indexesindexes=(#createauniqueon,限购一次(('user_id','product_id','activity_id'),True),)先来插入一些测试数据,侧面的测试顺序表insertinto`order`(user_id,product_id,activity_id,is_paid)values(1,2,3,0)insertinto`order`(user_id,product_id,activity_id,is_paid)values(2,3,4,0)account表insertinto`account`(name)values('jike')insertinto`account`(name)values('ponponon')看一看插入后的结果MySQLroot@192.168.31.245:seckill>select*fromaccount;+----+--------+----------------------+--------------------+|编号|姓名|创建时间|updated_at|+----+----------+--------------------+------------------+|1|极客|2022-04-0613:26:28|2022-04-0613:31:34||2|蓬蓬|2022-04-0613:26:35|2022-04-0613:26:35|+----+--------+--------------------+------------------+2rowsinsetTime:0.005s各两条MySQLroot@192.168.31.245:秒杀>select*from`order`+----+--------+------------+------------+--------+------------------+--------------------+|编号|用户ID|产品编号|活动编号|is_paid|创建时间|updated_at|+----+--------+------------+------------+---------+--------------------+----------------------+|1|1|2|3|0|2022-04-0613:25:19|2022-04-0613:25:19||2|2|3|4|0|2022-04-0613:25:28|2022-04-0613:25:28|+----+--------+------------+-------------+--------+--------------------+------------------+2rowsinsetTime:0.007s使用单元测试的脚底来验证importunittestfromloguruimportloggerfrommodelsimportdbfrompymysql.cursorsimportCursorclassTestProject(unittest.TestCase):defpeewee_exec_raw_sql(self):"""python-munittesttests.TestProject.peewee_exec_raw_sql"""#cursor=db.cursor()withdb.cursor()ascursor:cursor:Cursorsql="""SELECT`account`.`id`,`account`.`name`,`order`.`product_id`FROM`account`INNERJOIN`order`ON(`account`.`id`=`order`.`user_id`)"""cursor.execute(sql)cursor.connection.commit()#必须在这里提交,否则所有查询都会在一个事务行中:tuple[tuple]=cursor.fetchall()forrowinrows:logger.debug(row)因为peewee禁用了自动提交,所以查询后需要手动提交,使用peewee的MODEL.select()方法,会自动提交(前提是没有交易开启)执行结果─?python-munittesttests.TestProject.peewee_exec_raw_sql2022-04-0613:37:20.893|DEBUG|tests:peewee_exec_raw_sql:36-(1,'jike',2)2022-04-0613:37:20.894|DEBUG|tests:peewee_exec_raw_sql:36-(2,'ponponon',3).----------------------------------------------------------------------在0.005秒内进行1次测试参考文章:Python编程:peewee执行多个原生sql语句做这些事情:SETsql_mode='PIPES_AS_CONCAT'SETAUTOCOMMIT=0注意第二个SETAUTOCOMMIT=0!
