FastAPI项目使用databases库连接MySQL的URL问题求助

问题描述 #

FastAPI项目使用databases库连接MySQL时，构造的数据库URL导致端口解析错误，报错信息：

ValueError: Port could not be cast to integer value as 'PASSWORD'

需求：要么修复URL构造问题，要么提供不使用URL的替代连接方式。

错误原因 #

问题出在数据库密码（或用户名）包含特殊字符（比如@、:、/等），直接拼接URL时会破坏URL的结构规则，导致URL解析器错误地将密码片段识别为端口部分，从而抛出端口无法转为整数的错误。

解决方案 #

方案1：修复URL构造（编码特殊字符） #

使用urllib.parse.quote_plus对用户名和密码进行URL编码，确保特殊字符不会干扰URL的解析。修改后的代码如下：

from databases import Database
import os
from urllib.parse import quote_plus

# Database configuration
db_config = {
    "host": os.environ.get("DATABASE_HOSTNAME"),
    # 对用户名和密码进行URL编码
    "user": quote_plus(os.environ.get("DATABASE_USERNAME")),
    "password": quote_plus(os.environ.get("DATABASE_PASSWORD")),
    "database": os.environ.get("DATABASE_DBNAME"),
    "port": int(os.environ.get("DATABASE_PORT")),
}

# 构造正确的数据库URL
database_url = f"mysql+aiomysql://{db_config['user']}:{db_config['password']}@{db_config['host']}:{db_config['port']}/{db_config['database']}"

# Connect to the database
db = Database(database_url)


async def connect():
    await db.connect()


async def disconnect():
    await db.disconnect()

方案2：不使用URL，直接用参数初始化Database #

databases库支持直接传入连接参数，无需构造URL，彻底避免URL解析问题。代码示例：

from databases import Database
import os

# Database configuration
db_config = {
    "host": os.environ.get("DATABASE_HOSTNAME"),
    "user": os.environ.get("DATABASE_USERNAME"),
    "password": os.environ.get("DATABASE_PASSWORD"),
    "database": os.environ.get("DATABASE_DBNAME"),
    "port": int(os.environ.get("DATABASE_PORT")),
}

# 直接传入连接参数，无需构造URL
db = Database(
    url=None,
    dialect="mysql+aiomysql",
    host=db_config["host"],
    port=db_config["port"],
    username=db_config["user"],
    password=db_config["password"],
    database=db_config["database"],
)


async def connect():
    await db.connect()


async def disconnect():
    await db.disconnect()

说明 #

• 方案1适合习惯使用URL连接的场景，编码后能兼容所有特殊字符；
• 方案2更直观，无需处理URL编码逻辑，尤其适合密码包含特殊字符的情况；
• 两种方案都能兼容原有的startup和shutdown事件代码，无需修改启动逻辑。

内容的提问来源于stack exchange，提问作者oscar dev