ssl 如何使用FastAPI正确连接到Azure MySQL数据库

j8yoct9x 于 4个月前发布在 Mysql

关注(0)|答案(1)|浏览(65)

我正在尝试连接到我的MySQL Azure数据库，我的端点是用FastAPI编写的。当我单独运行我的连接方法时，连接正在工作，但一旦集成到端点中，我就会收到错误：

Failed to connect to the database: (mysql.connector.errors.InterfaceError) 2055: Lost connection to MySQL server at 'cluster_name.mysql.database.azure.com:port', system error: 1 [SSL: NO_CIPHERS_AVAILABLE] no ciphers available (_ssl.c:1006)

字符串
连接到数据库的代码看起来像这样：

def create_db_engine():
    try:
        # Format the connection string
        connection_string = f"mysql+mysqlconnector://{user}:{password}@{host}/{database}"
        ssl_args = {'ssl_ca': "DigiCertGlobalRootCA.crt.pem"}

        # Create the SQLAlchemy engine
        engine = create_engine(connection_string, connect_args=ssl_args)

        # Test the connection
        connection = engine.connect()
        print("Successfully connected to the database.")
        connection.close()

        return engine

    except exc.SQLAlchemyError as e:
        print(f"Failed to connect to the database: {e}")
        return None

engine = create_db_engine()
#conn = engine.connect()
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

型
当我从python单独运行这个方法时，它可以正确连接。我的API定义如下：

# Dependency
def get_db():
    db = db_init.SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.post("/users/")
def create_user(user: schema.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)

@app.post("/assistant_chat/")
def create_chat_message(chat_message: schema.ChatMessageCreate, db: Session = Depends(get_db)):
    return crud.create_chat_message(db=db, chat_message=chat_message)

型
当我使用uvicorn和uvicorn API：app --reload运行此服务器时，我收到以下错误消息：Failed to connect to the database：
(mysql.connector.errors.InterfaceError) 2055: Lost connection to MySQL server at 'cluster_name.mysql.database.azure.com:port', system error: 1 [SSL: NO_CIPHERS_AVAILABLE] no ciphers available (_ssl.c:1006)个

ssl

来源：https://stackoverflow.com/questions/77601826/how-to-properly-connect-to-an-azure-mysql-database-with-fastapi

1条答案

按热度按时间

jw5wzhpr1#

我尝试了你的代码，做了一些小的修改，我能够连接Azure MySql数据库并启用SSL。
请确保安装mysql-connector-python

main.py：

from fastapi import Depends, HTTPException
from sqlalchemy.orm import Session
import uvicorn
from . import app, database, crud, models, schemas

def get_db():
   db = database.SessionLocal()
   try:
       yield db
   finally:
       db.close()

@app.post("/users/")
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
   db_user = crud.get_user_by_email(db, email=user.email)
   if db_user:
       raise HTTPException(status_code=400, detail="Email already registered")
   return crud.create_user(db=db, user=user)

@app.post("/assistant_chat/")
def create_chat_message(chat_message: schemas.ChatMessageCreate, db: Session = Depends(get_db)):
   return crud.create_chat_message(db=db, chat_message=chat_message)

if __name__ == "__main__":
   uvicorn.run(app, host="127.0.0.1", port=8000, log_level="info")

字符串
这是我连接Azure Mysql数据库的连接代码

# app/database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "mysql+mysqlconnector://UserName:Password@HostName:port/databaseName"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, 
    connect_args={"ssl_ca": "DigiCertGlobalRootCA.crt(1).pem"}
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

型
我使用uvicorn运行了以下命令：