在实际的开发过程中,我们数据库,可以根据连接的环境不一样,我们会拆分成不一样的数据库,根据我们所要用的环境来选择对应的数据库即可,那么我们应该如何去实现根据选择去选择不一样的数据库呢。
首先,我们找一个配置的文件,我们这里演示 我们使用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": "test@example.com", "password": "leizi"}, ) assert response.status_code == 200 if __name__ == "__main__": test_create_user()
我们去执行下看下,
我们可以看到,执行完毕后数据落在了我们配置的测试环境的测试数据看中。