Python Web 数据库集成 Flask开发实现与MSSQL数据库高效对接，flask mssql

Python Web开发实战：用Flask高效对接MSSQL数据库

场景引入：当电商后台遇上老牌数据库

"这需求改不了！" 听到后端同事第N次拒绝修改订单查询功能时，作为产品经理的小李感到无比沮丧，他们公司电商平台的后台数据库是微软的SQL Server（MSSQL），而现有的Java系统过于笨重，每次业务调整都要等上两三周，直到某天，团队里新来的Python工程师小王提议："要不我们用Flask重构这部分？对接MSSQL其实很简单。"

一周后,当小李看到新的订单查询接口响应时间从原来的2秒降到200毫秒时，她终于明白为什么现在这么多企业选择Python+Flask来构建轻量级Web服务了。

环境准备：搭建开发基础

在开始之前,我们需要准备好开发环境，假设你已经安装了Python 3.8+（截至2025年8月，Python 3.10是主流稳定版本），下面是其他必要的准备工作：

# 创建虚拟环境（推荐）
python -m venv flask_mssql_env
source flask_mssql_env/bin/activate  # Linux/Mac
flask_mssql_env\Scripts\activate    # Windows
# 安装核心依赖
pip install flask pyodbc flask-sqlalchemy

对于MSSQL连接,Windows系统通常已经自带ODBC驱动，而Linux/Mac需要额外安装：

# Ubuntu/Debian
sudo apt-get install unixodbc unixodbc-dev freetds-dev tdsodbc
# CentOS/RHEL
sudo yum install unixODBC unixODBC-devel freetds freetds-devel

配置数据库连接

Flask连接MSSQL主要有两种方式：直接使用pyodbc，或者通过SQLAlchemy ORM，我们先看基础配置：

# config.py
class Config:
    MSSQL_SERVER = 'your_server.database.windows.net'  # 或本地服务器地址
    MSSQL_DATABASE = 'your_database'
    MSSQL_USERNAME = 'your_username'
    MSSQL_PASSWORD = 'your_password'
    SQLALCHEMY_DATABASE_URI = f"mssql+pyodbc://{MSSQL_USERNAME}:{MSSQL_PASSWORD}@{MSSQL_SERVER}/{MSSQL_DATABASE}?driver=ODBC+Driver+17+for+SQL+Server"
    SQLALCHEMY_TRACK_MODIFICATIONS = False

注意这里的ODBC Driver 17 for SQL Server是驱动名称，根据你的实际环境可能需要调整。

基础集成：从零开始构建Flask应用

让我们创建一个完整的商品管理模块示例：

# app.py
from flask import Flask, jsonify, request
from flask_sqlalchemy import SQLAlchemy
from config import Config
app = Flask(__name__)
app.config.from_object(Config)
db = SQLAlchemy(app)
# 定义商品模型
class Product(db.Model):
    __tablename__ = 'products'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(100), nullable=False)
    price = db.Column(db.Float, nullable=False)
    stock = db.Column(db.Integer, default=0)
    def to_dict(self):
        return {
            'id': self.id,
            'name': self.name,
            'price': self.price,
            'stock': self.stock
        }
@app.route('/products', methods=['GET'])
def get_products():
    products = Product.query.all()
    return jsonify([product.to_dict() for product in products])
@app.route('/products', methods=['POST'])
def add_product():
    data = request.get_json()
    new_product = Product(
        name=data['name'],
        price=data['price'],
        stock=data.get('stock', 0)
    )
    db.session.add(new_product)
    db.session.commit()
    return jsonify(new_product.to_dict()), 201
if __name__ == '__main__':
    app.run(debug=True)

性能优化：让MSSQL查询飞起来

直接使用ORM虽然方便,但在处理复杂查询或大数据量时可能会遇到性能瓶颈，以下是几个实战优化技巧：

原生SQL执行

对于复杂查询,可以直接执行原生SQL：

@app.route('/products/expensive', methods=['GET'])
def get_expensive_products():
    sql = "SELECT * FROM products WHERE price > 100 ORDER BY price DESC"
    result = db.session.execute(sql)
    products = [dict(row) for row in result]
    return jsonify(products)

