如何使用 SQLAlchemy 在运行时执行用户定义的 SQL 查询？

Question

我有一个用例实现，我试图确定解决方案，实现如下：

• 用户定义 SQL 服务器连接参数，如 DATABASE_TYPE（如 MySQL、pgsql、sqlite 或 SQLAlchemy 支持的任何数据库）、HOST、PORT、 SCHEMA_NAME、USER_NAME 和 PASSWORD
• 用户然后执行测试连接以验证提供的连接参数，并提供成功和失败反馈。
• 成功时，用户将键入sql 查询（以下 SQLAlchemy 的 SQL 表达式语言更多详细信息此处）需要被处决。

如果您注意到上面 SQLAlchemy 工作的所有参数都是用户在运行时定义的，我正在尝试在 python 中实现该函数，该函数使用上述连接参数和 sql 查询，在成功连接时执行查询并返回查询结果。对此的任何指导或帮助都会很棒。

谢谢

Answer 1

我有这样的东西作为一个更大项目的一部分，到目前为止仅适用于 sqlite 和 postgres，但这可以概括。

首先，如果您希望用户知道 DATABASE_TYPE（例如 MySQL、pgsql、sqlite 或 SQLAlchemy 支持的任何数据库）、HOST、PORT、SCHEMA_NAME、USER_NAME 和 PASSWORD，您可以只询问 DB URL，以使您的生活更轻松。

之后，您可以通过以下方式测试连接：

    from sqlalchemy import create_engine
    try:
        engine = create_engine(url)           
        connection = engine.connect()
        connection.close()
    except Exception, e:
        LOGGER.exception(e)
        raise MyCusomException('Could not initialize db. Invalid URL.') 

可以使用 text() 构造执行 Cusom sql 查询，如下所示 此处。