从 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模型中的进程字段(使用flask_sqlalchemy)