分页查询实现

from sqlalchemy import func
@app.route('/products/paged', methods=['GET'])
def get_paged_products():
    page = request.args.get('page', 1, type=int)
    per_page = request.args.get('per_page', 10, type=int)
    pagination = Product.query.paginate(page=page, per_page=per_page)
    return jsonify({
        'items': [product.to_dict() for product in pagination.items],
        'total': pagination.total,
        'pages': pagination.pages,
        'current_page': pagination.page
    })

连接池配置

在config.py中添加：

SQLALCHEMY_ENGINE_OPTIONS = {
    'pool_size': 10,
    'max_overflow': 20,
    'pool_timeout': 30,
    'pool_recycle': 3600
}

高级技巧：存储过程与事务处理

调用MSSQL存储过程

@app.route('/products/discount', methods=['POST'])
def apply_discount():
    discount_rate = request.json.get('rate', 0.1)
    try:
        # 调用名为sp_apply_discount的存储过程
        db.session.execute(
            "EXEC sp_apply_discount :rate",
            {'rate': discount_rate}
        )
        db.session.commit()
        return jsonify({'status': 'success'})
    except Exception as e:
        db.session.rollback()
        return jsonify({'error': str(e)}), 500

事务处理示例

@app.route('/orders', methods=['POST'])
def create_order():
    try:
        data = request.json
        # 开始事务
        db.session.begin()
        # 扣减库存
        for item in data['items']:
            product = Product.query.get(item['product_id'])
            if product.stock < item['quantity']:
                raise ValueError(f"产品 {product.name} 库存不足")
            product.stock -= item['quantity']
            db.session.add(product)
        # 创建订单记录
        new_order = Order(
            user_id=data['user_id'],
            total=data['total']
        )
        db.session.add(new_order)
        db.session.commit()
        return jsonify({'order_id': new_order.id}), 201
    except Exception as e:
        db.session.rollback()
        return jsonify({'error': str(e)}), 400

常见问题排坑指南

连接超时问题

如果在云服务器上连接MSSQL出现超时,检查以下几点：

• 服务器防火墙是否开放了1433端口
• MSSQL是否配置允许远程连接
• 连接字符串中添加超时参数：

  SQLALCHEMY_DATABASE_URI += "&connect_timeout=30&login_timeout=30"

中文乱码处理

确保数据库和连接使用相同的编码,可以在连接字符串中添加：

SQLALCHEMY_DATABASE_URI += "&charset=utf8"

时区问题

MSSQL默认使用服务器时区,可能导致时间字段显示不一致：

from datetime import datetime, timezone
# 写入时转换为UTC
now = datetime.now(timezone.utc)
# 或者从配置层面解决
app.config['SQLALCHEMY_ENGINE_OPTIONS'] = {
    'connect_args': {
        'timezone': 'utc'
    }
}

安全最佳实践

1. 永远不要硬编码凭据：使用环境变量或配置管理工具

   import os
   app.config['MSSQL_PASSWORD'] = os.getenv('DB_PASSWORD')

2. 使用参数化查询防止SQL注入

   # 错误做法
   db.session.execute(f"SELECT * FROM users WHERE name = '{name}'")
   # 正确做法
   db.session.execute("SELECT * FROM users WHERE name = :name", {'name': name})

3. 最小权限原则：数据库用户只授予必要权限

Flask + MSSQL的现代组合

通过本文的实践,我们可以看到Flask与MSSQL的集成既保持了Python的简洁高效，又能充分利用MSSQL作为企业级数据库的强大功能，特别是在需要快速迭代的业务场景中，这种组合能够显著提升开发效率。

随着Python生态的持续发展（截至2025年8月，PyODBC已经支持MSSQL 2022的所有新特性），以及Flask框架的轻量级优势，这种技术栈正在被越来越多的企业采用，既适用于初创公司的快速原型开发，也能满足中大型企业的性能需求。

下次当你面对老旧系统改造需求时,不妨考虑用Flask+MSSQL这个"轻量级拳击手"来替代那些"笨重的相扑选手"，或许会收获意想不到的效果。