开始免费使用
当前位置：首页 > Python > 基于Flask构建SQL API服务：从配置到IN参数处理

基于Flask构建SQL API服务：从配置到IN参数处理

Cora
Cora
发布时间：2026-10-15| 3人浏览过
原创

本文详细介绍使用Flask框架搭建轻量级API服务的完整流程。内容涵盖数据库配置、全局DB对象定义及数据转码工具类的构建。重点演示了五种SQL查询接口的实现方式，包括无参查询、常规参数传递及复杂的IN参数处理。针对IN参数在URL中的多种传值形式（逗号分隔或多次重复键），提供了可靠的解析方案，并展示了标准化的错误处理与JSON响应返回机制，旨在为开发者提供一套可直接复用的API开发模板。

Flask API服务构建概述

Flask是一款基于Python开发的轻量级Web应用框架，因其简洁高效的特点，非常适合个人开发者或小型项目使用。本文将以此为基础，演示如何构建一个基础的API接口。

为了便于调试，本文示例采用GET请求方式。GET接口实现简单，无需上传复杂数据，只需在URL路径后指定GET方法即可返回字符串结果。

本文仅作为Flask接口开发的入门文档，内容从最基础的接口构建延伸至稍复杂的场景。后续若有时间，计划逐步补充Token鉴权、跨域资源共享（CORS）认证、蓝图应用以及日志管理等高级功能。

项目基础配置与工具类

第一步，首先在configs模块中配置数据源连接信息。

configs.py

HOST = '127.0.0.1'
PORT = '5432'
DATABASE = 'runoobdb'
USERNAME = 'postgres'
PASSWORD = '*****'
# 配置主数据库
DB_URI = "postgresql+psycopg2://{username}:{password}@{host}:{port}/{db}".format(username=USERNAME, password=PASSWORD,
                                                                                 host=HOST, port=PORT, db=DATABASE)
# SQLALCHEMY_DATABASE_URI = 'postgresql+psycopg2://postgres:*****@127.0.0.1:5432/runoobdb'
# 连接其他数据库
SQLALCHEMY_BINDS = {
    'xxxdb': 'postgresql+psycopg2://postgres:123456@localhost:5432/lincms3',
    'yyydb': 'postgresql+psycopg2://postgres:123456@localhost:5432/lincms4',
    'zzzdb': 'sqlite:///users.db'
}
SQLALCHEMY_DATABASE_URI = DB_URI
SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLALCHEMY_ECHO = True

第二步，在exts模块中定义全局数据库对象db。

exts.py

from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()

第三步，构建flaskutils工具模块，在此定义接口开发中常用的公共类。例如数据转码功能，可将数据集转换为JSON格式，或解析URL中的逗号分隔参数等。后续功能将基于此模块进行扩展。

flaskutils.py

import decimal
 
import numpy as np
import json, datetime,configparser
 
 
class DataEncoder(json.JSONEncoder):
    """数据转码类    """
    def default(self, obj):
        """针对无法转json的数据类型进行转码
        目前支持的转码类型        1、将Numpy的intger,floating转为int和float
        2、将Numpy的ndarray转为list
        3、将np.datetime64转化为字符串前10位        4、将datetime.datetime转化为"%Y-%m-%d %H:%M:%S"
        5、将datetime.date转化为"%Y-%m-%d"
        6、将bytes转化为utf-8字符串
        入参:
            obj: 数据对象
        出参:
            转化后的数据
        异常:
            无        """
        if isinstance(obj, np.integer):
            return int(obj)
        elif isinstance(obj, np.floating):
            return float(obj)
        elif isinstance(obj, np.ndarray):
            return obj.tolist()
        elif isinstance(obj, np.datetime64):
            return str(obj)[:10]
        elif isinstance(obj, datetime.datetime):
            return obj.strftime("%Y-%m-%d %H:%M:%S")
        elif isinstance(obj, datetime.date):
            return obj.strftime("%Y-%m-%d")
        elif isinstance(obj, decimal.Decimal):
            return float(obj)
        elif isinstance(obj, bytes):
            return str(obj, encoding='utf-8')
        else:
            return json.JSONEncoder.default(self, obj)
 
 
