Codegen - Manipulate Code at Scale

Migrating from SQLAlchemy 1.4 to 2.0 involves several API changes to support the new 2.0-style query interface. This guide will walk you through using Codegen to automate this migration, handling query syntax, session usage, and ORM patterns.

You can find the complete example code in our examples repository.

Overview

The migration process involves three main steps:

Converting legacy Query objects to select() statements
Updating session execution patterns
Modernizing ORM relationship declarations

Let’s walk through each step using Codegen.

Step 1: Convert Query to Select

First, we need to convert legacy Query-style operations to the new select() syntax:

def convert_query_to_select(file):
    """Convert Query-style operations to select() statements"""
    for call in file.function_calls:
        if call.name == "query":
            # Convert query(Model) to select(Model)
            call.set_name("select")

            # Update method chains
            if call.parent and call.parent.is_method_chain:
                chain = call.parent
                if "filter" in chain.source:
                    # Convert .filter() to .where()
                    chain.source = chain.source.replace(".filter(", ".where(")
                if "filter_by" in chain.source:
                    # Convert .filter_by(name='x') to .where(Model.name == 'x')
                    model = call.args[0].value
                    conditions = chain.source.split("filter_by(")[1].split(")")[0]
                    new_conditions = []
                    for cond in conditions.split(","):
                        if "=" in cond:
                            key, value = cond.split("=")
                            new_conditions.append(f"{model}.{key.strip()} == {value.strip()}")
                    chain.edit(f".where({' & '.join(new_conditions)})")

This transforms code from:

# Legacy Query style
session.query(User).filter_by(name='john').filter(User.age >= 18).all()

to:

# New select() style
session.execute(
    select(User).where(User.name == 'john').where(User.age >= 18)
).scalars().all()

SQLAlchemy 2.0 standardizes on select() statements for all queries, providing better type checking and a more consistent API.

Step 2: Update Session Execution

Next, we update how queries are executed with the Session:

def update_session_execution(file):
    """Update session execution patterns for 2.0 style"""
    for call in file.function_calls:
        if call.name == "query":
            # Find the full query chain
            chain = call
            while chain.parent and chain.parent.is_method_chain:
                chain = chain.parent

            # Wrap in session.execute() if needed
            if not chain.parent or "execute" not in chain.parent.source:
                chain.edit(f"execute(select{chain.source[5:]})")

            # Add .scalars() for single-entity queries
            if len(call.args) == 1:
                chain.edit(f"{chain.source}.scalars()")

This converts patterns like:

# Old style
users = session.query(User).all()
first_user = session.query(User).first()

to:

# New style
users = session.execute(select(User)).scalars().all()
first_user = session.execute(select(User)).scalars().first()

The new execution pattern is more explicit about what’s being returned, making it easier to understand and maintain type safety.

Step 3: Update ORM Relationships

Finally, we update relationship declarations to use the new style:

Introduction

Tutorials

Building with Codegen

Migrating from SQLAlchemy 1.4 to 2.0

Overview

Step 1: Convert Query to Select

Step 2: Update Session Execution

Step 3: Update ORM Relationships

​Overview

​Step 1: Convert Query to Select

​Step 2: Update Session Execution

​Step 3: Update ORM Relationships

Overview

Step 1: Convert Query to Select

Step 2: Update Session Execution

Step 3: Update ORM Relationships