我收到“AttributeError: 'NoneType'对象没有'_instantiate_plugins'属性”错误

Question

我有一个简单的Flask应用，按照一个简单的教程做的，现在我想把数据库的连接信息从代码中拿出来，放到环境变量里。我在根目录下的.env文件中定义了DB_CONNECTION_STR。下面是我database.py文件的开头部分。这应该是个简单的修改，可能我漏掉了什么明显的错误。

from sqlalchemy import create_engine, text
import os

DB_CONNECTION_STR = os.environ.get('DB_CONNECTION_STR')

# These both print the right value:
print(DB_CONNECTION_STR)
print(os.environ.get('DB_CONNECTION_STR'))

# This works:
db_connection_string = "mysql+pymysql://ehcazu8mvtvv4i44ghs3:pscale_pw_i960dmnsHwOCf3IWmKZi4AzOMVi1ZueGLYRiPZAy3i9@aws.connect.psdb.cloud/joviancareers?charset=utf8mb4"

# Trying either of these two gives me a "AttributeError: 'NoneType' object has no attribute '_instantiate_plugins'"
# db_connection_string = DB_CONNECTION_STR
# db_connection_string = os.environ.get('DB_CONNECTION_STR')

engine = create_engine(db_connection_string, connect_args={
  "ssl": {
    "ca": "/etc/ssl/certs/ca-certificates.crt"
  }
})

这是我遇到的错误：

Traceback (most recent call last):
  File "/home/taylor/src/jovian-carrers/app.py", line 2, in <module>
    from database import load_jobs
  File "/home/taylor/src/jovian-carrers/database.py", line 17, in <module>
    engine = create_engine(db_connection_string,
  File "<string>", line 2, in create_engine
  File "/home/taylor/src/jovian-carrers/venv/lib/python3.10/site-packages/sqlalchemy/util/deprecations.py", line 298, in warned
    return fn(*args, **kwargs)
  File "/home/taylor/src/jovian-carrers/venv/lib/python3.10/site-packages/sqlalchemy/engine/create.py", line 518, in create_engine
    u, plugins, kwargs = u._instantiate_plugins(kwargs)
AttributeError: 'NoneType' object has no attribute '_instantiate_plugins'

Answer 1

我纠结了一个小时，终于在发完问题后一下子明白了。其实我需要把设置引擎的代码放在函数里面。我猜想当代码在函数外面时，它无法访问环境变量，因为应用在编译的时候需要这些变量。

这个方法不行：

DB_CONNECTION_STR = os.environ.get('DB_CONNECTION_STR')

db_connection_string = DB_CONNECTION_STR

engine = create_engine(db_connection_string, connect_args = {
  "ssl": {
    "ca": "/etc/ssl/certs/ca-certificates.crt"
  }
})

def load_jobs():
  with engine.connect() as conn:
    result = conn.execute(text("SELECT * FROM jobs"))
    jobs = [dict(row) for row in result.all()]
    return jobs

这个方法可以，但需要写死的凭证：

db_connection_string = "mysql+pymysql://ehcazu8mvtvv4i44ghs3:pscale_pw_i960dmnsHwOCf3IWmKZi4AzOMVi1ZueGLYRiPZAy3i9@aws.connect.psdb.cloud/joviancareers?charset=utf8mb4"

engine = create_engine(db_connection_string, connect_args = {
  "ssl": {
    "ca": "/etc/ssl/certs/ca-certificates.crt"
  }
})

def load_jobs():
  with engine.connect() as conn:
    result = conn.execute(text("SELECT * FROM jobs"))
    jobs = [dict(row) for row in result.all()]
    return jobs

这个方法可以，不需要写死的凭证：

def load_jobs():
  DB_CONNECTION_STR = os.environ.get('DB_CONNECTION_STR')
  db_connection_string = DB_CONNECTION_STR
  engine = create_engine(db_connection_string, connect_args = {
    "ssl": {
      "ca": "/etc/ssl/certs/ca-certificates.crt"
    }
  })
  with engine.connect() as conn:
    result = conn.execute(text("SELECT * FROM jobs"))
    jobs = [dict(row) for row in result.all()]
    return jobs