flask-sqlalchemy 简单笔记

作者: 人世间 | 来源:发表于2014-08-07 15:51 被阅读15286次

flask-sqlalchemy

SQLAlchemy已经成为了python世界里面orm的标准，flask是一个轻巧的web框架，可以自由的使用orm，其中flask-sqlalchemy是专门为flask指定的插件。

安装flask-sqlalchemy

pip install flask-sqlalchemy

初始化sqlalchemy

from flask import Flask
from flask.ext.sqlalchemy import SQLAlchemy

app = Flask(__name__)

#                  dialect+driver://username:password@host:port/database?charset=utf8
# 配置 sqlalchemy  数据库驱动://数据库用户名:密码@主机地址:端口/数据库?编码
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:@localhost:3306/sqlalchemy?charset=utf8'
# 初始化
db = SQLAlchemy(app)

定义model

class User(db.Model):
    """ 定义了三个字段， 数据库表名为model名小写
    """
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True)
    email = db.Column(db.String(120), unique=True)

    def __init__(self, username, email):
        self.username = username
        self.email = email

    def __repr__(self):
        return '<User %r>' % self.username

    def save(self):
        db.session.add(self)
        db.session.commit()

创建数据表

数据包的创建使用sqlalchemy app，如果表已经存在，则忽略，如果不存在，则新建

>>> from yourapp import db, User
>>> u = User(username='admin', email='admin@example.com') # 创建实例
>>> db.session.add(u)                                     # 添加session
>>> db.session.commit()                                   # 提交查询
>>> users = User.query.all()                              # 查询

需要注意的是，如果要插入中文，必须插入 unicode字符串

>>> u = User(username=u'人世间', email='rsj@example.com')
>>> u.save()

定义关系

关系型数据库，最重要的就是关系。通常关系分为一对一（例如无限级栏目），一对多（文章和栏目），多对多（文章和标签）

one to many

我们定义一个Category（栏目）和Post（文章），两者是一对多的关系，一个栏目有许多文章，一个文章属于一个栏目。

class Category(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50))

    def __init__(self, name):
        self.name = name

    def __repr__(self):
        return '<Category %r>' % self.name

class Post(db.Model):
    """ 定义了五个字段，分别是 id，title，body，pub_date，category_id
    """
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(80))
    body = db.Column(db.Text)
    pub_date = db.Column(db.String(20))
    # 用于外键的字段
    category_id = db.Column(db.Integer, db.ForeignKey('category.id'))
    # 外键对象，不会生成数据库实际字段
    # backref指反向引用，也就是外键Category通过backref(post_set)查询Post
    category = db.relationship('Category', backref=db.backref('post_set', lazy='dynamic'))


    def __init__(self, title, body, category, pub_date=None):
        self.title = title
        self.body = body
        if pub_date is None:
            pub_date = time.time()
        self.pub_date = pub_date
        self.category = category

    def __repr__(self):
        return '<Post %r>' % self.title

    def save(self):
        db.session.add(self)
        db.session.commit()

如何使用查询呢？

>>> c = Category(name='Python')
>>> c
<Category 'Python'>
>>> c.post_set
<sqlalchemy.orm.dynamic.AppenderBaseQuery object at 0x0000000003B58F60>
>>> c.post_set.all()
[]
>>> p = Post(title='hello python', body='python is cool', category=c)
>>> p.save()
>>> c.post_set
<sqlalchemy.orm.dynamic.AppenderBaseQuery object at 0x0000000003B73710>
>>> c.post_set.all()   # 反向查询
[<Post u'hello python'>]
>>> p
<Post u'hello python'>
>>> p.category
<Category u'Python'>
# 也可以使用category_id 字段来添加
>>> p = Post(title='hello flask', body='flask is cool', category_id=1)
>>> p.save()

many to many （评论已经指出，这样的做法无法关联删除，简书没有删除线格式，多多对例子作废，在此提示，以免被误导）

对于多对多的关系，往往是定义一个两个model的id的另外一张表，例如 Post 和 Tag之间是多对多，需要定义一个 Post_Tag的表

post_tag = db.Table('post_tag',
                    db.Column('post_id', db.Integer, db.ForeignKey('post.id')),
                    db.Column('tag_id', db.Integer, db.ForeignKey('tag.id'))
                )

class Post(db.Model):

    id = db.Column(db.Integer, primary_key=True)
    # ... 省略
    # 定义一个反向引用，tag可以通过 post_set查询到 post的集合
    tags = db.relationship('Tag', secondary=post_tag,
                           backref=db.backref('post_set', lazy='dynamic'))

