Python中flask-sqlalchemy能否自定义SQL语句并进行封装？

当然可以。Flask-SQLAlchemy 完全支持自定义 SQL 语句，并且有几种清晰的方式可以将其封装成可复用的组件。

最直接的方法是使用 db.session.execute() 来执行原生 SQL。你可以把相关的 SQL 语句和逻辑封装在模型类的方法或者单独的辅助函数/类里。

这里给你一个完整的代码示例，展示如何封装一个执行复杂查询的函数：

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from typing import List, Dict, Any

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///example.db'
db = SQLAlchemy(app)

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)

# 封装自定义 SQL 操作的类
class UserQueryService:
    @staticmethod
    def get_users_by_active_status(active: bool) -> List[Dict[str, Any]]:
        """
        通过原生 SQL 查询活跃/非活跃用户（假设有一个 is_active 字段，虽然模型里没定义）
        这是一个演示复杂 WHERE 子句和结果处理的例子。
        """
        # 使用参数化查询防止 SQL 注入
        sql = """
        SELECT id, username, email
        FROM user
        WHERE is_active = :active_status
        ORDER BY username
        """
        # 执行查询
        result = db.session.execute(sql, {'active_status': active})
        # 将结果转换为字典列表，便于使用
        users = [dict(row._mapping) for row in result]
        return users

    @staticmethod
    def get_user_order_summary(user_id: int) -> Dict[str, Any]:
        """
        执行一个多表 JOIN 的复杂查询，获取用户订单汇总。
        假设还有 Order 和 OrderItem 表。
        """
        sql = """
        SELECT
            u.username,
            COUNT(o.id) as total_orders,
            SUM(oi.quantity * oi.unit_price) as total_spent
        FROM user u
        LEFT JOIN `order` o ON u.id = o.user_id
        LEFT JOIN order_item oi ON o.id = oi.order_id
        WHERE u.id = :user_id
        GROUP BY u.id
        """
        result = db.session.execute(sql, {'user_id': user_id}).fetchone()
        if result:
            return dict(result._mapping)
        return {}

# 在视图或其他地方使用这个封装
@app.route('/users/active')
def get_active_users():
    service = UserQueryService()
    active_users = service.get_users_by_active_status(active=True)
    return {'users': active_users}

if __name__ == '__main__':
    with app.app_context():
        db.create_all() # 创建表用于测试
    app.run(debug=True)

关键点解释：

1. db.session.execute()：这是执行原生 SQL 的核心方法。始终使用参数化查询（如 :active_status）来传递值，这是防止 SQL 注入的关键。
2. 封装成类或函数：像 UserQueryService 这样，把相关的 SQL 操作组织在一起。这比把 SQL 字符串散落在各视图函数里要清晰、好维护得多。
3. 结果处理：execute() 返回一个 ResultProxy 对象。你可以遍历它，或者用 .fetchone()、.fetchall() 获取结果。使用 row._mapping 可以方便地将行对象转为字典。
4. 在应用中使用：封装好后，在蓝图或视图函数里调用即可，保持业务逻辑的整洁。

另一种更“SQLAlchemy”的方式是使用 text() 构造器和 db.session.query() 结合，但本质相同。对于极度复杂或数据库特定的 SQL，直接写原生语句并用上面的方式封装是最直接、可控的。

总结：用 db.session.execute() 加参数化查询，然后把逻辑包起来。