Flask中ORM的使用方法

• 原文 | 大纲 | 首页

在Flask中，ORM（对象关系映射）是一种将数据库表与Python对象进行映射的技术。它允许我们使用面向对象的方式来操作数据库，而不需要直接编写SQL语句。在本篇博客中，我将详细介绍Flask中ORM的使用方法，并提供一些示例代码。

1. 安装依赖

首先，我们需要安装Flask和ORM库的依赖。Flask提供了多个ORM库的选择，例如SQLAlchemy、Peewee和SQLObject等。在本文中，我们将以SQLAlchemy为例进行说明。

使用以下命令来安装所需的依赖：

pip install flask
pip install flask_sqlalchemy

2. 配置数据库连接

在Flask应用程序中使用ORM之前，我们需要配置数据库连接。通常，你需要指定数据库的URL，其中包括数据库类型、用户名、密码、主机和端口等信息。这些配置项可以存储在配置文件中，也可以直接在应用程序代码中硬编码。

以下是一个示例配置文件config.py的内容：

# config.py

SQLALCHEMY_DATABASE_URI = 'sqlite:///mydatabase.db'
SQLALCHEMY_TRACK_MODIFICATIONS = False

在上面的示例中，我们使用SQLite数据库，并将数据库文件存储为mydatabase.db。

3. 创建数据库模型

在Flask中使用ORM，我们需要定义数据库模型。模型代表了数据库中的表，每个模型类对应一个表。我们可以在模型类中定义属性和方法，以便与数据库进行交互。

以下是一个示例模型的代码：

# models.py

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)

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

在上面的示例中，我们定义了一个名为User的模型类，它具有id、username和email等属性。db.Model是所有模型类的基类。

4. 初始化应用程序和数据库

在应用程序的入口文件中，我们需要初始化Flask应用程序和数据库。我们还需要将数据库与应用程序关联起来。

以下是一个示例应用程序的代码：

# app.py

from flask import Flask
from config import SQLALCHEMY_DATABASE_URI
from models import db

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = SQLALCHEMY_DATABASE_URI

db.init_app(app)

@app.route('/')
def hello():
    return 'Hello, World!'

if __name__ == '__main__':
    app.run()

在上面的示例中，我们从config.py导入了数据库URL，并将其配置给应用程序。然后，我们通过调用db.init_app(app)将数据库与应用程序关联起来。

5. 使用ORM进行数据库操作

使用ORM进行数据库操作非常简单。我们可以通过创建模型类的实例来表示数据库中的记录，并使用模型类的方法来操作这些记录。

以下是一些常见的数据库操作示例：

添加记录

from models import db, User

user = User(username='john', email='john@example.com')
db.session.add(user)
db.session.commit()

在上面的示例中，我们创建一个User对象，并使用db.session.add()将其添加到会话中。最后，我们使用db.session.commit()提交会话，将记录保存到数据库中。

查询记录

from models import User

users = User.query.all()
for user in users:
    print(user.username, user.email)

在上面的示例中，我们使用User.query.all()查询所有的User记录，并遍历打印每个用户的用户名和邮箱。

更新记录

from models import db, User

user = User.query.filter_by(username='john').first()
user.email = 'new_email@example.com'
db.session.commit()

在上面的示例中，我们使用code>User.query.filter_by()过滤出用户名为'john'的用户，并将其邮箱更新为'new_email@example.com'。然后，我们使用<code< a="">db.session.commit()提交更改。</code<>

删除记录

from models import db, User

user = User.query.filter_by(username='john').first()
db.session.delete(user)
db.session.commit()

在上面的示例中，我们使用User.query.filter_by()过滤出用户名为'john'的用户，并使用db.session.delete()删除该用户。最后，我们使用db.session.commit()提交更改。

注意：本文示例代码仅用于演示目的，实际使用时请根据你的需求进行适当修改和调整。

5.更多sqlalchemy查询语句

1.带条件查询

rows = session.query(User).filter_by(username='jingqi').all()
print(rows)
rows1 = session.query(User).filter(User.username=='jingqi').all()
print(rows1)
rows2 = session.query(User.username).filter(User.username=='jingqi').all()
print(rows2)
rows3 = session.query(User.username).filter(User.username=='jingqi')
print(rows3)

filter_by和filter都是过滤条件，只是用法有区别filter_by里面不能用!= 还有> < 等等，所有filter用得更多,filter_by只能用=。

前两个查询的是User,所以返回结果也是一个对象，但是rows2查询的是属性值，所以返回的是属性值。

rows3可以看到SQLAlchemy 转成的SQL语句，SQLAlchemy最后都是会转成SQL语句，通过这个方法可以查看原生SQL,甚至有些时候我们需要把SQLAlchemy转成的SQL交给DBA审查，合适的才能使用。

2.获取结果

print( session.query(User).filter(User.username=='jingqi').all() )
print( session.query(User).filter(User.username=='jingqi').first())
print( session.query(User).filter(User.username=='jingqi').one())#结果为一个时正常，多了就报错
print( session.query(User).get(2))#通过id查询

