likes
comments
collection
share

构建Flask接口:整合蓝图、SQLAlchemy与多参数查询

作者站长头像
站长
· 阅读数 29

首先,你需要确保已经安装了必要的库,如Flask、Flask-SQLAlchemy和Flask-Blueprint。如果没有,可以使用pip安装:

pip install Flask Flask-SQLAlchemy Flask-Blueprint

然后,你可以按照以下步骤编写代码:

  1. 设置Flask应用
  2. 配置SQLAlchemy
  3. 定义模型
  4. 创建蓝图
  5. 编写路由和视图函数
  6. 运行应用

下面是代码示例:

from flask import Flask, Blueprint, request, jsonify  
from flask_sqlalchemy import SQLAlchemy  
  
# 创建Flask应用实例  
app = Flask(__name__)  
  
# 配置SQLAlchemy(这里只是一个示例,你可能需要在配置文件中设置这些值)  
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///example.db'  
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False  
  
# 初始化SQLAlchemy  
db = SQLAlchemy(app)  
  
# 定义模型(例如:User)  
class User(db.Model):  
    id = db.Column(db.Integer, primary_key=True)  
    name = db.Column(db.String(80), nullable=False)  
    age = db.Column(db.Integer, nullable=False)  
  
# 创建蓝图  
user_blueprint = Blueprint('user', __name__)  
  
# 路由和视图函数(查询示例)  
@user_blueprint.route('/users', methods=['GET'])  
def get_users():  
    # 从请求中获取参数(例如:name和age)  
    name = request.args.get('name')  
    age = request.args.get('age', type=int)  # 假设age是整数,并提供默认值(如果需要)  
  
    # 根据参数查询数据库  
    query = User.query  
    if name:  
        query = query.filter_by(name=name)  
    if age:  
        query = query.filter_by(age=age)  
  
    # 获取查询结果并返回  
    users = query.all()  
    return jsonify([user.to_dict() for user in users])  
  
# 将蓝图注册到Flask应用  
app.register_blueprint(user_blueprint, url_prefix='/api')  
  
# 定义to_dict方法(可选,用于将模型转换为字典)  
def to_dict(self):  
    return {  
        'id': self.id,  
        'name': self.name,  
        'age': self.age,  
    }  
  
# 将to_dict方法添加到User类(作为类方法)  
User.to_dict = to_dict  
  
# 运行应用(仅用于开发,生产环境请使用WSGI服务器)  
if __name__ == '__main__':  
    # 创建数据库表(如果它们还不存在)  
    db.create_all()  
    app.run(debug=True)

这个示例创建了一个简单的Flask应用,它使用蓝图来处理与用户相关的路由,并使用SQLAlchemy进行数据库操作。你可以通过/api/users?name=John&age=30这样的URL来查询数据库中的用户。

注意:在实际应用中,你可能需要处理更多的错误情况,例如验证输入、处理数据库错误等。此外,为了安全起见,你可能还需要限制哪些参数可以被查询,并考虑使用更复杂的查询逻辑(例如:使用and_or_等SQLAlchemy操作符来构建更复杂的查询)。 当然,我们可以使用create_enginesessionmakerscoped_session来更明确地管理数据库会话,而不是直接依赖Flask-SQLAlchemy的默认会话管理。以下是更新后的代码示例:

from flask import Flask, Blueprint, request, jsonify  
from sqlalchemy import create_engine, and_  
from sqlalchemy.orm import sessionmaker, scoped_session  
from sqlalchemy.ext.declarative import declarative_base  
  
# 创建Flask应用实例  
app = Flask(__name__)  
  
# 数据库配置(这里只是一个示例,你可能需要在配置文件中设置这些值)  
DATABASE_URI = 'sqlite:///example.db'  
  
# 创建数据库引擎  
engine = create_engine(DATABASE_URI, echo=True)  # echo=True 用于显示SQL语句,生产环境应设为False  
  
# 创建会话工厂  
Session = sessionmaker(bind=engine)  
  
# 使用scoped_session来创建线程安全的会话  
db_session = scoped_session(Session)  
  
# 声明基础模型类  
Base = declarative_base()  
  
# 定义模型(例如:User)  
class User(Base):  
    __tablename__ = 'users'  
    id = Base.Column(Base.Integer, primary_key=True)  
    name = Base.Column(Base.String(80), nullable=False)  
    age = Base.Column(Base.Integer, nullable=False)  
  
    def to_dict(self):  
        return {  
            'id': self.id,  
            'name': self.name,  
            'age': self.age,  
        }  
  
# 创建蓝图  
user_blueprint = Blueprint('user', __name__)  
  
# 路由和视图函数(查询示例)  
@user_blueprint.route('/users', methods=['GET'])  
def get_users():  
    # 从请求中获取参数(例如:name和age)  
    name = request.args.get('name')  
    age = request.args.get('age', type=int, default=None)  # 假设age是整数,并提供默认值(如果需要)  
  
    # 创建会话  
    session = db_session()  
  
    # 根据参数查询数据库  
    query = session.query(User)  
    if name:  
        query = query.filter_by(name=name)  
    if age is not None:  
        query = query.filter_by(age=age)  
  
    # 获取查询结果并返回  
    users = query.all()  
    results = [user.to_dict() for user in users]  
      
    # 关闭会话  
    session.close()  
  
    return jsonify(results)  
  
# 将蓝图注册到Flask应用  
app.register_blueprint(user_blueprint, url_prefix='/api')  
  
# 创建数据库表(如果它们还不存在)  
Base.metadata.create_all(engine)  
  
# 运行应用(仅用于开发,生产环境请使用WSGI服务器)  
if __name__ == '__main__':  
    app.run(debug=True)

在这个示例中,我们使用了create_engine来创建数据库引擎,sessionmaker来创建会话工厂,并使用scoped_session来确保会话在线程之间是安全的。我们定义了一个基础模型类Base,并继承了它来创建User模型。在视图函数中,我们显式地创建和关闭会话,以确保资源的正确管理。最后,我们使用Base.metadata.create_all(engine)来创建数据库表(如果它们还不存在)。

转载自:https://juejin.cn/post/7374307218387189769
评论
请登录