"""Migration context for passing runtime information to migrations."""
import asyncio
import inspect
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any
from sqlglot.dialects.dialect import Dialect
from sqlspec.protocols import HasStatementConfigProtocol
from sqlspec.utils.logging import get_logger
from sqlspec.utils.type_guards import has_statement_config_factory
if TYPE_CHECKING:
from sqlspec.driver import AsyncDriverAdapterBase, SyncDriverAdapterBase
__all__ = ("MigrationContext",)
logger = get_logger("sqlspec.migrations.context")
@dataclass(slots=True)
class MigrationContext:
"""Context object passed to migration functions.
Provides runtime information about the database environment
to migration functions, allowing them to generate dialect-specific SQL.
"""
config: "Any | None" = None
"""Database configuration object."""
dialect: "str | None" = None
"""Database dialect."""
metadata: "dict[str, Any] | None" = None
"""Additional metadata for the migration."""
extension_config: "dict[str, Any] | None" = None
"""Extension-specific configuration options."""
driver: "SyncDriverAdapterBase | AsyncDriverAdapterBase | None" = None
"""Database driver instance (available during execution)."""
_execution_metadata: "dict[str, Any]" = field(default_factory=dict)
"""Internal execution metadata for tracking async operations."""
def __post_init__(self) -> None:
"""Initialize metadata and extension config if not provided."""
if not self.metadata:
self.metadata = {}
if not self.extension_config:
self.extension_config = {}