#Peewee是一个简单而小的ORM。它几乎没有(但富有表现力的)概念,使其易于学习和直观使用。
#安装
# pip install peewee -i "http://mirrors.aliyun.com/pypi/simple" --trusted-host "mirrors.aliyun.com"
#使用Mysql安装驱动
# pip install pymysql -i "http://mirrors.aliyun.com/pypi/simple" --trusted-host "mirrors.aliyun.com"
#导入包
import peewee
from peewee import Model,CharField,ForeignKeyField,TextField,DateTimeField,BooleanField,AutoField,fn,chunked
import datetime
import Mek_master
#连接数据库
db_sqlite = peewee.SqliteDatabase('My_sqlite.db') #连接sqlite数据库
db_mysql = peewee.MySQLDatabase(host = '10.86.1.102',port = 3306, user = 'mek', #连接MySQL
passwd = 'passwd', database = 'testpeewee')
#抽象
'''
字段初始化参数及默认参数:
null = False – 允许空值
index = False – 在此列上创建索引
unique = False – 在此列上创建唯一索引。另请参阅添加复合索引。
column_name = None – 显式指定数据库中的列名。
default = None – 任何值或可调用的值,可用作未初始化模型的默认值
primary_key = False – 表的主键
constraints = None - 一个或多个约束,例如[Check('price > 0')]
sequence = None – 序列名称(如果后端支持)
collation = None – 用于排序字段/索引的排序规则
unindexed = False – 指示虚拟表上的字段应为未编制索引(仅限 SQLite)
choices = None – 可选的可迭代,包含 2 元组 ,valuedisplay
help_text = None – 表示此字段的任何有用文本的字符串
verbose_name = None – 表示此字段的"用户友好"名称的字符串
index_type = None – 指定自定义索引类型,例如,对于 Postgres,您可以指定一个 或 索
''' #字段初始化参数及默认参数
'''
字段类型表:
字段类型对应数据库 Sqlite Postgresql MySQL
AutoField integer serial integer
BigAutoField integer bigserial bigint
IntegerField integer integer integer
BigIntegerField integer bigint bigint
SmallIntegerField integer smallint smallint
IdentityField not supported int identity not supported
FloatField real real real
DoubleField real double precision double precision
DecimalField decimal numeric numeric
CharField varchar varchar varchar
FixedCharField char char char
TextField text text text
BlobField blob bytea blob
BitField integer bigint bigint
BigBitField blob bytea blob
UUIDField text uuid varchar(40)
BinaryUUIDField blob bytea varbinary(16)
DateTimeField datetime timestamp datetime
DateField date date date
TimeField time time time
TimestampField integer integer integer
IPField integer bigint bigint
BooleanField integer boolean bool
BareField untyped not supported not supported
ForeignKeyField integer integer integer
''' #字段类型表
#基础类,用于指定数据库
class BaseModel(Model):
class Meta:
database = db_sqlite
#具体抽象类
class User(BaseModel):
username = CharField(unique=True)
#具体抽象类
class Tweet(BaseModel):
user = ForeignKeyField(User, backref='tweets') #外键
message = TextField()
created_date = DateTimeField(default=datetime.datetime.now)
is_published = BooleanField(default=True)
def create_Model(): #基于数据库创建模型
pass
'''
MySQL
python -m pwiz -e mysql -H localhost -p 3306 -u root -P -t table_names database_name > model.py
python -m pwiz -e mysql -H 10.86.0.10 -p 3306 -u root -P -t image weiboSearch > model.py
'''
def create_tab(db): #创建数据表
db.connect() #连接至数据库
db.create_tables([User, Tweet]) #创建数据表
def insert_data(): #插入数据
import Mek_master
#插入数据
User.create(username=Mek_master.get_random_letters(12))
New_User = User(username=Mek_master.get_random_letters(12))
New_User.save()
#外键插入数据
new_tweet = Tweet(user=New_User,message=f'{Mek_master.get_random_hax(128)}')
new_tweet.save()
new = Tweet(user=69, message=Mek_master.get_random_hax(128))
new.save()
#批量插入
data = [(Mek_master.get_random_letters(4),), (Mek_master.get_random_letters(4),),
(Mek_master.get_random_letters(4),)]
User.insert_many(data, fields=[User.username]).execute()
#大规模插入数据 根据数据源中的行数,您可能需要将其分解为块。特别是 SQLite 通常每个查询有 999 或 32766 个变量的限制(批大小为 999 // 行长度或 32766 // 行长度)。
data_source = [(Mek_master.get_random_letters(4),),(Mek_master.get_random_letters(4),),(Mek_master.get_random_letters(4),),(Mek_master.get_random_letters(4),),(Mek_master.get_random_letters(4),),
(Mek_master.get_random_letters(4),),(Mek_master.get_random_letters(4),),(Mek_master.get_random_letters(4),),(Mek_master.get_random_letters(4),),(Mek_master.get_random_letters(4),),
(Mek_master.get_random_letters(4),),(Mek_master.get_random_letters(4),),(Mek_master.get_random_letters(4),),(Mek_master.get_random_letters(4),),(Mek_master.get_random_letters(4),),
(Mek_master.get_random_letters(4),),(Mek_master.get_random_letters(4),),(Mek_master.get_random_letters(4),),(Mek_master.get_random_letters(4),),(Mek_master.get_random_letters(4),)]
with db_sqlite.atomic():
for batch in chunked(data_source, 2):
User.insert_many(batch, fields=[User.username]).execute()
def Query_data(): #查询对象1
'''单个对象查询'''
#tweet = Tweet.get(message='OK') #字段匹配查询
#print(tweet.user.username)
tweet = Tweet.get_by_id(898) #主键查询
#print(tweet,tweet.message)
'''多个对象查询'''
users = User.select() #查询该表全部数据,结果可迭代
username_list = [i.username for i in users] #将查询结果某字段转换为列表
#print(username_list)
'''
查询条件连接符:
| 或
& 和
~ 否定条件
查询条件对比符:
== x 等于 y
< x 小于 y
<= x 小于或等于 y
> x 大于 y
>= x 大于或等于 y
!= x 不等于 y
<< x IN y,其中 y 是列表或查询
>> x 是 y,其中 y 是无/空
% x LIKE y,其中 y 可能包含通配符
** x ILIKE y,其中 y 可能包含通配符
^ x 异或 y
~ 一元否定(例如,不是 x)
查询条件对比函数:
.in_(value) IN 查找(与 相同)。<< *
.not_in(value) 不在查找中。
.is_null(is_null) 为空或不为空。接受布尔参数。
.contains(substr) 子字符串的通配符搜索。 *
.startswith(prefix) 搜索以 开头的值。prefix *
.endswith(suffix) 搜索以 结尾的值。suffix *
.between(low, high) 在 和 之间搜索值。lowhigh *
.regexp(exp) 正则表达式匹配(区分大小写)。
.iregexp(exp) 正则表达式匹配(不区分大小写)。
.bin_and(value) 二进制和。
.bin_or(value) 二进制或。
.concat(other) 使用 连接两个字符串或对象。||
.distinct() 标记列以进行不同的选择。
.collate(collation) 指定具有给定排序规则的列。
.cast(type) 将列的值转换为给定类型。
'''#查询操作符
#where() 匹配查询条件 order_by() 排序 .desc()倒序 join()联表
#演示语句
users = User.select().where(User.username.contains('CXC') | User.username.contains('AAB')).order_by(User.username.desc())
username_list = [i.username for i in users]
#print(username_list)
#联表查询演示语句
tweets = Tweet.select().join(User).where(User.username.contains('A')).order_by(User.username,Tweet.created_date.desc())
#for i in tweets:
#print(i.message,i.user.username,i.created_date)
#分页记录
users = User.select().paginate(2,10) #分页 页码:每页个数
users2 = User.select().limit(100) #取前100个
users3 = User.select().offset(100).limit(10) #从100开始向后取10个
#for i in users3:
#print(i.username)
#随机获取
#LotteryNumber.select().order_by(fn.Random()).limit(5) Postgresql和Sqlite
#LotteryNumber.select().order_by(fn.Rand()).limit(5) Mysql
user = User.select().order_by(fn.Random()).limit(1)[0]
print(user.username)
for i in user.tweets: #外键查询,结果可迭代
print(i.message,i.created_date)
def delete_data():
#第一种 查询后删除
user = User.get_by_id(23)
print(user.username)
user.delete_instance()
#第二种 根据主键删除
User.delete_by_id(24)
#第三种 根据条件批量删除
users = User.delete().where(User.username.contains('SB')) #打印删除行数
users.execute() #提交删除
def update_data():
import Mek_master
#第一种 查询后保存
tweet = Tweet.get(Tweet.message.contains('EDEAC'))
print(tweet.message)
tweet.message=tweet.message+'AMD'
tweet.save()
tweet2 = Tweet.get(Tweet.message.contains('AMD'))
print(tweet2.message)
#第二种 批量更新
tweets = Tweet.update(message='0000000000A').where(Tweet.message.contains(Mek_master.get_random_hax(4)))
#tweets.execute() #提交更新
#原子更新
tweets = Tweet.update(message='0x'+Tweet.message).where(Tweet.message.contains(Mek_master.get_random_hax(4)))
tweets.execute() #提交更新
def create_Test_data(): #创建测试数据
import Mek_master
for i in range(1):
User.create(username=Mek_master.get_random_letters(7))
for i in range(70000):
Tweet.create(user=User.select().order_by(fn.Random()).limit(1)[0], message=Mek_master.get_random_hax(128))
if __name__ == '__main__':
pass
#Query_data()
#delete_data()
#update_data()
#批量添加