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 Graph-sitter to automate this migration, handling query syntax, session usage, and ORM patterns.
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: