import datetime
from peewee import *
db = MySQLDatabase(
 "test", host="127.0.0.1", port=3306, user="root", passwd="123456"
)
db.connect()

class BaseModel(Model):
 class Meta:
  database = db
class Person(BaseModel):
 name = CharField()
 age = IntegerField()
 height = IntegerField()
 sex = BooleanField(default='male')
if __name__ == "__main__":
 Person.create_table()
 # 創(chuàng)建
 Person.create(name='tom', age=30, height=177)
 
 # 查詢
 res = Person.select().where(Person.name=='tom')
 print(res)
 print(res[0])
 print(res[0].name)
 print(res[0].age)
 print(res[0].height)
 print(res[0].sex)
>>>>
SELECT `t1`.`id`, `t1`.`name`, `t1`.`age`, `t1`.`High`, `t1`.`sex` FROM `person` AS `t1` WHERE (`t1`.`name` = 'ljk')
1
tom
30
177
True

import datetime
from peewee import *
db = SqliteDatabase('my_app.db')
class BaseModel(Model):
 class Meta:
  database = db
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)

db = SqliteDatabase('my_app.db')

class BaseModel(Model):
 class Meta:
  database = db

class User(BaseModel):
 username = CharField(unique=True)

contacts_db = SqliteDatabase('contacts.db')
class Person(Model):
 name = CharField()
 class Meta:
  database = contacts_db

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
AttributeError: type object 'Person' has no attribute 'Meta'
>>> Person._meta
<peewee.modeloptions object="" at="" 0x7f51a2f03790="">

{'id': <peewee.autofield object="" at="" 0x7f51a2e92750="">,
 'name': <peewee.charfield object="" at="" 0x7f51a2f0a510="">}
>>> Person._meta.primary_key
<peewee.autofield object="" at="" 0x7f51a2e92750="">
>>> Person._meta.database
<peewee.sqlitedatabase object="" at="" 0x7f519bff6dd0="">

class User(Model):
 username = CharField()
 join_date = DateTimeField()
 about_me = TextField()

class Message(Model):
 context = TextField()
 read_count = IntegerField(default=0)

class Message(Model):
 context = TextField()
 timestamp = DateTimeField(default=datetime.datetime.now)

def house_defaults():
 return {'beds': 0, 'baths': 0}
class House(Model):
 number = TextField()
 street = TextField()
 attributes = JSONField(default=house_defaults)

class User(Model):
 username = CharField(unique=True)
 email = CharField(index=True)

class Product(Model):
 name = CharField(unique=True)
 price = DecimalField(constraints=[Check('price < 10000')])
 created = DateTimeField(
  constraints=[SQL("DEFAULT (datetime('now'))")])

class Transaction(Model):
 from_acct = CharField()
 to_acct = CharField()
 amount = DecimalField()
 date = DateTimeField()
 class Meta:
  indexes = (
# create a unique on from/to/date
(('from_acct', 'to_acct', 'date'), True),
# create a non-unique on from/to
(('from_acct', 'to_acct'), False),
  )

>>> User.create(username='Charlie')
<__main__.User object at 0x2529350>

data_source = [
 {'field1': 'val1-1', 'field2': 'val1-2'},
 {'field1': 'val2-1', 'field2': 'val2-2'},
 # ...
]
for data_dict in data_source:
 MyModel.create(**data_dict)

# This is much faster.
with db.atomic():
 for data_dict in data_source:
  MyModel.create(**data_dict)

data_source = [
 {'field1': 'val1-1', 'field2': 'val1-2'},
 {'field1': 'val2-1', 'field2': 'val2-2'},
 # ...
]
# Fastest way to INSERT multiple rows.
MyModel.insert_many(data_source).execute()

# We can INSERT tuples as well...
data = [('val1-1', 'val1-2'),
  ('val2-1', 'val2-2'),
  ('val3-1', 'val3-2')]
# But we need to indicate which fields the values correspond to.
MyModel.insert_many(data, fields=[MyModel.field1, MyModel.field2]).execute()

# You can, of course, wrap this in a transaction as well:
with db.atomic():
 MyModel.insert_many(data, fields=fields).execute()

# Insert rows 100 at a time.
with db.atomic():
 for idx in range(0, len(data_source), 100):
  MyModel.insert_many(data_source[idx:idx+100]).execute()

from peewee import chunked
# Insert rows 100 at a time.
with db.atomic():
 for batch in chunked(data_source, 100):
  MyModel.insert_many(batch).execute()

# Read list of usernames from a file, for example.
with open('user_list.txt') as fh:
 # Create a list of unsaved User instances.
 users = [User(username=line.strip()) for line in fh.readlines()]
# Wrap the operation in a transaction and batch INSERT the users
# 100 at a time.
with db.atomic():
 User.bulk_create(users, batch_size=100)

