注:这篇文章核心是介绍 python 中异步使用 mysql 涉及到的文件会比较多，但我保证你看完后能快速使用在自己的项目中来提升并发读写性能。
1 背景介绍
在平时的开发中你是否会收到来自不同方的抱怨？ eg1: 运营 1:"哎，我们的后台刷新一个页面怎么就这么慢呀，让它快点呀，老是在这里转圈圈，我......" 你:"哦，知道啦。我再看看"。此时的你也在思考，心里想:我设计的表也没问题了呀，也加索引了呀，可就还是慢，我要不再加大一下线程池？要不加一下 cache? eg2: 领导 1:"小王呀，我看线上监控，你那边负责的项目中读库的性能很差呀，分析一下拿出一个优化方案" 你:"好的，领导我知道了，我追踪一下"，虽硬着头皮应下来了不过此时你内心在想:"领导，我看了半天也是头大，没办法再优化了"。 ...... 笔者在与面试者沟通的过程中会经常问一个问题:"现在页面发现加载数据很慢，若是你会从那些方面去思考和解决这方面的问题呢？" 能很系统的回答这个问题的面试者寥寥无几，本篇我们不是系统的介绍如何进行优化系统，只是介绍一种提高并发的一种解决方案(业务读写数据层面)
2 基础必备知识
2.1 databases 简单介绍
首先，我们要介绍一下 databases 模块，我们先看一下官方介绍: 1.Databases gives you simple asyncio support for a range of databases. 2.Databases is suitable for integrating against any async Web framework, such as Starlette, Sanic, Responder, Quart, aiohttp, Tornado, or FastAPI. 3.Requirements: Python 3.6
一句话总结:在 fastapi 中 我们可以通过 Databases 来达到异步操作数据库的目的
2.2 databases 安装
代码语言：javascript复制$ pip install databases
You can install the required database drivers with:
$ pip install databases[postgresql]
$ pip install databases[mysql]
$ pip install databases[sqlite]2.3 databases 案例
代码语言：javascript复制
# install
# $ pip install databases[sqlite]
# Create a database instance, and connect to it.
from databases import Database
database = Database('sqlite:///example.db')
await database.connect()
# Create a table.
query = """CREATE TABLE HighScores (id INTEGER PRIMARY KEY, name VARCHAR(100), score INTEGER)"""
await database.execute(query=query)
# Insert some data.
query = "INSERT INTO HighScores(name, score) VALUES (:name, :score)"
values = [
{"name": "Daisy", "score": 92},
{"name": "Neil", "score": 87},
{"name": "Carol", "score": 43},
]
await database.execute_many(query=query, values=values)
# Run a database query.
query = "SELECT * FROM HighScores"
rows = await database.fetch_all(query=query)
print('High Scores:', rows)
我们可以看到在操作数据库的过程中无论是连接数据库还是进行 CURD 都是加入了关键字 await ，我们知道在 python3.6 之后 通过官方提供的 async/await 来处理异步的解决方案，so 我们就知道了 可以使用 databases 来达到异步处理数据的目的
3 fastapi 中引入 databases
3.1 配置数据库的基础信息
首先我们需要在进行配置数据库的基础信息，包括连接的域名、端口、用户名、密码等等。
代码语言：javascript复制# 内容放置在 settings.py 文件中
DATABASES = {
'haishiniu_database': {
'url': 'mysql://user1:password1@127.0.0.1:3306/haishiniu_test',
'host': '127.0.0.1',
'port': 3306,
'user': 'user1',
'password': 'password1',
'db': 'haishiniu',
'options': {
'init_command': 'SET foreign_key_checks = 0;',
'charset': 'utf8mb4',
'min_size': 1,
'max_size': 8,
'pool_recycle': 100,
}
}
}
3.2 基于配置生成 mysql 数据库的连接句柄
代码语言：javascript复制# 放置在 connector.py 文件中
import databases
import settings
from typing import Any, Dict
haishiniu_db_settings: Dict[str, Any] = settings.DATABASES["haishiniu_database"]
db_mysql_haishiniu = databases.Database(
url=haishiniu_db_settings["url"], **haishiniu_db_settings["options"]
)这样就可以得到一个数据库的连接/操作 句柄:db_mysql_haishiniu
3.3 在 fastapi 中使用 databases
完成配置、连接之后就可以进行对数据库进行访问操作
代码语言：javascript复制# 配置在 get_table_info.py 中
from dataclasses import dataclass
@dataclass(frozen=True)
class HaishiniuSQL(object):
# 获取用户信息sql
get_user_info_sql = "
select user_name, age, sex from user where user_id =:user_id
"
haishiniu_sql = HaishiniuSQL()
封装完成 sql 语句之后我们再在主逻辑中引用其完成相应的功能
代码语言：javascript复制
from fastapi import FastAPI
from connector import db_mysql_haishiniu
from get_table_info import haishiniu_sql
app = FastAPI()
@app.get("/get_user_info/")
async def read_user_info(*, user_id: str = None):
user_info = await db_mysql_haishiniu.fetch_all(
haishiniu_sql.get_user_info_sql,
values={"user_id": user_id},
)
return {"user_info": user_info}
以上就完成了一个从配置到最终获取数据的完整案例，有兴趣的小伙伴可以实践起来，对此方案的性能有疑惑的小伙伴建议做一下性能测试。若您不清楚如何验证其性能，可以在文章下方留言哦，我看需求多的话可以带大家一起做一次性能测试。
4 总结
1.介绍了 fastapi 中 如何进行异步的操作常用的数据库 2.通过一个例子完整的展示了如何在生产环境中去基于 databases 完成异步操作数据库
我坚信：思考问题的方法远大于具体解决问题的方案，让我们继续一路前行，下期再见！