def getsqlresultjson(db, sql,params={}):
    """根据db和sql语句，将结果集转换为json格式
    根据db和sql语句，将结果集转换为json格式
    第一步：根据cursor获取元数据，生成键值列表
    第二步：遍历结果集，将键值列表和结果集组装成字典，加入列表
    第三步：将列表通过DataEncoder进行转码
    入参:
        db: 数据库实例.
        sql: 待运行的SQL语句
    出参:
        Json格式:
        举例：        {'Serak': ('Rigel VII', 'Preparer'),
          'Zim': ('Irk', 'Invader'),
         'Lrrr': ('Omicron Persei 8', 'Emperor')}
    异常:
        无    """
    resultdict = []
 
    cursor = db.session.execute(sql,params=params).cursor
    resultproxy = db.session.execute(sql,params=params).fetchall()
 
    # 获取元数据
    colname = [i[0] for i in cursor.description]
    # 获取结果集，组成字典，加入列表
    for rowproxy in resultproxy:
        rowresult = dict(zip(colname, rowproxy))
        resultdict.append(rowresult)
 
    # 生成json格式
    jsonstr = json.dumps(resultdict, cls=DataEncoder)
    return jsonstr
 
 
def parasecommaparamtolist(param):
    '''
    处理in传递参数，in传递参数可适用于两种传递方式，逗号传递参数或参数传递
    此处主要是处理,逗号传递参数，返回为list
    # http://127.0.0.1:5000/getresultbysqlgetparaminbylist?sqlid=sql10&begindate=2018&enddate=2020&kpicode=03010101
    # http://127.0.0.1:5000/getresultbysqlgetparaminbylist?sqlid=sql10&begindate=2018&enddate=2020&kpicode=03010101&kpicode=031111111
    # http://127.0.0.1:5000/getresultbysqlgetparaminbylist?sqlid=sql10&begindate=2018&enddate=2020
    # http://127.0.0.1:5000/getresultbysqlgetparaminbylist?sqlid=sql10&begindate=2018&enddate=2020&kpicode=03010101,222222222
    # http://127.0.0.1:5000/getresultbysqlgetparaminbylist?sqlid=sql10&begindate=2018&enddate=2020&kpicode=03010101&kpicode=03010101
    :param param:
    :return:
        字符串列表    '''
    result = []
    for val in param.split(','):
        if val:
            result.append(val)
    return result

第四步，在app文件中构建初始版本的Flask应用。

app.py

import configs
from exts import db
from flask import Flask
from flaskutils import *
from flask import request,jsonify
 
app = Flask(__name__)
 
# 加载配置文件
app.config.from_object(configs)
app.debug = True
 
db.init_app(app)
 
if __name__ == '__main__':
    print(app.url_map)
    app.run(host='0.0.0.0', port=8080)

SQL接口实现与IN参数处理

第五步，在app文件中配置SQL语句。原本计划尝试使用类似MyBatis的配置方式，但最终决定简化处理。主要包含三条SQL语句：第一条无需传参；第二条传递常规参数；第三条涉及IN参数。特别是IN参数的处理，网上常见方法往往不可靠，本文提供的方法经过验证，具有原创性。

sqldict={}
sqldict['sql1'] = """select a.*
from kpi_value a
where a.kpicode in ('01010101','02010101','03010101')
and a.datelevel='01'
and a.regionlevel='02'
"""
sqldict['sql2'] = """select a.*
from kpi_value a
where a.kpicode in ('01010101','02010101','03010101')
and a.datelevel='01'
and a.regionlevel='02'
and a.datecode>=:begindate and a.datecode<=:enddate
"""
sqldict['sql3'] = """select a.*
from kpi_value a
and a.datelevel='01'
and a.regionlevel='02'
and a.datecode>=:begindate and a.datecode<=:enddate
and a.kpicode in :kpicode
"""

1、构建第一个最简单的SQL返回接口，该接口无需传递SQL参数，但必须传递sqlid参数以指定查询模板。

@app.route('/getresultbysql', methods=['GET', 'POST'])
def index1():
    sqlid = request.args.get('sqlid')
    sqltext=sqldict[sqlid]
    jsonstr = getsqlresultjson(db,sqltext)
    return jsonstr, 200, {"Content-Type": "application/json"}

2、构建一个支持SQL内部传参的接口，通过字典形式传递参数。

@app.route('/getresultbysqlparam', methods=['GET', 'POST'])
def index2():
    sqlid = request.args.get('sqlid')
    sqltext=sqldict[sqlid]
    params = {"begindate": '2017',"enddate":'2019'}
    jsonstr = getsqlresultjson(db,sqltext,params)
    return jsonstr, 200, {"Content-Type": "application/json"}

3、通过URL传递SQL参数。

@app.route('/getresultbysqlgetparam', methods=['GET', 'POST'])
def index3():
    sqlid = request.args.get('sqlid')
    begindate = request.args.get('begindate')
    enddate = request.args.get('enddate')
    sqltext=sqldict[sqlid]
    params = {"begindate": begindate,"enddate":enddate}
    jsonstr = getsqlresultjson(db,sqltext,params)
    return jsonstr, 200, {"Content-Type": "application/json"}

4、通过URL传递SQL参数，但不传递IN参数，而是在路由函数内部指定IN参数值。