# List of row data to insert.
row_data = [{'username': 'u1'}, {'username': 'u2'}, ...]
# Assume there are 789 items in row_data. The following code will result in
# 8 total transactions (7x100 rows + 1x89 rows).
for row in db.batch_commit(row_data, 100):
 User.create(**row)

res = (TweetArchive
 .insert_from(
  Tweet.select(Tweet.user, Tweet.message),
  fields=[TweetArchive.user, TweetArchive.message])
 .execute())

>>> user.save()  # save() returns the number of rows modified.
1
>>> user.id
1
>>> user.save()
>>> user.id
1
>>> huey.save()
1
>>> huey.id
2

>>> today = datetime.today()
>>> query = Tweet.update(is_published=True).where(Tweet.creation_date < today)
>>> query.execute()  # Returns the number of rows that were updated.
4

>>> User.get(User.id == 1)
<__main__.User object at 0x25294d0>
>>> User.get_by_id(1)  # Same as above.
<__main__.User object at 0x252df10>
>>> User[1]  # Also same as above.
<__main__.User object at 0x252dd10>
>>> User.get(User.id == 1).username
u'Charlie'
>>> User.get(User.username == 'Charlie')
<__main__.User object at 0x2529410>
>>> User.get(User.username == 'nobody')
UserDoesNotExist: instance matching query does not exist:
SQL: SELECT t1."id", t1."username" FROM "user" AS t1 WHERE t1."username" = ?
PARAMS: ['nobody']

try:
 with db.atomic():
  return User.create(username=username)
except peewee.IntegrityError:
 # `username` is a unique column, so this username already exists,
 # making it safe to call .get().
 return User.get(User.username == username)

user, created = User.get_or_create(username=username)
person, created = Person.get_or_create(
 first_name=first_name,
 last_name=last_name,
 defaults={'dob': dob, 'favorite_color': 'green'})

>>> query = User.select()
>>> [user.username for user in query]
['Charlie', 'Huey', 'Peewee']
>>> query[1]
<__main__.User at 0x7f83e80f5550>
>>> query[1].username
'Huey'
>>> query[:2]
[<__main__.User at 0x7f83e80f53a8>, <__main__.User at 0x7f83e80f5550>]

>>> query = User.select().dicts()
>>> for row in query:
...  print(row)
{'id': 1, 'username': 'Charlie'}
{'id': 2, 'username': 'Huey'}
{'id': 3, 'username': 'Peewee'}

stats = Stat.select()
# Our imaginary serializer class
serializer = CSVSerializer()
# Loop over all the stats and serialize.
for stat in stats.iterator():
 serializer.serialize_object(stat)

query = (Tweet
.select(Tweet, User)  # Select tweet and user data.
.join(User))
# Note that the user columns are stored in a separate User instance
# accessible at tweet.user:
for tweet in query:
 print(tweet.user.username, tweet.content)
# Using ".objects()" will not create the tweet.user object and assigns all
# user attributes to the tweet instance:
for tweet in query.objects():
 print(tweet.username, tweet.content)

query = Tweet.select(Tweet.content, User.username).join(User)
cursor = database.execute(query)
for (content, username) in cursor:
 print(username, '->', content)

with db.atomic():
 for data_dict in data_source:
  MyModel.create(**data_dict)

@db.atomic()
def insert_data()
 for data_dict in data_source:
  MyModel.create(**data_dict)

>>> user = User.get(User.username == 'Charlie')
>>> for tweet in Tweet.select().where(Tweet.user == user, Tweet.is_published == True):
...  print(tweet.user.username, '->', tweet.message)
...
Charlie -> hello world
Charlie -> this is fun
>>> for tweet in Tweet.select().where(Tweet.created_date < datetime.datetime(2011, 1, 1)):
...  print(tweet.message, tweet.created_date)
...
Really old tweet 2010-01-01 00:00:00
...  
print(tweet.message)
hello world
this is fun
look at this picture of my food

>>> for t in Tweet.select().order_by(Tweet.created_date):
...  print(t.pub_date)

Tweet.select().order_by(Tweet.created_date.desc())
Tweet.select().order_by(-Tweet.created_date)  # Note the "-" prefix.

User.select().order_by(+User.username)

query = (User
.select(User.username, fn.COUNT(Tweet.id).alias('num_tweets'))
.join(Tweet, JOIN.LEFT_OUTER)
.group_by(User.username))

query = (User
.select(User.username, fn.COUNT(Tweet.id).alias('num_tweets'))
.join(Tweet, JOIN.LEFT_OUTER)
.group_by(User.username)
.order_by(fn.COUNT(Tweet.id).desc()))

query = (User
.select(User.username, fn.COUNT(Tweet.id).alias('num_tweets'))
.join(Tweet, JOIN.LEFT_OUTER)
.group_by(User.username)
.order_by(SQL('num_tweets').desc()))

