Source code for sqlspec.migrations.version

"""Migration version parsing and comparison utilities.

Provides structured parsing of migration versions supporting both legacy sequential
(0001) and timestamp-based (20251011120000) formats with type-safe comparison.
"""

import re
from dataclasses import dataclass
from datetime import datetime, timezone
from enum import Enum
from typing import TYPE_CHECKING, final

if TYPE_CHECKING:
    from pathlib import Path

from sqlspec.utils.logging import get_logger

__all__ = (
    "MigrationVersion",
    "VersionType",
    "convert_to_sequential_version",
    "generate_conversion_map",
    "generate_timestamp_version",
    "get_next_sequential_number",
    "is_sequential_version",
    "is_timestamp_version",
    "parse_extension_stem",
    "parse_version",
)

logger = get_logger(__name__)

SEQUENTIAL_PATTERN = re.compile(r"^(?!\d{14}$)\d+$")
TIMESTAMP_PATTERN = re.compile(r"^(\d{14})$")
EXTENSION_PATTERN = re.compile(r"^ext_(\w+)_(.+)$")
EXTENSION_STEM_PATTERN = re.compile(r"^ext_(?P<name>\w+?)_(?P<version>\d+)(?:_.*)?$")




class VersionType(Enum):
    """Migration version format type."""

    SEQUENTIAL = "sequential"
    TIMESTAMP = "timestamp"

@final @dataclass(frozen=True) class MigrationVersion: """Parsed migration version with structured comparison support. Attributes: raw: Original version string. type: Version format type (sequential or timestamp). sequence: Numeric value for sequential versions. timestamp: Parsed datetime for timestamp versions (UTC). extension: Extension name for extension-prefixed versions. """ raw: str type: VersionType sequence: "int | None" timestamp: "datetime | None" extension: "str | None"

def __lt__(self, other: "MigrationVersion") -> bool: """Compare versions supporting mixed formats. Comparison Rules: 1. Extension migrations sort by extension name first, then version 2. Sequential < Timestamp (legacy migrations first) 3. Sequential vs Sequential: numeric comparison 4. Timestamp vs Timestamp: chronological comparison Args: other: Version to compare against. Returns: True if this version sorts before other. """ if not isinstance(other, MigrationVersion): return NotImplemented if self.extension != other.extension: if self.extension is None: return True if other.extension is None: return False return self.extension < other.extension if self.type == other.type: if self.type == VersionType.SEQUENTIAL: return (self.sequence or 0) < (other.sequence or 0) return (self.timestamp or datetime.min.replace(tzinfo=timezone.utc)) < ( other.timestamp or datetime.min.replace(tzinfo=timezone.utc) ) return self.type == VersionType.SEQUENTIAL