用SQLAlchemy模拟连接错误

Question

在生产环境中，我的Flask/SQLAlchemy应用程序会随机抛出psycopg2.OperationalError: server closed the connection unexpectedly异常，这是在RDS Aurora数据库上发生的。在我找到问题之前，我希望我的单元测试能够确保我可以正确处理此问题，例如，我的回滚机制是否有效等。

目前，我使用虚拟的异常来模拟Session.commit方法的行为，但我不确定（afaik？）是否留下了需要实际回滚的失败状态。

有哪些可靠的方法来模拟本地Docker Compose开发环境中的实际连接故障？

- infojunkie

---

你有没有找到解决办法，@infojunkie？我也陷入同样的困境。 - ghukill

很遗憾 - 我已经解决了这个问题。 - infojunkie

Answer 1

使用psycopg2进行回滚操作

conn = psycopg2.connect(
    f"host='{credentials['host']}' port={credentials['port']} dbname='{credentials['dbname']}' user='{credentials['user']}' password='{credentials['password']}'"
)
print("Connected to database")
print("Setting autocommit false")
conn.autocommit = False

func_name= 'procedure_update_users'

print(f"Executing function {func_name}...")
try:
    cur = conn.cursor()
    query = f"""SELECT {func_name}();"""
    print(query)
    cur.execute(query)
    row = cur.fetchone()
    conn.commit()
    print(f"Function {func_name} executed.")
except Exception as e:
    print(e)
    conn.rollback()
    raise Exception(e)

获取OperationalError的方法如下：

我这样做：

try:
    cur.execute("LOCK TABLE mytable IN ACCESS EXCLUSIVE MODE NOWAIT")
except psycopg2.OperationalError as e:
    cur.execute("rollback")
    raise Exception(e)

psycopg2.OperationalError: fe_sendauth:未提供密码

- Ruscinc

---

谢谢，但我不明白这如何模拟连接错误。 - infojunkie