如何使用SQLAlchemy将整个sql文件执行到数据库中?文件中可以有许多不同的sql查询,包括begin和commit / rollback。
不幸的是,我不知道一个很好的一般答案。一些dbapi(例如psycopg2)支持一次执行许多语句。如果文件不是很大,您可以将它们加载到字符串中并在连接上执行它们。对于其他人,我会尝试为该数据库使用命令行客户端,并使用子进程模块将数据传输到该数据库。
如果这些方法不可接受,那么你将不得不继续实现一个小的SQL解析器,它可以将文件拆分成单独的语句。获得100%正确非常棘手,因为您必须考虑数据库方言特定的文字转义规则,使用的字符集,影响文字解析的任何数据库配置选项(例如PostgreSQL standard_conforming_strings)。
如果你只需要获得99.9%的正确率,那么一些正则表达式的魔法应该可以让你在那里获得大部分。
我能够使用纯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文件中的所有行,忽略注释行。然后它连接形成完整语句的行并尝试执行该语句。您只需要一个文件处理程序和一个会话对象。
如果你正在使用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之后的会话?)
FYI sqlite3包含一个相关的例程:sqlite3.complete_statement(sql)如果你自己编译解析器...
您可以使用SQLalchemy和psycopg2来完成。
file = open(path)
engine = sqlalchemy.create_engine(db_url)
escaped_sql = sqlalchemy.text(file.read())
engine.execute(escaped_sql)
您可以通过此访问原始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驱动程序的文档,以查看在一次执行中是否允许多个语句,或者您是否需要使用像qysxswpoi这样的pysqlite独有的帮助程序。