Migration Strategy

Version Management System

import re
from typing import List, Optional, Tuple
from dataclasses import dataclass
from enum import Enum

class VersionType(Enum):
    MAJOR = "major"
    MINOR = "minor"
    PATCH = "patch"
    PRERELEASE = "prerelease"

@dataclass
class Version:
    major: int
    minor: int
    patch: int
    prerelease: Optional[str] = None
    build: Optional[str] = None

    @classmethod
    def parse(cls, version_string: str) -> 'Version':
        """Parse version string (e.g., '1.2.3-beta.1+build.123')"""

        # Remove 'v' prefix if present
        version_string = version_string.lstrip('v')

        # Regex for semantic version
        pattern = r'^(\d+)\.(\d+)\.(\d+)(?:-([0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*))?(?:\+([0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*))?$'
        match = re.match(pattern, version_string)

        if not match:
            raise ValueError(f"Invalid version string: {version_string}")

        return cls(
            major=int(match.group(1)),
            minor=int(match.group(2)),
            patch=int(match.group(3)),
            prerelease=match.group(4),
            build=match.group(5)
        )

    def __str__(self) -> str:
        version = f"{self.major}.{self.minor}.{self.patch}"
        if self.prerelease:
            version += f"-{self.prerelease}"
        if self.build:
            version += f"+{self.build}"
        return version

    def __lt__(self, other: 'Version') -> bool:
        """Compare versions according to semver rules"""

        # Compare major.minor.patch
        self_tuple = (self.major, self.minor, self.patch)
        other_tuple = (other.major, other.minor, other.patch)

        if self_tuple != other_tuple:
            return self_tuple < other_tuple

        # Handle prerelease versions
        if self.prerelease is None and other.prerelease is None:
            return False
        elif self.prerelease is None:
            return False  # Release > prerelease
        elif other.prerelease is None:
            return True   # Prerelease < release
        else:
            return self._compare_prerelease(self.prerelease, other.prerelease)

    def _compare_prerelease(self, pre1: str, pre2: str) -> bool:
        """Compare prerelease versions"""
        parts1 = pre1.split('.')
        parts2 = pre2.split('.')

        for i in range(max(len(parts1), len(parts2))):
            part1 = parts1[i] if i < len(parts1) else ''
            part2 = parts2[i] if i < len(parts2) else ''

            # Try to compare as integers
            try:
                int1, int2 = int(part1), int(part2)
                if int1 != int2:
                    return int1 < int2
            except ValueError:
                # String comparison
                if part1 != part2:
                    return part1 < part2

        return len(parts1) < len(parts2)

    def bump(self, version_type: VersionType) -> 'Version':
        """Bump version according to type"""

        if version_type == VersionType.MAJOR:
            return Version(self.major + 1, 0, 0)
        elif version_type == VersionType.MINOR:
            return Version(self.major, self.minor + 1, 0)
        elif version_type == VersionType.PATCH:
            return Version(self.major, self.minor, self.patch + 1)
        else:
            raise ValueError(f"Unsupported version type: {version_type}")

    def is_compatible(self, other: 'Version') -> bool:
        """Check if versions are backward compatible"""

        # Same major version is generally compatible
        if self.major == other.major:
            return True

        # Major version 0 is special - minor versions are not compatible
        if self.major == 0 or other.major == 0:
            return self.major == other.major and self.minor == other.minor

        return False


class VersionManager:
    def __init__(self, db: Database):
        self.db = db

    async def get_current_version(self) -> Version:
        """Get current system version"""
        result = await self.db.fetch_one("""
            SELECT version FROM system_version
            ORDER BY created_at DESC LIMIT 1
        """)

        if not result:
            raise Exception("No version information found")

        return Version.parse(result["version"])

    async def get_version_history(self, limit: int = 50) -> List[Tuple[Version, dict]]:
        """Get version deployment history"""
        results = await self.db.fetch_all("""
            SELECT version, deployed_at, deployed_by, release_notes
            FROM deployment_history
            ORDER BY deployed_at DESC
            LIMIT $1
        """, limit)

        return [(Version.parse(row["version"]), {
            "deployed_at": row["deployed_at"],
            "deployed_by": row["deployed_by"],
            "release_notes": row["release_notes"]
        }) for row in results]

    async def validate_version_compatibility(self,
                                           target_version: Version) -> List[str]:
        """Validate if target version is compatible for deployment"""

        current_version = await self.get_current_version()
        issues = []

        # Check backward compatibility
        if not current_version.is_compatible(target_version):
            if target_version < current_version:
                # Downgrade scenario
                if target_version.major < current_version.major:
                    issues.append(
                        f"Major version downgrade from {current_version.major} "
                        f"to {target_version.major} may cause data compatibility issues"
                    )
            else:
                # Upgrade scenario
                if target_version.major > current_version.major:
                    issues.append(
                        f"Major version upgrade from {current_version.major} "
                        f"to {target_version.major} requires migration planning"
                    )

        # Check database schema compatibility
        schema_issues = await self._check_schema_compatibility(
            current_version, target_version
        )
        issues.extend(schema_issues)

        # Check API compatibility
        api_issues = await self._check_api_compatibility(
            current_version, target_version
        )
        issues.extend(api_issues)

        return issues

    async def _check_schema_compatibility(self,
                                        current: Version,
                                        target: Version) -> List[str]:
        """Check database schema compatibility"""

        # Get schema versions for both versions
        current_schema = await self._get_schema_version(current)
        target_schema = await self._get_schema_version(target)

        issues = []

        if target_schema < current_schema:
            # Check if rollback is safe
            migrations_to_rollback = await self._get_migrations_between_versions(
                current_schema, target_schema
            )

            irreversible_migrations = [
                m for m in migrations_to_rollback if not m.reversible
            ]

            if irreversible_migrations:
                issues.append(
                    f"Cannot rollback due to irreversible migrations: "
                    f"{[m.version for m in irreversible_migrations]}"
                )

        return issues