|
7 | 7 | from api.db.database import Base, get_db |
8 | 8 | from main import app |
9 | 9 |
|
10 | | -DB_TYPE = config("DB_TYPE") |
11 | | -DB_NAME = config("DB_NAME") |
12 | | -DB_USER = config("DB_USER") |
13 | | -DB_PASSWORD = config("DB_PASSWORD") |
14 | | -DB_HOST = config("DB_HOST") |
15 | | -DB_PORT = config("DB_PORT") |
16 | | -MYSQL_DRIVER = config("MYSQL_DRIVER") |
17 | | -DATABASE_URL = "" |
18 | | - |
19 | | -SQLALCHEMY_DATABASE_URL = f"{DB_TYPE}+{MYSQL_DRIVER}://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}_test" |
20 | | - |
21 | | - |
22 | | -engine = create_engine(SQLALCHEMY_DATABASE_URL) |
| 10 | +DB_TYPE = config("DB_TYPE", default="sqlite") |
| 11 | +DB_NAME = config("DB_NAME", default="test.db") |
| 12 | +DB_USER = config("DB_USER", default="") |
| 13 | +DB_PASSWORD = config("DB_PASSWORD", default="") |
| 14 | +DB_HOST = config("DB_HOST", default="") |
| 15 | +DB_PORT = config("DB_PORT", default="") |
| 16 | +MYSQL_DRIVER = config("MYSQL_DRIVER", default="pymysql") |
| 17 | + |
| 18 | +if DB_TYPE == "sqlite": |
| 19 | + SQLALCHEMY_DATABASE_URL = f"sqlite:///./{DB_NAME}" |
| 20 | +elif DB_TYPE == "mysql": |
| 21 | + SQLALCHEMY_DATABASE_URL = f"mysql+{MYSQL_DRIVER}://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}_test" |
| 22 | +elif DB_TYPE == "postgresql": |
| 23 | + SQLALCHEMY_DATABASE_URL = ( |
| 24 | + f"postgresql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}_test" |
| 25 | + ) |
| 26 | +else: |
| 27 | + raise ValueError(f"Unsupported DB_TYPE: {DB_TYPE}") |
| 28 | + |
| 29 | +engine = create_engine( |
| 30 | + SQLALCHEMY_DATABASE_URL, |
| 31 | + connect_args={"check_same_thread": False} if DB_TYPE == "sqlite" else {}, |
| 32 | +) |
23 | 33 |
|
24 | 34 | TestingSessionLocal = sessionmaker( |
25 | 35 | autocommit=False, |
|
0 commit comments