Flask框架 之Flask-SQLAlchemy操作数据库

一、代码

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import desc
 
class Config(object):
    ''' sqlalchemy 参数配置 '''
 
    # 配置数据库
    SQLALCHEMY_DATABASE_URI = 'mysql://test01:123456@127.0.0.1:3306/test01?charset=utf8'
    # 自动会追踪对象的修改并且发送信号
    SQLALCHEMY_TRACK_MODIFICATIONS = True
    # 输出sql语句
    SQLALCHEMY_ECHO = True
 
app = Flask(__name__)
# 载入数据库配置
app.config.from_object(Config)
 
# 创建sqlalchemy工具对象
db = SQLAlchemy(app)
 
class User(db.Model):
    __tablename__ = 'user'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(32), unique = True)
    email = db.Column(db.String(255), unique = True)
    role_id = db.Column(db.Integer, db.ForeignKey('role.id'))
 
class Role(db.Model):
    __tablename__ = 'role'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(32), unique = True)
    user = db.relationship('User')
 
if __name__ == '__main__':
 
    # 删除所有表
    db.drop_all()
 
    # 创建所有表
    db.create_all()
 
    # 创建数据
    role1 = Role(name = '管理员')
    db.session.add(role1)
    db.session.commit()
 
    role2 = Role(name = '普通用户')
    db.session.add(role2)
    db.session.commit()
 
    u1 = User(username='admin', email='admin@example.com', role_id=role1.id)
    u2 = User(username='peter', email='peter@example.com', role_id=role2.id)
    db.session.add_all([u1, u2])
    db.session.commit()
 
    # 查询数据
    for r in db.session.query(Role).order_by(desc(Role.id)).all():
        print(r.name)
    
    u = User.query.filter_by(id=1).first()
    print(u)
 
    app.run()

二、分页器paginate

　paginate 属性：

    print(paginate.pages) #总共能生成多少页

    print(paginate.page) #当前页码数

    print(paginate.has_next) #True

    print(paginate.has_prev) #Flase

    print(paginate.next_num) #获取下一页的页码数

    print(paginate.prev_num) #获取上一页的页码数

    print(paginate.items) #获当前页的对象 列表

三、filter与filter_by的用法

模型类.query.fillter_by(字段名=值) 里面是布尔的条件   这个无法实现复杂查询

模型类.query.fillter(模型类.字段名==值) 里面是布尔的条件   【常用】

四、复杂查询

模型类.query.fillter(模型类.字段名.endswith('z')).all() 等价于 select * from user where 字段名 like '%z'

模型类.query.fillter(模型类.字段名.startswith('z')).all() 等价于 select * from user where 字段名 like 'z%'；

模型类.query.fillter(模型类.字段名.contains('z')).all() 等价于 select * from user where 字段名 like '%z%'；

模型类.query.fillter(模型类.字段名.like('%z%')).all() 等价于 select * from user where 字段名 like '%z%'；

模型类.query.fillter(模型类.字段名.in_(['a','b','c'])).all() 等价于 select * from user where 字段名  in ('a','b','c')；

模型类.query.fillter(模型类.字段名.between(开始,结束)).all() 等价于 select * from user where 字段名  between 开始 and  结束；

模型类.query.fillter(or_(模型类.字段名.like('z%'),模型类.字段名.contains('a'))).all() 等价于 select * from user where 字段名 like 'z%' or 字段名 like '%a%'；

模型类.query.fillter(and_(模型类.字段名.like('z%'),模型类.字段名 < '2021-12-12 00:00:00')).all() 等价于 select * from user where 字段名 like 'z%' and 字段名 < '2021-12-12 00:00:00'；

模型类.query.fillter(and_(模型类.字段名.like('z%'),模型类.字段名.__lt__( '2021-12-12 00:00:00'))).all() 等价于 select * from user where 字段名 like 'z%' and 字段名 < '2021-12-12 00:00:00'；

模型类.query.fillter(not_(模型类.字段名.contains('a'))).all() 等价于 select * from user where 字段名  not like '%a%' ；

模型类.query.order_by (模型类.字段名).offset(2).limit(2).all() 

>       __gt__ 

 >=     __ge__(gt equal)

<=     __le__(lt euqal)

!=      not_

五、排序

模型类.query.order_by (模型类.字段名.desc()).all() 等价于 select * from user order by 字段名 desc；

六、多对多的多条件查询

''' 标签 '''
class Tag(db.Model, Base):
    __tablename__ = "blog_tag"
 
    id = db.Column(db.Integer, primary_key=True)
    tagname = db.Column(db.String(32), unique=True)
 
article_tags = db.Table(
    "blog_article_tag",
    db.Column("tag_id", db.ForeignKey("blog_tag.id")),
    db.Column("article_id", db.ForeignKey("blog_article.id"))
)
 
''' 文章 '''
class Article(db.Model, Base):
    __tablename__ = "blog_article"
 
    id = db.Column(db.Integer, primary_key=True)
    # 发布人
    admin_id = db.Column(db.Integer, db.ForeignKey("blog_admin.id"))
    # 文章标题
    title = db.Column(db.String(32))
    # 作者
    author = db.Column(db.String(32))
    # 文章描述
    description = db.Column(db.String(255))
    # 文章关键词
    keywords = db.Column(db.String(255), default='')
    # 文章内容
    content = db.Column(db.Text)
    # 文章缩略图
    pic = db.Column(db.String(255))
    # 文章点击数
    click = db.Column(db.Integer, default = 0)
    # 是否推荐
    is_state = db.Column(db.SmallInteger, default=0)
    # 所属栏目
    category_id = db.Column(db.Integer, db.ForeignKey("blog_category.id"))
 
    tags = db.relationship("Tag", secondary    = article_tags, backref="article")

@home_search.route("/index")
def index():
    page = request.args.get('page', 1, type=int)
    tid = request.args.get('tid', 0, type=int)
    keywords = request.args.get('keywords', '', type=int)
 
    article = Article.query.join(Tag, Article.tags)
    if keywords:
        article = article.filter(Article.title.like('%'+keywords+'%'))
 
    if tid:
        article = article.filter(Tag.id==tid)
 
    try:
        pagination = article.order_by(Article.id.desc()).paginate(page = page, per_page=PAGE_LIMIT)
    except Exception as e:
        flash(str(e))
        abort(500)
 
    return render_template("home/search.html",pagination=pagination, keywords=keywords)