Peewee-笔记，轻量快速的ORM_playhouse.shortcuts.reconnectmixin

安装

pip install peewee

使用背景：

1.相比于操作pymysql，peewee增删改查更方便

2.对于大数据量存储mysql，peewee可更好的胜任，（如：实例情况，10w条100列的txt文件，peewee在1分钟左右完成导入）

文件结构：

 models.py内容：

import peewee
import traceback
import datetime
from peewee import *
 
from playhouse.shortcuts import ReconnectMixin
 
class ReconnectMySQLDatabase(ReconnectMixin, MySQLDatabase):
    pass
 
 
try:
    db = ReconnectMySQLDatabase('db', host='127.0.0.1', port= 3306, user= 'root', passwd='123456',sql_mode='NO_AUTO_CREATE_USER')
 
except:
    traceback.print_exc()
    exit()
 
class BaseModel(peewee.Model):
    """A base model that will use our mysql database"""
 
    class Meta:
        database = db
 
 
class Test(BaseModel):
    rec_id = peewee.AutoField(primary_key=True)
    class Meta:
        db_table = 'Test'
        verbose_name = '测试'
        verbose_name_plural = verbose_name

insert.py内容:

import peewee
import pandas as pd
import traceback
import click
from .models import *
 
 
try:
    db.connect()
except:
    traceback.print_exc()
    exit()
        #判断连接是否终端
if db.is_closed():
   print('mysql reconnect')
   db = ReconnectMySQLDatabase('db', host='127.0.0.1', port= 3306, user= 'root', passwd='123456',sql_mode='NO_AUTO_CREATE_USER')
   db.connect()
 
 
“”“操作”“”

事务：

with db.atomic() as  txn:
    try:
        """操作"""
        pass
    except Exception as e:
        txn.rollback()
        exit(traceback.print_exc())

判空：

Test.select().where(Test.rec_id== 1).exists()

 查找：

Test.get(rec_id=1).rec_id

 批次新增：

#list为Test表结构相同的字典组成的列表，如[{“rec_id”:1}]
for batch in peewee.chunked(list, 1000):
    Test.insert_many(batch).execute()

 更新：

#dict为Test表结构相同的字典，如{“rec_id”:1}
Test.update(dict).where(Test.rec_id== 1).execute()