python轻量级orm框架 peewee常用功能速查详情

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()

    # 创建
    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)
************************************
*              聚合函数              *
************************************
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对分组的数据筛选
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))