ntweets = fn.COUNT(Tweet.id)
query = (User
.select(User.username, ntweets.alias('num_tweets'))
.join(Tweet, JOIN.LEFT_OUTER)
.group_by(User.username)
.order_by(ntweets.desc())

>>> Tweet.select().count()
100
>>> Tweet.select().where(Tweet.id > 50).count()
50

>>> for tweet in Tweet.select().order_by(Tweet.id).paginate(2, 10):
...  print(tweet.message)
...
tweet 10
tweet 11
tweet 12
tweet 13
tweet 14
tweet 15
tweet 16
tweet 17
tweet 18
tweet 19

Tweet.select().order_by(Tweet.id).offset(10).limit(10)

mysql> select * from person;
+----+------+-----+------+-----+
| id | name | age | High | sex |
+----+------+-----+------+-----+
|  1 | ljk  |  30 |  177 |1 |
|  2 | aghj |  23 |  168 |1 |
+----+------+-----+------+-----+
2 rows in set (0.00 sec)
************************************
*  聚合函數(shù)  *
************************************
mysql> select count(*) from person;
+----------+
| count(*) |
+----------+
|  2 |
+----------+
1 row in set (0.00 sec)
----------------------------------------
mysql> select sum(age) from person;
+----------+
| sum(age) |
+----------+
| 53 |
+----------+
1 row in set (0.00 sec)
----------------------------------------
mysql> select avg(high) from person;
+-----------+
| avg(high) |
+-----------+
|  172.5000 |
+-----------+
1 row in set (0.00 sec)
----------------------------------------
mysql> select max(high) from person;
+-----------+
| max(high) |
+-----------+
| 177 |
+-----------+
1 row in set (0.00 sec)
mysql> select * from person;
+----+------+-----+------+-----+
| id | name | age | High | sex |
+----+------+-----+------+-----+
|  1 | ljk  |  30 |  177 |1 |
|  2 | aghj |  23 |  168 |1 |
|  3 | 0 |  22 |  165 |0 |
+----+------+-----+------+-----+
3 rows in set (0.00 sec)

mysql> select avg(High) from person group by sex;
+-----------+
| avg(High) |
+-----------+
|  172.5000 |
|  165.0000 |
+-----------+
2 rows in set (0.00 sec)
# 使用having對分組的數(shù)據(jù)篩選
mysql> select avg(High) as high from person group by sex having high > 170;
+----------+
| high  |
+----------+
| 172.5000 |
+----------+
1 row in set (0.00 sec)

on->join->where->group by->having

query = (User
.select(User, fn.Count(Tweet.id).alias('count'))
.join(Tweet, JOIN.LEFT_OUTER)
.group_by(User))

class Photo(Model):
 image = CharField()
class Tag(Model):
 name = CharField()
class PhotoTag(Model):
 photo = ForeignKeyField(Photo)
 tag = ForeignKeyField(Tag)

query = (Tag
.select()
.join(PhotoTag)
.join(Photo)
.group_by(Tag)
.having(fn.Count(Photo.id) > 5))

>>> Employee.select(
...  fn.Min(Employee.salary), fn.Max(Employee.salary)
... ).scalar(as_tuple=True)
(30000, 50000)

>>> Employee.select(
...  fn.Min(Employee.salary), fn.Max(Employee.salary)
... ).scalar(as_tuple=True)
(30000, 50000)

# Find the user whose username is "charlie".
User.select().where(User.username == 'charlie')
# Find the users whose username is in [charlie, huey, mickey]
User.select().where(User.username.in_(['charlie', 'huey', 'mickey']))
Employee.select().where(Employee.salary.between(50000, 60000))
Employee.select().where(Employee.name.startswith('C'))
Blog.select().where(Blog.title.contains(search_string))

query = (User
.select(User, fn.COUNT(Tweet.id).alias('tweet_count'))
.join(Tweet, JOIN.LEFT_OUTER)
.group_by(User)
.order_by(fn.COUNT(Tweet.id).desc()))
for user in query:
 print('%s -- %s tweets' % (user.username, user.tweet_count))

# We'll query the user table and annotate it with a count of tweets for
# the given user
query = (User
.select(User, fn.Count(Tweet.id).alias('ct'))
.join(Tweet)
.group_by(User))
# Now we will order by the count, which was aliased to "ct"
query = query.order_by(SQL('ct'))
# You could, of course, also write this as:
query = query.order_by(fn.COUNT(Tweet.id))

query = MyModel.raw('SELECT * FROM my_table WHERE data = %s', user_data)
query.execute_sql()

query = MyModel.raw('SELECT * FROM my_table WHERE data = %s' % (user_data,))
# Good. `user_data` will be treated as a parameter to the query.
query = MyModel.raw('SELECT * FROM my_table WHERE data = %s', user_data)
# Bad! DO NOT DO THIS!
query = MyModel.select().where(SQL('Some SQL expression %s' % user_data))
# Good. `user_data` will be treated as a parameter.
query = MyModel.select().where(SQL('Some SQL expression %s', user_data))