Implementation:Microsoft Autogen SchemaManager

Knowledge Sources	Microsoft_Autogen
Domains	Database, Schema Management, Migrations, Alembic
Last Updated	2026-02-11 17:00 GMT

Overview

SchemaManager manages database schema validation and migrations using Alembic for AutoGen Studio's SQLModel-based database operations.

Description

The SchemaManager class provides a comprehensive interface for managing database schema migrations using Alembic. It handles the initialization of Alembic configuration, generates migration revisions, detects schema differences between models and the database, and applies migrations to keep the database schema synchronized with SQLModel definitions. The class operates in conjunction with DatabaseManager and supports operations like force reinitialization, automatic schema upgrades, and migration status tracking. It generates and maintains Alembic configuration files (alembic.ini, env.py, script templates) dynamically and customizes them to work seamlessly with SQLModel metadata.

Usage

Use SchemaManager when you need to manage database schema evolution in AutoGen Studio, detect unmigrated changes, generate new migration revisions, upgrade database schemas, or ensure database consistency across application versions.

Code Reference

Source Location

• Repository: Microsoft_Autogen
• File: python/packages/autogen-studio/autogenstudio/database/schema_manager.py
• Lines: 1-557

Signature

class SchemaManager:
    def __init__(
        self,
        engine: Engine,
        base_dir: Optional[Path] = None,
    )

    def initialize_migrations(self, force: bool = False) -> bool
    def get_alembic_config(self) -> Config
    def get_current_revision(self) -> Optional[str]
    def get_head_revision(self) -> Optional[str]
    def get_schema_differences(self) -> List[tuple]
    def check_schema_status(self) -> Tuple[bool, str]
    def upgrade_schema(self, revision: str = "head") -> bool
    def check_and_upgrade(self) -> Tuple[bool, str]
    def generate_revision(self, message: str = "auto") -> Optional[str]
    def get_pending_migrations(self) -> List[str]
    def print_status(self) -> None
    def ensure_schema_up_to_date(self) -> bool

Import

from autogenstudio.database.schema_manager import SchemaManager
from sqlalchemy import create_engine

engine = create_engine("sqlite:///./database.db")
schema_manager = SchemaManager(engine=engine)

I/O Contract

Inputs

Name	Type	Required	Description
engine	Engine	Yes	SQLAlchemy engine instance for database connection
base_dir	Optional[Path]	No	Base directory for Alembic files; defaults to current working directory if not provided
force	bool	No	Force reinitialization of migrations (used in initialize_migrations)
revision	str	No	Target revision for schema upgrade (default: "head")
message	str	No	Message for new migration revision (default: "auto")

Outputs

Name	Type	Description
initialize_migrations	bool	Returns True if initialization was successful, False otherwise
get_current_revision	Optional[str]	Current database revision string or None if no revision exists
get_head_revision	Optional[str]	Latest available revision string or None if no revisions exist
get_schema_differences	List[tuple]	List of differences between database and models
check_schema_status	Tuple[bool, str]	Tuple of (needs_upgrade, status_message)
upgrade_schema	bool	Returns True if upgrade was successful, False otherwise
check_and_upgrade	Tuple[bool, str]	Tuple of (action_taken, status_message)
generate_revision	Optional[str]	Revision ID if successful, None otherwise
get_pending_migrations	List[str]	List of pending migration revision IDs

Usage Examples

Basic Schema Management

from sqlalchemy import create_engine
from autogenstudio.database.schema_manager import SchemaManager
from pathlib import Path

# Initialize schema manager
engine = create_engine("sqlite:///./autogen_studio.db")
schema_manager = SchemaManager(
    engine=engine,
    base_dir=Path("./database")
)

# Initialize migrations
success = schema_manager.initialize_migrations(force=False)
print(f"Migrations initialized: {success}")

# Check schema status
needs_upgrade, status = schema_manager.check_schema_status()
print(f"Needs upgrade: {needs_upgrade}")
print(f"Status: {status}")

# Upgrade if needed
if needs_upgrade:
    success = schema_manager.upgrade_schema()
    print(f"Schema upgraded: {success}")

Checking for Schema Differences

# Get current and head revisions
current = schema_manager.get_current_revision()
head = schema_manager.get_head_revision()

print(f"Current revision: {current}")
print(f"Head revision: {head}")

# Get pending migrations
pending = schema_manager.get_pending_migrations()
print(f"Pending migrations: {len(pending)}")

# Get schema differences
differences = schema_manager.get_schema_differences()
if differences:
    print("Schema differences detected:")
    for diff in differences:
        print(f"  - {diff}")

Generating New Migration

# Generate a new migration revision
revision_id = schema_manager.generate_revision(
    message="Add new user profile fields"
)

if revision_id:
    print(f"Generated revision: {revision_id}")
    # Apply the migration
    schema_manager.upgrade_schema()
else:
    print("Failed to generate revision")

Auto-upgrade with Error Handling

# Check and automatically upgrade if needed
action_taken, message = schema_manager.check_and_upgrade()

if action_taken:
    print(f"Schema was upgraded: {message}")
else:
    print(f"No upgrade needed: {message}")

# Print detailed status
schema_manager.print_status()

Related Pages

• DatabaseManager - Uses SchemaManager for database initialization
• Web Dependencies - Initializes database with schema management
• SQLModel Models - Defines the schema that SchemaManager manages