How to execute raw SQL in SQLAlchemy

Overview

SQLAlchemy is a powerful ORM that provides a high-level API for database interactions in Python. However, sometimes you may need to execute raw SQL for efficiency or to leverage database-specific features. This guide covers executing raw SQL within the SQLAlchemy framework.

Introduction to Raw SQL in SQLAlchemy

SQLAlchemy, while known for its object-relational mapping (ORM) capabilities, also allows direct execution of raw SQL statements. This can be beneficial when you have complex queries, need to optimize performance, or utilize features unique to your database engine. Executing raw SQL gives you the power and flexibility to do just that.

To execute raw SQL you’ll have to use SQLAlchemy’s Connection object, which can be obtained either from a Engine or a Session context. Let’s explore some common patterns for executing raw SQL within SQLAlchemy through progressive examples.

Executing Simple SQL Queries

To execute raw SQL, obtain a connection from your engine:

from sqlalchemy import create_engine
# Replace 'dialect+driver://username:password@host/dbname' with your actual database URI
db_engine = create_engine('dialect+driver://username:password@host/dbname')

with db_engine.connect() as connection:
    result = connection.execute("SELECT * FROM my_table")
    for row in result:
        print(row)

This will print out each row in the result set from the ‘my_table’ table.

Parameterized Queries

For security reasons and to prevent SQL injection attacks, never simply interpolate variables directly into your SQL strings. Instead, use named parameters or positional placeholders:

with db_engine.connect() as connection:
    result = connection.execute(
        "SELECT * FROM users WHERE username = :username", {'username': 'example_user'}
    )
    user = result.fetchone()
    print(user)

In the above example, “:username” is a placeholder that gets safely replaced by “example_user”.

Using Textual SQL

SQLAlchemy’s text function can be used to create SQL expressions that carry placeholders:

from sqlalchemy.sql import text

sql = text("SELECT * FROM users WHERE username = :username")

with db_engine.connect() as connection:
    result = connection.execute(sql, username='example_user')
    user = result.fetchone()
    print(user)

Here, the text function wraps the SQL with named parameters offering flexibility and injection prevention.

Executing Inserts, Updates, Deletes

Modification operations such as INSERT, UPDATE, and DELETE can also be performed in a similar manner:

# Inserting a new user
insert_sql = text("INSERT INTO users (username, email) VALUES (:username, :email)")

with db_engine.connect() as connection:
    connection.execute(insert_sql, username='new_user', email='[email protected]')

# Updating a user's email
update_sql = text("UPDATE users SET email = :email WHERE username = :username")

with db_engine.connect() as connection:
    connection.execute(update_sql, email='[email protected]', username='existing_user')

# Deleting a user
delete_sql = text("DELETE FROM users WHERE username = :username")

with db_engine.connect() as connection:
    connection.execute(delete_sql, username='obsolete_user')

Working with Transactions

For running transactions, you’ll use the connection object to first begin a transaction before executing your SQL statements. This ensures atomicity:

with db_engine.connect() as connection:
    transaction = connection.begin()
    try:
        connection.execute(insert_sql, {...})
        connection.execute(update_sql, {...})
        transaction.commit()
    except:
        transaction.rollback()
        raise

This wraps the insert and update operations in a transaction that can be rolled back on failure.

Executing Stored Procedures

Stored procedures can also be called via raw SQL:

call_procedure_sql = text("CALL my_stored_procedure(:param)")

with db_engine.connect() as connection:
    result = connection.execute(call_procedure_sql, param='value')
    for row in result:
        print(row)

Using SQLAlchemy Core for Complex Queries

Beyond simple text statements, SQLAlchemy’s Core language enables the combination of textual SQL with Python logic:

from sqlalchemy.sql import select, table, column

t_user = table('users', column('username'), column('email'))
stmt = select([t_user]).where(t_user.c.username == 'example_user')

with db_engine.connect() as connection:
    for row in connection.execute(stmt):
        print(row)

This example demonstrates how to select from a ‘users’ table where the username matches ‘example_user’, using SQLAlchemy Core constructs.

Accessing Native Database Functionality

Finally, with SQLAlchemy, you pass your raw SQL directly to the underlying DBAPI connection for cases where database function-specific features are needed:

with db_engine.raw_connection() as raw_conn:
    cursor = raw_conn.cursor()
    cursor.execute("YOUR_VENDOR_SPECIFIC_SQL_HERE")
    results = cursor.fetchall()
    for result in results:
        print(result)
    cursor.close()

Conclusion

This guide has highlighted various methods for executing raw SQL with SQLAlchemy, ranging from simple queries to complex transactions and even direct access to DBAPI functionalities. Use these methods responsibly, always parameterize queries to protect against SQL injection, and remember to take advantage of SQLAlchemy’s robust ORM features where possible.