从 SQLAlchemy 中的文件执行 SQL

Question

如何使用 SQLAlchemy 将整个 sql 文件执行到数据库中？文件中可以有许多不同的 sql 查询，包括 begin 和 commit/rollback。

Answer 1

我能够使用纯 SQLAlchemy 和一些字符串操作来运行 .sql 模式文件。这肯定不是一种优雅的方法，但它确实有效。

# Open the .sql file
sql_file = open('file.sql','r')

# Create an empty command string
sql_command = ''

# Iterate over all lines in the sql file
for line in sql_file:
    # Ignore commented lines
    if not line.startswith('--') and line.strip('\n'):
        # Append line to the command string
        sql_command += line.strip('\n')

        # If the command string ends with ';', it is a full statement
        if sql_command.endswith(';'):
            # Try to execute statement and commit it
            try:
                session.execute(text(sql_command))
                session.commit()

            # Assert in case of error
            except:
                print('Ops')

            # Finally, clear command string
            finally:
                sql_command = ''

它遍历 .sql 文件中的所有行，忽略注释行。 然后它连接形成完整语句的行并尝试执行该语句。您只需要一个文件处理程序和一个会话对象。

Answer 2

使用sqlalchemy.sql.text

通过使用内置的text 构造，有一种更直接的方法来执行.sql 文件。

from sqlalchemy import create_engine
from sqlalchemy.sql import text

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

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

    con.execute(query)

Answer 3

您可以使用 SQLalchemy 和 psycopg2 来做到这一点。

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

Answer 4

不幸的是，我不知道对此有一个好的一般答案。一些 dbapi（例如 psycopg2）支持一次执行许多语句。如果文件不是很大，您可以将它们加载到字符串中并在连接上执行它们。对于其他人，我会尝试为该数据库使用命令行客户端，并使用 subprocess 模块将数据传输到该数据库中。

如果这些方法不可接受，那么您必须继续实施一个小型 SQL 解析器，该解析器可以将文件拆分为单独的语句。要获得 100% 正确，这真的很棘手，因为您必须考虑数据库方言特定的文字转义规则、使用的字符集、影响文字解析的任何数据库配置选项（例如 PostgreSQL standard_conforming_strings）。

如果您只需要得到这 99.9% 的正确率，那么一些正则表达式魔法应该可以帮助您完成大部分工作。

Answer 5

如果您使用的是 sqlite3，它对 dbapi 有一个有用的扩展名为 conn.executescript(str)，我已经通过类似这样的方式将其连接起来，它似乎可以工作：（未显示所有上下文，但应该足够了得到漂移）

def init_from_script(script):
    Base.metadata.drop_all(db_engine)
    Base.metadata.create_all(db_engine)     

    # HACK ALERT: we can do this using sqlite3 low level api, then reopen session.
    f = open(script)
    script_str = f.read().strip()
    global db_session
    db_session.close()
    import sqlite3
    conn = sqlite3.connect(db_file_name)
    conn.executescript(script_str)
    conn.commit()

    db_session = Session()

我想知道这是纯粹的邪恶吗？我徒劳地寻找“纯” sqlalchemy 等价物，也许可以将其添加到库中，例如 db_session.execute_script(file_name) ？我希望 db_session 在这一切之后都能正常工作（即不需要重新启动引擎）但还不确定......需要进一步研究（即我们是否需要在 sqlalchemy 背后获得一个新引擎或只是一个会话?)

仅供参考 sqlite3 包含一个相关例程：如果您使用自己的解析器，则 sqlite3.complete_statement(sql)...

Answer 6

您可以通过此访问原始 DBAPI 连接

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

但这取决于您使用的方言/驱动程序，可以通过此list 引用。

对于 pyscog2，你可以这样做

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

但是你需要做 pysqlite

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

根据这一点，您需要检查您正在使用的任何 DBAPI 驱动程序的文档，以查看是否允许在一次执行中执行多个语句，或者您是否需要使用像 executescript 这样的助手，这是 pysqlite 独有的.

Answer 7

以下是如何运行脚本拆分语句，并使用 SQLAlchemy 引擎直接使用 "connectionless" execution 运行每个语句。这假定每个语句都以 ; 结尾，并且每行不超过一个语句。

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))

Answer 8

在当前的答案中，当 .SQL 文件中存在这些功能的组合时，我没有找到可行的解决方案：

• 带有“--”的评论
• 多行语句，在“--”之后带有额外的 cmets
• 具有多个以“;”结尾的 SQL 查询的函数定义但必须作为一个完整的语句执行

A 找到了一个相当简单的解决方案：

# 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)

Answer 9

David 的回答here 取得了成功，但做了两处细微的修改：

1. 使用get_bind()，因为我使用的是会话而不是引擎
2. 在原始连接上调用 cursor()

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