Execute SQL from file in SQLAlchemy

sqlalchemy.text or sqlalchemy.sql.text

• The text construct provides a straightforward method to directly execute .sql files.

from sqlalchemy import create_engine
from sqlalchemy import text
# or from sqlalchemy.sql import text

engine = create_engine('mysql://{USR}:{PWD}@localhost:3306/db', echo=True)

with engine.connect() as con:
    with open("src/models/query.sql") as file:
        query = text(file.read())
        con.execute(query)

• SQLAlchemy: Using Textual SQL
  • text()

You can do it with SQLalchemy and psycopg2.

file = open(path)
engine = sqlalchemy.create_engine(db_url)
escaped_sql = sqlalchemy.text(file.read())
engine.execute(escaped_sql)

Unfortunately I'm not aware of a good general answer for this. Some dbapi's (psycopg2 for instance) support executing many statements at a time. If the files aren't huge you can just load them into a string and execute them on a connection. For others, I would try to use a command-line client for that db and pipe the data into that using the subprocess module.

If those approaches aren't acceptable, then you'll have to go ahead and implement a small SQL parser that can split the file apart into separate statements. This is really tricky to get 100% correct, as you'll have to factor in database dialect specific literal escaping rules, the charset used, any database configuration options that affect literal parsing (e.g. PostgreSQL standard_conforming_strings).

If you only need to get this 99.9% correct, then some regexp magic should get you most of the way there.

You can access the raw DBAPI connection through this

    raw_connection = mySqlAlchemyEngine.raw_connection()
    raw_cursor = raw_connection() #get a hold of the proxied DBAPI connection instance

but then it will depend on which dialect/driver you are using which can be referred to through this list.

For pyscog2, you can just do

    raw_cursor.execute(open("my_script.sql").read())

but pysqlite you would need to do

    raw_cursor.executescript(open("my_script").read())

and in line with that you would need to check the documentation of whichever DBAPI driver you are using to see if multiple statements are allowed in one execute or if you would need to use a helper like executescript which is unique to pysqlite.

Here's how to run the script splitting the statements, and running each statement directly with a "connectionless" execution with the SQLAlchemy Engine. This assumes that each statement ends with a ; and that there's no more than one statement per line.

engine = create_engine(url)
with open('script.sql') as file:
    statements = re.split(r';\s*$', file.read(), flags=re.MULTILINE)
    for statement in statements:
        if statement:
            engine.execute(text(statement))

In the current answers, I did not found a solution which works when a combination of these features in the .SQL file is present:

• Comments with "--"
• Multi-line statements with additional comments after "--"
• Function definitions which have multiple SQL-queries ending with ";" butmust be executed as a whole statement

A found a rather simple solution:

# check for /* */
with open(file, 'r') as f:
    assert '/*' not in f.read(), 'comments with /* */ not supported in SQL file python interface'

# we check out the SQL file line-by-line into a list of strings (without \n, ...)
with open(file, 'r') as f:
    queries = [line.strip() for line in f.readlines()]

# from each line, remove all text which is behind a '--'
def cut_comment(query: str) -> str:
    idx = query.find('--')
    if idx >= 0:
        query = query[:idx]
    return query

# join all in a single line code with blank spaces
queries = [cut_comment(q) for q in queries]
sql_command = ' '.join(queries)

# execute in connection (e.g. sqlalchemy)
conn.execute(sql_command)

Code bellow works for me in alembic migrations

from alembic import op
import sqlalchemy as sa
from ekrec.common import get_project_root


def upgrade():
    path = f'{get_project_root()}/migrations/versions/fdb8492f75b2_.sql'
    op.execute(open(path).read())

I had success with David's answer here, with two slight modifications:

1. Use get_bind() as I was working with a Session rather than an Engine
2. Call cursor() on the raw connection

raw_connection = myDbSession.get_bind().raw_connection()
raw_cursor = raw_connection.cursor()
raw_cursor.execute(open("my_script.sql").read())