Apply migration script to database :param conn: asyncpg database connection :param version: the target version of migration :param sql: sql script content
(
conn,
version: int,
sql: str,
)
| 45 | |
| 46 | |
| 47 | async def apply_migration( |
| 48 | conn, |
| 49 | version: int, |
| 50 | sql: str, |
| 51 | ): |
| 52 | """ |
| 53 | Apply migration script to database |
| 54 | :param conn: asyncpg database connection |
| 55 | :param version: the target version of migration |
| 56 | :param sql: sql script content |
| 57 | """ |
| 58 | |
| 59 | try: |
| 60 | async with conn.transaction(): |
| 61 | if sql.strip() == "": |
| 62 | logger.warning(f"Migration version {version} script is empty") |
| 63 | else: |
| 64 | await conn.execute(sql) |
| 65 | await conn.execute(f"INSERT INTO {MIGRATION_TABLE} (version) VALUES ($1)", version) |
| 66 | logger.info(f"Migration version {version} applied successfully") |
| 67 | except Exception as e: |
| 68 | logger.error(f"Error applying migration version {version}: {e}") |
| 69 | |
| 70 | |
| 71 | def extract_version_from_filename( |
no test coverage detected