FastAPI 學習之路（五十三）根據環境不同連接不同數據庫

  在實際的開發過程中，我們數據庫，可以根據連接的環境不一樣，我們會拆分成不一樣的數據庫，根據我們所要用的環境來選擇對應的數據庫即可，那么我們應該如何去實現根據選擇去選擇不一樣的數據庫呢。

        首先，我們找一個配置的文件，我們這里演示 我們使用config.py去配置環境。配置如下

EVENT="test"

   接下來，我們去創建一個測試用的數據庫配置，起名：testDatabse.py

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
SQLALCHEMY_DATABASE_URL = "sqlite:///./test1.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base =   declarative_base()

  然后我們修改models.py。如下

from  config import EVENT
if EVENT=="test":
    from models.testDatabase import Base,engine

else:
    from  models.database import Base,engine
 ....數據庫設計
 Base.metadata.create_all(bind=engine)

    接下來我們去根據不同環境去連接不一樣的數據庫即可。

from models.database import *
from models.testDatabase import TestingSessionLocal
from config import EVENT
def get_test_db():
    db=TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()

def get_db_pro():
    """
    每一個請求處理完畢后會關閉當前連接，不同的請求使用不同的連接
    :return:
    """
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
if EVENT =="test":
    get_db=get_test_db
else:
    get_db=get_db_pro

    最后在連接數據庫的地方，我們把get_db替換成從

from get_db import get_db

因為我們去創建了不一樣的數據庫，根據不一樣的環境去連接的。我們寫一個測試方法，去測試下

def test_create_user():
    client = TestClient(app)
    response = client.post(
        "/user/users/",
        json={"email": "[e-mail]", "password": "leizi"},
    )
    assert response.status_code == 200
if __name__ == "__main__":
    test_create_user()

 我們去執行下看下，

 

       我們可以看到，執行完畢后數據落在了我們配置的測試環境的測試數據看中。

文章首發在公眾號，歡迎關注。