从 SQLAlchemy 中的文件执行 SQL

Posted

技术标签:

【中文标题】从 SQLAlchemy 中的文件执行 SQL【英文标题】:Execute SQL from file in SQLAlchemy 【发布时间】:2011-01-17 02:31:03 【问题描述】:

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

【问题讨论】:

【参考方案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 文件中的所有行,忽略注释行。 然后它连接形成完整语句的行并尝试执行该语句。您只需要一个文件处理程序和一个会话对象。

【讨论】:

很好,使用with open('','') 并在session 区域中使用SQLAlchemy 类会很好。这就是它的工作原理,前者有助于在文件读取访问后进行清理......【参考方案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)

【讨论】:

绝对一流的信息在这里。只需添加一个:在“sqlalchemy.sql”上找不到“文本”相反,我以这种方式直接从 sqlalchemy 包中找到(并成功使用它):from sqlalchemy import text 这是迄今为止最好的方法。干杯,伙计【参考方案3】:

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

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

【讨论】:

这不适用于所有方言。 SQLite 例如明确禁止在一个“执行”命令中执行多个查询。 @ThePadawan pysqlite 确实提供了执行脚本,因此简单的条件检查将允许它通过 raw_connection 我相信这会对DDL产生问题。 Alchemy 检测 DDL 并创建一个保持打开状态的隐式事务。这是我遇到的问题,因为如果存在 DDL,我的所有脚本都会通过炼金术破解。【参考方案4】:

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

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

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

【讨论】:

【参考方案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)...

【讨论】:

【参考方案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 独有的.

【讨论】:

【参考方案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))

【讨论】:

【参考方案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)

【讨论】:

【参考方案9】:

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

    使用get_bind(),因为我使用的是会话而不是引擎 在原始连接上调用 cursor()
raw_connection = myDbSession.get_bind().raw_connection()
raw_cursor = raw_connection.cursor()
raw_cursor.execute(open("my_script.sql").read())

【讨论】:

以上是关于从 SQLAlchemy 中的文件执行 SQL的主要内容,如果未能解决你的问题,请参考以下文章

如何从sqlalchemy中的查询中获取列表[重复]

使用 SQLAlchemy 转义文件路径中的特殊字符

SQLAlchemy模型中的进程字段(使用flask_sqlalchemy)

石墨烯突变未映射 SQLAlchemy 中的模型

如何对 SQLAlchemy 中的整个列强制执行唯一性约束?

如何将一个类映射到sqlalchemy orm中的多个数据库