在 Python 中管理连接创建？

Question

应用程序通常需要连接到其他服务（数据库、缓存、API 等）。出于理智和 DRY 的考虑，我们希望将所有这些连接保留在一个模块中，以便我们的代码库的其余部分可以共享连接。

为了减少样板，下游使用应该很简单：

# app/do_stuff.py
from .connections import AwesomeDB

db = AwesomeDB()

def get_stuff():
    return db.get('stuff')

而且设置连接也应该很简单：

# app/cli.py or some other main entry point
from .connections import AwesomeDB

db = AwesomeDB()
db.init(username='stuff admin')    # Or os.environ['DB_USER']

像 Django 和 Flask 这样的 Web 框架会做这样的事情，但感觉有点笨拙：

Connect to a Database in Flask, Which Approach is better? http://flask.pocoo.org/docs/0.10/tutorial/dbcon/

这样做的一个大问题是我们希望引用实际的连接对象而不是代理，因为我们希望在 iPython 和其他开发环境中保留制表符补全。

那么正确的方法（tm）是什么？经过几次迭代，这是我的想法：

#app/connections.py
from awesome_database import AwesomeDB as RealAwesomeDB
from horrible_database import HorribleDB as RealHorribleDB


class ConnectionMixin(object):
    __connection = None

    def __new__(cls):
        cls.__connection = cls.__connection or object.__new__(cls)
        return cls.__connection

    def __init__(self, real=False, **kwargs):
        if real:
            super().__init__(**kwargs)

    def init(self, **kwargs):
        kwargs['real'] = True
        self.__init__(**kwargs)


class AwesomeDB(ConnectionMixin, RealAwesomeDB):
    pass


class HorribleDB(ConnectionMixin, RealHorribleDB):
    pass

改进空间：将初始 __connection 设置为通用 ConnectionProxy 而不是 None，它会捕获所有属性访问并引发异常。

我在这里做了很多关于 SO 和各种 OSS 项目的探索，但还没有见过这样的事情。感觉非常可靠，尽管这确实意味着一堆模块将实例化连接对象作为导入时的副作用。这会在我脸上炸开吗？这种方法还有其他负面影响吗？

Answer 1

首先，在设计方面，我可能遗漏了一些东西，但我不明白为什么你需要繁重的 mixin+singleton 机制，而不是像这样定义一个帮助器：

_awesome_db = None
def awesome_db(**overrides):
    global _awesome_db
    if _awesome_db is None:
        # Read config/set defaults.
        # overrides.setdefault(...)
        _awesome_db = RealAwesomeDB(**overrides)
    return _awesome_db

此外，还有一个错误可能看起来不像受支持的用例，但无论如何：如果您连续进行以下 2 次调用，即使您传递了不同的参数，您也会错误地获得两次相同的连接对象：

db = AwesomeDB()
db.init(username='stuff admin')

db = AwesomeDB()
db.init(username='not-admin')    # You'll get admin connection here.

一个简单的解决方法是使用输入参数的连接字典。

现在，关于问题的本质。

我认为答案取决于您的“连接”类是如何实际实现的。

我认为您的方法的潜在缺点是：

• 在多线程环境中，您可能会遇到从多个线程对全局连接对象进行非同步并发访问的问题，除非它已经是线程安全的。如果您关心这一点，您可以稍微更改您的代码和接口并使用线程局部变量。

• 如果进程在创建连接后分叉怎么办？ Web 应用程序服务器往往会这样做，这可能并不安全，这又取决于底层连接。

• 连接对象有状态吗？如果连接对象变得无效（由于连接错误/超时）会发生什么？您可能需要用新的连接替换断开的连接，以便在下次请求连接时返回。

连接管理通常已经通过客户端库中的connection pool 有效且安全地实施。

例如redis-py Redis客户端使用如下实现：

https://github.com/andymccurdy/redis-py/blob/1c2071762ad9b9288e786665990083e61c1cf355/redis/connection.py#L974

Redis 客户端然后像这样使用连接池：

• Requests a connection 来自连接池。
• Tries to execute a command 在连接上。
• 如果连接失败，客户端closes it。
• 无论如何，最终它是returned to the connection pool，因此它可以被后续调用或其他线程重用。

因此，由于 Redis 客户端在后台处理所有这些，因此您可以安全地直接做您想做的事情。连接将被延迟创建，直到连接池达到最大容量。

# app/connections.py
def redis_client(**kwargs):
    # Maybe read configuration/set default arguments
    # kwargs.setdefault()
    return redis.Redis(**kwargs)

同样，SQLAlchemy 可以使用connection pooling as well。

总结一下，我的理解是：

• 如果您的客户端库支持连接池，则无需执行任何特殊操作即可在模块甚至线程之间共享连接。您可以只定义一个类似于redis_client() 的助手来读取配置或指定默认参数。

• 如果您的客户端库仅提供低级连接对象，您将需要确保对它们的访问是线程安全和 fork 安全的。此外，您需要确保每次返回有效连接（如果无法建立或重用现有连接，则引发异常）。