class Tag(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    content = db.Column(db.String(10), unique=True)
    # 定义反向查询
    posts = db.relationship('Post', secondary=post_tag,
                            backref=db.backref('tag_set', lazy='dynamic'))

    def __init__(self, content):
        self.content = content

    def save(self):
        db.session.add(self)
        db.session.commit()

查询

>>> tag_list = []
>>> tags = ['python', 'flask', 'ruby', 'rails']
>>> for tag in tags:
        t = Tag(tag)
        tag_list.append(t)
>>> tag_list
[<f_sqlalchemy.Tag object at 0x0000000003B7CF28>, <f_sqlalchemy.Tag object at 0x0000000003B7CF98>, <f_sqlalchemy.Tag object at 0x0000000003B7CEB8>, <f_sqlalchemy.Tag object at 0x0000000003B7CE80>]
>>> p
<Post u'hello python'>
>>> p.tags
[]
>>> p.tags = tag_list   # 添加多对多的数据
>>> p.save()
>>> p.tags
[<f_sqlalchemy.Tag object at 0x0000000003B7CF28>, <f_sqlalchemy.Tag object at 0x0000000003B7CF98>, <f_sqlalchemy.Tag object at 0x0000000003B7CEB8>, <f_sqlalchemy.Tag object at 0x0000000003B7CE80>]
>>> p.tag_set           # 反向查询
<sqlalchemy.orm.dynamic.AppenderBaseQuery object at 0x0000000003B7C080>
>>> p.tag_set.all()
[<f_sqlalchemy.Tag object at 0x0000000003B7CF28>, <f_sqlalchemy.Tag object at 0x0000000003B7CF98>, <f_sqlalchemy.Tag object at 0x0000000003B7CEB8>, <f_sqlalchemy.Tag object at 0x0000000003B7CE80>]
>>> t = Tag.query.all()[1]
>>> t
<f_sqlalchemy.Tag object at 0x0000000003B7CF28>
>>> t.content
u'python'
>>> t.posts
[<Post u'hello python'>]
>>> t.post_set
<sqlalchemy.orm.dynamic.AppenderBaseQuery object at 0x0000000003B7C358>
>>> t.post_set.all()
[<Post u'hello python'>]

self one to one

自身一对一也是常用的需求，比如无限分级栏目

class Category(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50))
    # 父级 id
    pid = db.Column(db.Integer, db.ForeignKey('category.id'))
    # 父栏目对象
    pcategory = db.relationship('Category', uselist=False, remote_side=[id], backref=db.backref('scategory', uselist=False))

    def __init__(self, name, pcategory=None):
        self.name = name
        self.pcategory = pcategory

    def __repr__(self):
        return '<Category %r>' % self.name

    def save(self):
        db.session.add(self)
        db.session.commit()

查询

>>> p = Category('Python')
>>> p
<Category 'Python'>
>>> p.pid
>>> p.pcategory  # 查询父栏目
>>> p.scategory  # 查询子栏目
>>> f = Category('Flask', p)
>>> f.save()
>>> f
<Category u'Flask'>
>>> f.pid
1L
>>> f.pcategory  # 查询父栏目
<Category u'Python'>
>>> f.scategory  # 查询父栏目
>>> p.scategory  # 查询子栏目
<Category u'Flask'>

关于 flask-sqlalchemy 定义models的简单应用就这么多，更多的技巧在于如何查询。

完整代码下载

网友评论

真是明明:还是按照书上写的吧你这个和书的有差别敲出来报错一堆
人世间:这是一篇很早的笔记，使用方式主要是依据官方的文档和源代码。刚看了一眼，已经2.0版本了，当时还是0.1x的版本。不能保证至于兼容性。至于是否有类似是书，本人不是很了解。建议查看官网文档。
真是明明:提示错误：sqlalchemy.exc.NoForeignKeysError: Could not determine join condition between parent/child tables on relationship Category.category_attribute_id - there are no foreign keys linking these tables via secondary table 'category_attribute_register'. Ensure that referencing columns are associated with a ForeignKey or ForeignKeyConstraint, or specify 'primaryjoin' and 'secondaryjoin' expressions.
ccccmatser:感谢，之前被一个问题折腾了几天，看到这里找到了解决方案
饥渴计科极客杰铿:为啥我存在数据库的里的中文都是问号
人世间:@饥渴计科极客杰铿数据库的编码不对吧
执放:请问一对多的例子中，post_set 是哪里来的？没看懂它是什么意思
lukexw:@执放
仔细看：
class Post(db.Model):
.....
category = db.relationship('Category', backref=db.backref('post_set', lazy='dynamic'))
里面的db.backref,这一句会向类Category建立一个反向的引用post_set，所以Category类的实例就可以调用post_set这个引用了
lukexw:题主，在many-to-many的时候，不要创建两个backref,
tags = db.relationship('Tag', secondary=post_tag,backref=db.backref('post_set', lazy='dynamic'))
posts = db.relationship('Post', secondary=post_tag,backref=db.backref('tag_set', lazy='dynamic'))
否则你根本删除不掉文章，或者tag。会一直报错StaleDataError: DELETE statement on table 'post_tag' expected to delete 1 row(s); Only 0 were matched.
别问我怎么知道的。。。
fad8a9dafcc2:@lukexw 直接一个反向引用就可以了是吗

flask-sqlalchemy 简单笔记