(
conn,
db_name: str,
latest_version: int,
migration_script_dir: str,
migration_script_filename_format: str,
)
| 83 | |
| 84 | |
| 85 | async def migrate_if_needed( |
| 86 | conn, |
| 87 | db_name: str, |
| 88 | latest_version: int, |
| 89 | migration_script_dir: str, |
| 90 | migration_script_filename_format: str, |
| 91 | ): |
| 92 | current_version = await get_current_version(conn) |
| 93 | if current_version >= latest_version: |
| 94 | logger.info(f"Database {db_name} is already up to date with version {current_version}") |
| 95 | return |
| 96 | |
| 97 | logger.debug(f"Init database {db_name} with version {current_version}") |
| 98 | migration_files = sorted(os.listdir(migration_script_dir)) |
| 99 | |
| 100 | version = current_version |
| 101 | for filename in migration_files: |
| 102 | version = extract_version_from_filename(migration_script_filename_format, filename) |
| 103 | if version and version > current_version: |
| 104 | filepath = os.path.join(migration_script_dir, filename) |
| 105 | with open(filepath, "r") as f: |
| 106 | sql_content = f.read() |
| 107 | await apply_migration(conn, version, sql_content) |
| 108 | |
| 109 | if version > current_version: |
| 110 | # todo: also clean redis data |
| 111 | pass |
| 112 | # logger.info(f"Database {db_name} upgraded to version {version}, target version is {latest_version}") |
| 113 | # await redis_conn.clean_data() |
no test coverage detected