db.py 文件源码

def connection_pool(max_conn, dsn_dict: Dict[str, str], application_name=psycopg2.__name__):
    """
    Create a connection pool (with up to max_conn connections), where all connections will use the
    given connection string.
    """
    dsn_values = _dsn_connection_values(dsn_dict, application_name)
    return psycopg2.pool.ThreadedConnectionPool(1, max_conn, **dsn_values)