Coverage for src / lexigram / contracts / data / sql / migrations.py: 0%

1"""Migration runner protocol. 

2 

3Defines the contract for database migration execution so that ``lexigram-cli`` 

4and other consumers can resolve a runner from the container without importing 

5``lexigram-sql`` directly. 

6 

7Concrete implementations (Alembic adapter, ``SimpleMigrationManager``, etc.) 

8live in ``lexigram-sql`` and are registered by ``DatabaseProvider`` at boot time. 

9 

10Example:: 

11 

12 from lexigram.contracts.data.sql.migrations import MigrationRunnerProtocol 

13 

14 runner = await container.resolve(MigrationRunnerProtocol) 

15 await runner.run_migrations() 

16""" 

17 

18from __future__ import annotations 

19 

20from typing import Protocol, runtime_checkable 

21 

22 

23@runtime_checkable 

24class MigrationRunnerProtocol(Protocol): 

25 """Executes database migrations in the correct order. 

26 

27 Implementations must be idempotent: running ``run_migrations()`` on an 

28 already up-to-date schema must be a no-op. Each migration is identified 

29 by a unique string version identifier (e.g. ``"20240101_001"`` or the 

30 Alembic revision hash). 

31 

32 Typical usage:: 

33 

34 runner = await container.resolve(MigrationRunnerProtocol) 

35 result = await runner.run_migrations() 

36 if result: 

37 logger.info("migrations_applied", count=len(result)) 

38 else: 

39 logger.info("schema_up_to_date") 

40 """ 

41 

42 async def run_migrations(self) -> list[str]: 

43 """Apply all pending migrations. 

44 

45 Returns: 

46 List of migration version identifiers that were applied. 

47 Empty list means the schema was already up-to-date. 

48 """ 

49 ... 

50 

51 async def rollback(self, target: str | None = None) -> list[str]: 

52 """Roll back migrations to *target*. 

53 

54 Args: 

55 target: Version identifier to roll back to. ``None`` rolls back 

56 the most recent migration only. 

57 

58 Returns: 

59 List of migration version identifiers that were rolled back. 

60 """ 

61 ... 

62 

63 async def get_current_version(self) -> str | None: 

64 """Return the version identifier of the most recently applied migration. 

65 

66 Returns: 

67 Version string, or ``None`` if no migrations have been applied. 

68 """ 

69 ... 

70 

71 async def get_pending_migrations(self) -> list[str]: 

72 """Return version identifiers of migrations not yet applied. 

73 

74 Returns: 

75 Ordered list of pending migration version identifiers. 

76 """ 

77 ... 

78 

79 

80__all__ = ["MigrationRunnerProtocol"]