简介
比较小巧的ORM，支持
- sqlite
- mysql
- postgresql
- cockroachdb
数据库及表创建
自行创建数据库
CREATE DATABASE peewee_learn
创建与删除表
import datetime from peewee import MySQLDatabase,Model, CharField, TextField, IntegerField,BooleanField,DateTimeField,UUIDField my_db = MySQLDatabase( 'peewee_learn', user='root', password='123456', host='192.168.31.100', port=3306 ) class BaseModel(Model): id = UUIDField(primary_key=True) time = DateTimeField(formats='%Y-%m-%d %H:%M:%S',default=datetime.datetime.now()) class Meta: database = my_db class Users(BaseModel): name=CharField(max_length=20) age=IntegerField(null=True) gender=BooleanField() desc=TextField() class Meta: db_table = "users" def create_tables(db,models): db.connect() db.create_tables(models,safe=True) def drop_tables(db,models): db.connect() db.drop_tables(models,safe=True) if __name__ == '__main__': my_models = [Users] create_tables(my_db,my_models) # drop_tables(my_db,my_models)
IntegerField()对应的是int类型
CharField()对应的是varchar
DateTimeField()对应的是datetime类型
BooleanField()对应的是tinyint(1)类型
基本操作
增
def insert(model,**value): try: o = model() for k,v in value.items(): setattr(o,k,v) o.save() except Exception as e: print(e) def insert_many(db,model,values:List): try: with db.atomic(): q = model.insert_many(values) q.execute() except Exception as e: print(e)
查
# 查询所有 users = Users.select() print([u.name for u in users]) # 按条件查询 users = Users.select().where(Users.age==18) print([u.name for u in users]) # 排序 users = Users.select().order_by(Users.age) print([u.name for u in users]) # 函数使用 cnt = Users.select(fn.Count(Users.id).alias("cnt")) print([u.name for u in users])
改
users = Users.select().where(Users.age==18) u = users[0] u.name='liwu' u.save()
删
# 删除 users = Users.select().where(Users.age == 20) u = users[0] u.delete_instance()
全部代码
from typing import List import uuid from peeweetest.models import my_db,Users from peewee import fn def insert(model,**value): try: o = model() for k,v in value.items(): setattr(o,k,v) o.save() except Exception as e: print(e) def insert_many(db,model,values:List): try: with db.atomic(): q = model.insert_many(values) q.execute() except Exception as e: print(e) if __name__ == '__main__': # insert test zhangsan = { "id": uuid.uuid4(), "name": "three", "age": 18, "gender": True, "desc": "test for insert" } # insert(Users,name="four",age=18,gender=True,desc="test for insert") # insert_many test lisi={ "id": uuid.uuid4(), "name": "four", "age": 20, "gender": False, "desc": "test for insert many" } insert_many(my_db,Users,[zhangsan,lisi]) # 查询所有 users = Users.select() print([u.name for u in users]) # 按条件查询 users = Users.select().where(Users.age==18) print([u.name for u in users]) # 排序 users = Users.select().order_by(Users.age) print("sql:",users) print([u.name for u in users]) # 函数使用 cnt = Users.select(fn.Count(Users.id).alias("cnt")) print([u.name for u in users]) # 修改 users = Users.select().where(Users.age==18) u = users[0] u.name='liwu' u.save() # 删除 users = Users.select().where(Users.age == 20) u = users[0] u.delete_instance()
select的太多，无法列举，可以看看文档，打印出sql语句