all() 返回查询到的所有的结果。这个方法比较危险的地方是，如果数据量大且没有使用limit子句限制的话，所有的结果都会加载到内存中。它返回的是一个列表，如果查询不到任何结果，返回的是空列表。

first() 返回查询到的第一个结果，如果没有查询到结果，返回None。

scalar() 这个方法与.one_or_none()的效果一样。 如果查询到很多结果，抛出sqlalchemy.orm.exc.MultipleResultsFound异常。如果只有一个结果，返回它，没有结果返回None。

one() 如果只能查询到一个结果，返回它，否则抛出异常。没有结果时抛sqlalchemy.orm.exc.NoResultFound，有超过一个结果时抛sqlalchemy.orm.exc.MultipleResultsFound。

one_or_none()比起one()来，区别只是查询不到任何结果时不再抛出异常而是返回None。

get()这是个比较特殊的方法。它用于根据主键来返回查询结果，因此它有个参数就是要查询的对象的主键。如果没有该主键的结果返回None，否则返回这个结果。

3.limit 和 order by

#限制查询返回结果
print( session.query(User).filter(User.username!='jingqi').limit(2).all())
print( session.query(User).filter(User.username!='jingqi').offset(2).all())
print( session.query(User).filter(User.username!='jingqi').slice(2,3).all())

#可以排序之后再进行限制
from sqlalchemy import desc
print( session.query(User).filter(User.username!='budong').order_by(User.username).all())
print( session.query(User).filter(User.username!='budong').order_by(desc(User.username)).slice(1,3).all())

4.条件查询

#不等于
print( session.query(User).filter(User.username!='jingqi').all() )
#模糊匹配 like
print( session.query(User).filter(User.username.like('jingqi')).all() )
print( session.query(User).filter(User.username.notlike('jingqi')).all() )
#成员属于  in_
print( session.query(User).filter(User.username.in_(['jingqi','jingqi1'])).all() )
#成员不属于  notin_
print( session.query(User).filter(User.username.notin_(['jingqi','jingqi2'])).all() )
#空判断
print( session.query(User).filter(User.username==None).all() )
print( session.query(User).filter(User.username.is_(None)).all() )
print( session.query(User).filter(User.username.isnot(None)).all() )
#多条件
print( session.query(User).filter(User.username.isnot(None),User.password=='qwe123').all() )
#选择条件
from sqlalchemy import or_,and_,all_,any_
print( session.query(User).filter(or_(User.username=='jingqi',User.password=='qwe123')).all() )
print( session.query(User).filter(and_(User.username=='jingqi2',User.password=='111')).all() )

5.常用聚合

from sqlalchemy import func,extract
print( session.query(User.password,func.count(User.id)).group_by(User.password).all() )
print( session.query(User.password,func.count(User.id)).group_by(User.password).having(func.count(User.id)>1).all() )
print( session.query(User.password,func.sum(User.id)).group_by(User.password).all() )
print( session.query(User.password,func.max(User.id)).group_by(User.password).all() )
print( session.query(User.password,func.min(User.id)).group_by(User.password).all() )
#使用extract提取时间中的分钟或者天来分组
print( session.query(extract('minute', User.creatime).label('minute'),func.count('*').label('count')).group_by('minute').all() )
print( session.query(extract('day', User.creatime).label('day'),func.count('*').label('count')).group_by('day').all() )

6.其他

# session = db.session

#简单查询
print(session.query(User).all())
print(session.query(User.name, User.fullname).all())
print(session.query(User, User.name).all())

#带条件查询
print(session.query(User).filter_by(name='user1').all())
print(session.query(User).filter(User.name == "user").all())
print(session.query(User).filter(User.name.like("user%")).all())
session.query(Article).filter(Article.record_time.between(yesterday, today)).filter_by(company_id=1).count()

#多条件查询
print(session.query(User).filter(and_(User.name.like("user%"), User.fullname.like("first%"))).all())
print(session.query(User).filter(or_(User.name.like("user%"), User.password != None)).all())

#关联查询 
print(session.query(User, Address).filter(User.id == Address.user_id).all())
print(session.query(User).join(User.addresses).all())
print(session.query(User).outerjoin(User.addresses).all())

#聚合查询
print(session.query(User.name, func.count('*').label("user_count")).group_by(User.name).all())
print(session.query(User.name, func.sum(User.id).label("user_id_sum")).group_by(User.name).all())

#子查询
stmt = session.query(Address.user_id, func.count('*').label("address_count")).group_by(Address.user_id).subquery()
print(session.query(User, stmt.c.address_count).outerjoin((stmt, User.id == stmt.c.user_id)).order_by(User.id).all())

#exists
print(session.query(User).filter(exists().where(Address.user_id == User.id)))
print(session.query(User).filter(User.addresses.any()))

#限制返回字段查询
person = session.query(Person.name, Person.created_at,                     
         Person.updated_at).filter_by(name="zhongwei").order_by(            
         Person.created_at).first()

参考: https://blog.csdn.net/xuefeng_210/article/details/123024428