编译 SQL of `create_all`

Question

我想加快我的集成测试速度并执行相当于 create_all()

的原始 SQL 代码

我的想法是运行 create_all（为了得到它 SQL 等效）在测试会话开始时只使用一次，并在测试之间使用 SQL 代码迁移表。

你知道怎么做吗？

提前致谢！

Answer 1

您可以通过使用 sqlalachemy after_cursor_execute 挂钩来挂钩您的代码来完成任务。

https://docs.sqlalchemy.org/en/13/core/events.html#sqlalchemy.events.ConnectionEvents.after_cursor_execute

class QueryLogger:
    """Log query duration and SQL as a context manager."""

    def __init__(self,
                engine: sqlalchemy.engine.Engine,
                f: io.StringIO):
        """
        Initialize for an engine and file.
        engine: The sqlalchemy engine for which events should be logged.
                You can pass the class `sqlalchemy.engine.Engine` to capture all engines
        f: file you want to write your output to
        """
        self.engine = engine
        self.file = f

    def _after_cursor_execute(self, conn, cursor, statement, parameters, context, executemany):
        """Listen for the 'after_cursor_execute' event and log sqlstatement and time."""
        # check if it's a ddl operation create_all execute a bunch of select statements 
        if context.isddl:
            s = statement % parameters
            self.file.write(f"{s};")

    def __enter__(self, *args, **kwargs):
        """Context manager."""
        if isinstance(self.engine, sqlalchemy.engine.Engine):
            sqlalchemy.event.listen(self.engine, "after_cursor_execute", self._after_cursor_execute)
        return self

    def __exit__(self, *args, **kwargs) -> None:
        """Context manager."""
        if isinstance(self.engine, sqlalchemy.engine.Engine):
            sqlalchemy.event.remove(self.engine, "after_cursor_execute", self._after_cursor_execute)

然后您可以使用上下文管理器将查询记录到内存文件中以写入 SQL

with open("x.sql", "w") as f:
    with QueryLogger(db.engine, f):
        db.create_all()

大部分代码的灵感来自

编译 SQL of `create_all`

Get compiled SQL of `create_all`

python

sqlalchemy

flask-sqlalchemy

alembic