方法1：sqlalchemy

安装库
pip install sqlalchemy
pip install mysql-connector-python
驱动包如果没有安装成功会报错，无法链接！

示例代码

from sqlalchemy import create_engine,Table,MetaData
from sqlalchemy.orm import Session

metadata = MetaData()
engine = create_engine(
    'mysql+mysqlconnector://root:password@localhost:3306/my_db',
    echo = True
)
# 获取指定数据库链接对象
Admin = Table('servers',metadata,autoload=True,autoload_with=engine)
session = Session(engine)
# 返回查询数据的第一条记录
res = session.query(Admin).first()
print(res)

方法2：flask_sqlalchemy

示例代码

from flask_sqlalchemy import SQLAlchemy,make_url

def get_mysql_conn_url(config):
    """
    :description: 生成sqlalchemy使用的连接url
    :param hy_config: hy_config
    :return: url
    """
    mysql_conn_map = dict(
        dialect="mysql",
        driver="mysqlconnector",
        host=config["MYSQL_HOST"],
        port=config["MYSQL_PORT"],
        database=config["MYSQL_DB"],
        user=config["MYSQL_USER"],
        password=config["MYSQL_PASSWORD"],
    )
    s = "mysql+mysqlconnector://{user}:{password}@{host}:{port}/{database}".format(**mysql_conn_map)
    url = make_url(s)
    return url

def create_mysql_ORM(app):
    """
    创建MySQL的ORM对象并反射数据库中已存在的表，获取所有存在的表对象
    :param app: app:flask实例
    :return: (db:orm-obj, all_table:数据库中所有已存在的表的对象(dict))
    """
    # 创建mysql连接对象
    url = get_mysql_conn_url(config=app.config)
    app.config["SQLALCHEMY_DATABASE_URI"] = url
    app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = True  # 每次请求结束时自动commit数据库修改
    app.config["SQLALCHEMY_ECHO"] = False   # 如果设置成 True，SQLAlchemy将会记录所有发到标准输出(stderr)的语句,这对调试很有帮助.
    app.config["SQLALCHEMY_RECORD_QUERIES"] = None  # 可以用于显式地禁用或者启用查询记录。查询记录 在调试或者测试模式下自动启用。
    app.config["SQLALCHEMY_POOL_SIZE"] = 5  # 数据库连接池的大小。默认是数据库引擎的默认值(通常是 5)。
    app.config["SQLALCHEMY_POOL_TIMEOUT"] = 10  # 指定数据库连接池的超时时间。默认是 10。
    """
    自动回收连接的秒数。
    这对 MySQL 是必须的，默认 情况下 MySQL 会自动移除闲置 8 小时或者以上的连接。 
    需要注意地是如果使用 MySQL 的话， Flask-SQLAlchemy 会自动地设置这个值为 2 小时。
    """
    app.config["SQLALCHEMY_POOL_RECYCLE"] = None
    """
    控制在连接池达到最大值后可以创建的连接数。
    当这些额外的 连接回收到连接池后将会被断开和抛弃。
    """
    app.config["SQLALCHEMY_MAX_OVERFLOW"] = None
    # 获取SQLAlchemy实例对象
    db = SQLAlchemy(app)

    # 反射数据库中已存在的表，并获取所有存在的表对象。
    db.reflect()
    all_table = {table_obj.name: table_obj for table_obj in db.get_tables_for_bind()}

    return db, all_table

实例化，载入config配置信息

app = Flask(__name__)
app.config.from_object('config.BaseConfig')
db, all_table = create_mysql_ORM(app=app)

查询数据

@app.route('/query',methods=['POST'])
def query():
    form_dict = request.form
    name = form_dict['name']
    limit = form_dict['limit']
    page = form_dict['page']

    data = {"code": 0,"msg": "Successful",}
    try:
        table = all_table[name]
        _count = db.session.query(db.func.count(table.c.host)).scalar()
        # result = db.session.query(table).all()
        result = db.session.query(table).filter(table.c.address.like('%上海%')).limit(limit).offset((int(page) - 1))
        data['count'] = _count
        data['data'] = [x._asdict() for x in result]
    except Exception as err:
        data['code'] = 1
        data['msg'] = 'Failed! {}'.format(err)

    return jsonify(data)

返回结果