跳转至

Flask 中用 SQLAlchemy 操作数据库

约 217 个字 75 行代码 预计阅读时间 2 分钟

安装包

1
pip install flask-redis

集成

以 MySQL 为例（使用 pymysql）：

配置类
1
2
3
class FlaskPlaygroundConfig:
    SQLALCHEMY_DATABASE_URI = f"mysql+pymysql://用户名:密码@地址:端口号/数据库名?charset=utf8mb4"
    # 如密码有特殊字符，可用 urllib.parse.quote_plus('密码') 替代
ext.py
1
2
3
from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()
app.py
1
2
3
4
5
6
7
8
9
app = Flask(__name__)
app.config.from_object(FlaskPlaygroundConfig)
db.init_app(app)

# 测试连接
with app.app_context():
    with db.engine.connect() as conn:
        result = conn.execute(text('select 1;'))
        print('Database connection successful: ', result.fetchone())

定义表结构

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
class UserEntity(db.Model):
    __tablename__ = 'users'

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    create_time = db.Column(db.DateTime, default=datetime.now)
    create_user = db.Column(db.Integer, default=0)
    is_deleted = db.Column(db.Boolean, default=False)
    username = db.Column(db.String(20), unique=True)
    nickname = db.Column(db.String(20), nullable=False, default='Default Nickname')


class RouteEntity(db.Model):
    __tablename__ = 'routes'

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    create_time = db.Column(db.DateTime, default=datetime.now)
    create_user = db.Column(db.Integer, db.ForeignKey('users.id'), default=0)   # 外键，填表名
    # 关联表，此例中关联了 users 表，与定义了外键的 create_user 对应
    # UserEntity 实例可以通过 user_entity_instance.routes 属性访问自己创建的线路信息
    # RouteEntity 实例可以通过 route_entity_instance.user 属性访问 create_user 的用户信息
    user = db.relationship('UserEntity', backref=db.backref('routes'))


class RoutePointEntity(db.Model):
    __tablename__ = 'route_points'

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    create_time = db.Column(db.DateTime, default=datetime.now)
    create_user = db.Column(db.Integer, default=0)
    route_id = db.Column(db.Integer, db.ForeignKey('routes.id'), nullable=False)
    longitude = db.Column(db.Numeric(10, 7), nullable=False)
    latitude = db.Column(db.Numeric(10, 7), nullable=False)
    memo = db.Column(db.Text)

    parent_route = db.relationship('RouteEntity', backref=db.backref('points'))

如果在主程序中添加：

app.py
1
2
with app.app_context():
    db.create_all()

理论上会创建不存在的数据表。不会更改已有的数据表。

Warning

实测，似乎不会立即新增，但不影响写入内容，写入内容后就有表结构了。

CRUD

新增

1
2
3
user = UserEntity(username=username, nickname=nickname)
db.session.add(user)
db.session.commit() # 所有更改操作都要提交后才生效

此后可通过 user.id 得知新增的列的 ID。

涉及到关联表的也会一并新增：

1
2
3
4
route_points = [RoutePointEntity(...), ...]
route = RouteEntity(points = route_points, ...)
db.session.add(route)   # 会一并新增路线和对应的点
db.session.commit()

Warning

仅适用于子项较少的情况。如果子项特别多，要分开新增。参见 Flask-SQLAlchemy 批量导入数据

查询

1
2
3
4
5
6
7
8
# 查询 ID = 1 的用户信息
UserEntity.query.get(1)

# 查询给定 ID，未删除的第一条用户信息
UserEntity.query.filter_by(id=user_id, is_deleted=False).first()

# 查询给定路线 ID，未删除的全部路线点信息
db.session.query(RoutePointEntity).filter(RoutePointEntity.route_id == route_entity_bare.id, RoutePointEntity.is_deleted == False).all()

其中 UserEntity.query 可视为 db.session.query(UserEntity)。

如果没有，则返回 None。

修改

对查询到的实体直接修改属性，提交即可。

1
2
3
4
user = UserEntity.query.filter_by(id=user_id, is_deleted=False).first()
user.password = new_password
user.update_time = datetime.now()
db.session.commit()

删除

1
2
3
user = UserEntity.query.filter_by(id=user_id, is_deleted=False).first()
db.session.delete(user)
db.session.commit()

很少用，因为现在基本上要求软删除。

注意事项

  • 实体类必须有主键