Andrey
be8af704ba
- Реализована возможность получения тела последнего объединенного PR по коммиту в GitHub Actions. - Добавлен шаг для отправки описания PR в важные логи через Telegram. - Обновлены тесты для проверки нового функционала и улучшения логики обработки сообщений.
79 lines
3.4 KiB
Python
79 lines
3.4 KiB
Python
"""Репозиторий для работы с миграциями базы данных."""
|
||
import aiosqlite
|
||
from database.base import DatabaseConnection
|
||
|
||
|
||
class MigrationRepository(DatabaseConnection):
|
||
"""Репозиторий для управления миграциями базы данных."""
|
||
|
||
async def create_table(self):
|
||
"""Создает таблицу migrations, если она не существует."""
|
||
query = """
|
||
CREATE TABLE IF NOT EXISTS migrations (
|
||
id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
|
||
script_name TEXT NOT NULL UNIQUE,
|
||
applied_at INTEGER NOT NULL DEFAULT (strftime('%s', 'now'))
|
||
)
|
||
"""
|
||
await self._execute_query(query)
|
||
self.logger.info("Таблица migrations создана или уже существует")
|
||
|
||
async def get_applied_migrations(self) -> list[str]:
|
||
"""Возвращает список имен примененных скриптов миграций."""
|
||
conn = None
|
||
try:
|
||
conn = await self._get_connection()
|
||
cursor = await conn.execute("SELECT script_name FROM migrations ORDER BY applied_at")
|
||
rows = await cursor.fetchall()
|
||
await cursor.close()
|
||
return [row[0] for row in rows]
|
||
except Exception as e:
|
||
self.logger.error(f"Ошибка при получении списка миграций: {e}")
|
||
raise
|
||
finally:
|
||
if conn:
|
||
await conn.close()
|
||
|
||
async def is_migration_applied(self, script_name: str) -> bool:
|
||
"""Проверяет, применена ли миграция."""
|
||
conn = None
|
||
try:
|
||
conn = await self._get_connection()
|
||
cursor = await conn.execute(
|
||
"SELECT COUNT(*) FROM migrations WHERE script_name = ?",
|
||
(script_name,)
|
||
)
|
||
row = await cursor.fetchone()
|
||
await cursor.close()
|
||
return row[0] > 0 if row else False
|
||
except Exception as e:
|
||
self.logger.error(f"Ошибка при проверке миграции {script_name}: {e}")
|
||
raise
|
||
finally:
|
||
if conn:
|
||
await conn.close()
|
||
|
||
async def mark_migration_applied(self, script_name: str) -> None:
|
||
"""Отмечает миграцию как примененную."""
|
||
conn = None
|
||
try:
|
||
conn = await self._get_connection()
|
||
await conn.execute(
|
||
"INSERT INTO migrations (script_name) VALUES (?)",
|
||
(script_name,)
|
||
)
|
||
await conn.commit()
|
||
self.logger.info(f"Миграция {script_name} отмечена как примененная")
|
||
except aiosqlite.IntegrityError:
|
||
self.logger.warning(f"Миграция {script_name} уже была применена ранее")
|
||
except Exception as e:
|
||
self.logger.error(f"Ошибка при отметке миграции {script_name}: {e}")
|
||
raise
|
||
finally:
|
||
if conn:
|
||
await conn.close()
|
||
|
||
async def create_table_from_sql(self, sql_script: str) -> None:
|
||
"""Создает таблицу из SQL скрипта. Используется в миграциях."""
|
||
await self._execute_query(sql_script)
|