peewee是一款轻量级、丰富的ORM(Object Relation Mapping,对象关系映射),支持postgresql、mysql和sqlite。
简单示例
当使用peewee开始一个项目,通常最好通过定义一个或者多个模型类开始你的数据模型。
from peewee import *
db = SqliteDatabase('people.db')
class Person(Model):
name = CharField()
birthday = DateField()
is_relative = BooleanField()
class Meta:
database = db # 这个模型使用 "people.db"数据库
大量的插件支持:
- Postgresql HStore,Json,Arrays
- SQLite全文搜索、自定义函数、虚拟表格
- 结构迁移和model代码生成器
- 连接池
- 加密
- 更多
示例
模型定义和Django、SQLAlchemy类似
from peewee import *
from playhouse.sqlite_ext import SqliteExtDatabase
import datetime
db = SqliteExtDatabase('my_database.db')
class BaseModel(Model):
class Meta:
database = db
class User(BaseModel):
username = CharField(unique=True)
class Tweet(BaseModel):
user = ForeignKeyField(User, related_name='tweets')
message = TextField()
created_date = DateTimeField(default=datetime.datetime.now)
is_published = BooleanField(default=True)
连接到数据库并创建表格和添加数据
db.connect()
db.create_tables([User, Tweet])
charlie = User.create(username='charlie')
huey = User(username='huey')
huey.save()
# No need to set `is_published` or `created_date` since they
# will just use the default values we specified.
Tweet.create(user=charlie, message='My first tweet')
查询
# A simple query selecting a user.
User.get(User.username == 'charles')
# Get tweets created by one of several users. The "<<" operator
# corresponds to the SQL "IN" operator.
usernames = ['charlie', 'huey', 'mickey']
users = User.select().where(User.username << usernames)
tweets = Tweet.select().where(Tweet.user << users)
# We could accomplish the same using a JOIN:
tweets = (Tweet
.select()
.join(User)
.where(User.username << usernames))
# How many tweets were published today?
tweets_today = (Tweet
.select()
.where(
(Tweet.created_date >= datetime.date.today()) &
(Tweet.is_published == True))
.count())
# Paginate the user table and show me page 3 (users 41-60).
User.select().order_by(User.username).paginate(3, 20)
# Order users by the number of tweets they've created:
tweet_ct = fn.Count(Tweet.id)
users = (User
.select(User, tweet_ct.alias('ct'))
.join(Tweet, JOIN.LEFT_OUTER)
.group_by(User)
.order_by(tweet_ct.desc()))
# Do an atomic update
Counter.update(count=Counter.count + 1).where(
Counter.url == request.url)
网友评论