@app.route('/getresultbysqlgetparamin', methods=['GET', 'POST'])
def index4():
    sqlid = request.args.get('sqlid')
    sqlid='sql3'
    begindate = request.args.get('begindate')
    enddate = request.args.get('enddate')
    sqltext=sqldict[sqlid]
    incond = ['01010101',  '03010101']
    params = {"begindate": begindate,"enddate":enddate,'kpicode':tuple(incond)}
    jsonstr = getsqlresultjson(db,sqltext,params)
    return jsonstr, 200, {"Content-Type": "application/json"}

5、通过URL同时传递IN参数和普通参数。此处支持两种传值方式：一种是使用&aa=xxx&aa=yyy的形式，另一种是使用aa=xxx,yyy（逗号分隔）的形式。

@app.route('/getresultbysqlgetparaminbylist', methods=['GET', 'POST'])
def index5():
    sqlid = request.args.get('sqlid')
    sqlid='sql3'
    begindate = request.args.get('begindate')
    enddate = request.args.get('enddate')
    incond=request.args.getlist('kpicode')
 
    if len(incond) == 1 and ',' in incond[0]:
        incond = parasecommaparamtolist(incond[0])
 
    sqltext=sqldict[sqlid]
    params = {"begindate": begindate,"enddate":enddate,'kpicode':tuple(incond)}
    jsonstr = getsqlresultjson(db,sqltext,params)
 
    return jsonstr, 200, {"Content-Type": "application/json"}

标准化响应与总结

6、实现标准化的接口响应返回结果，包括状态码、数据内容及错误信息。

@app.route('/getresultbysqlgetparaminbylistresponse', methods=['GET', 'POST'])
def index6():
 
    retinfo={}
    errorflag=False
    retinfo['returncode'] = 200
    retinfo['returndata'] = ''
    retinfo['returninfo'] = '处理成果'
    sqlid = request.args.get('sqlid')
    begindate = request.args.get('begindate')
    enddate = request.args.get('enddate')
    incond = request.args.getlist('kpicode')
 
    if len(incond) == 1 and ',' in incond[0]:
        incond = parasecommaparamtolist(incond[0])
    if not incond:
        retinfo['returninfo']=retinfo['returninfo'] +'未传入KPI编码'
        errorflag=True
    if not begindate:
        retinfo['returninfo'] = retinfo['returninfo'] + '未传入开始时间'
        errorflag=True
    if not enddate:
        retinfo['returninfo'] = retinfo['returninfo'] + '未传入结束时间'
        errorflag=True
    if begindate>enddate:
        retinfo['returninfo'] = retinfo['returninfo'] + '开始时间大于结束时间'
        errorflag=True
    if errorflag==True:
        retinfo['returncode'] = 400
        response = jsonify(retinfo)
        response.status_code = 400
        return response
 
    sqltext = sqldict[sqlid]
    params = {"begindate": begindate, "enddate": enddate, 'kpicode': tuple(incond)}
    jsonstr = getsqlresultjson(db, sqltext, params)
    retinfo['returndata']  = jsonstr
    response = jsonify(retinfo)
    response.status_code = 200
    return response

至此，关于Flask搭建API服务的实现步骤介绍完毕。如需了解更多Flask API相关内容，请搜索本站历史文章或继续阅读后续教程。感谢大家的支持！

相关文章

相关标签: Python Flask API开发 后端技术 JSON

免责声明：本文内容仅供参考，不代表本站观点。转载或引用请注明出处。

作者最新文章

全部文章

Python 相关产品和应用

全部下载
PythonSelenium
Selenium

Selenium

4.51.0 · Windows

用于浏览器自动化和 Web 测试的 Python 工具。

Python机器学习
scikit-learn

scikit-learn

1.9.1 · Windows

常用 Python 机器学习算法与模型评估工具库。

Python代码质量
Black

Black

26.10.1 · Windows

强调一致性的 Python 自动代码格式化工具。

Python代码质量
Ruff

Ruff

0.17.1 · macOS

由 Rust 编写的高速 Python 代码检查和格式化工具。

Pythonpytest
pytest

pytest

9.1.1 · Windows

简洁而强大的 Python 自动化测试框架。

PythonFastAPI
FastAPI

FastAPI

0.146.0 · Windows

高性能 Python API 框架，支持类型提示和自动接口文档。

PythonFlask
Flask

Flask

3.1.3 · Windows

轻量灵活的 Python Web 应用框架。

PythonDjango
Django

Django

6.1.2 · Windows

成熟的 Python Web 框架，适合构建内容站点和业务系统。

PythonSpyder
Spyder

Spyder

6.1.7 · Windows

面向科学计算与数据分析的 Python 集成开发环境。

PythonSciPy
SciPy

SciPy

1.18.1 · Windows

面向科学、工程和技术计算的 Python 算法库。