在python中创建连接到postgres数据库的方法答案

【问题标题】：Creating a method to connect to postgres database in python在python中创建连接到postgres数据库的方法
【发布时间】：2021-08-04 18:54:59
【问题描述】：

我正在开发一个 python 程序，其功能包括使用 psycopg2 从 postgres 数据库中插入和检索值。问题是每次我想创建一个查询时，我都必须连接到数据库，因此以下代码 sn-p 在整个文件中多次出现：

# Instantiate Connection
    try:
        conn = psycopg2.connect(
            user=userName,
            password=passwrd,
            host=hostAddr,
            database=dbName
        )

        # Instantiate Cursor
        cur = conn.cursor()
        return cur

    except psycopg2.Error as e:
        print(f"Error connecting to Postgres Platform: {e}")
        sys.exit(1)

我的问题是：

有没有一种方法可以创建一个方法来在每次我希望连接到数据库时调用？我尝试创建一个，但由于变量 cur 和 conn 不是全局变量，我得到了一堆错误
我可以在程序开始时只连接一次数据库并在程序运行的整个过程中保持连接打开吗？这似乎是最简单的选择，但我不确定这是否是不好的做法（作为参考，程序将 24/7 运行，所以我认为最好只在进行查询时进行连接）。

感谢您的帮助。

【问题讨论】：

标签： python postgresql psycopg2

【解决方案1】：

您可以将自己的数据库处理类包装在上下文管理器中，这样您就可以在一个地方管理连接：

import psycopg2
import traceback
from psycopg2.extras import RealDictCursor


class Postgres(object):
    def __init__(self, *args, **kwargs):
        self.dbName = args[0] if len(args) > 0 else 'prod'
        self.args = args

    def _connect(self, msg=None):
        if self.dbName == 'dev':
            dsn = 'host=127.0.0.1 port=5556 user=xyz password=xyz dbname=development'
        else:
            dsn = 'host=127.0.0.1 port=5557 user=xyz password=xyz dbname=production'

        try:
            self.con = psycopg2.connect(dsn)
            self.cur = self.con.cursor(cursor_factory=RealDictCursor)
        except:
            traceback.print_exc()

    def __enter__(self, *args, **kwargs):
        self._connect()
        return (self.con, self.cur)

    def __exit__(self, *args):
        for c in ('cur', 'con'):
            try:
                obj = getattr(self, c)
                obj.close()
            except:
                pass # handle it silently!?
        self.args, self.dbName = None, None

用法：

with Postgres('dev') as (con, cur):
    print(con)
    print(cur.execute('select 1+1'))
print(con) # verify connection gets closed!

输出：

<connection object at 0x109c665d0; dsn: '...', closed: 0>
[RealDictRow([('sum', 2)])]
<connection object at 0x109c665d0; dsn: '...', closed: 1>

【讨论】：

【解决方案2】：

保持连接打开应该不会太糟糕。服务器本身应该负责关闭它认为已经存在太久或太不活动的连接。然后我们只需要让我们的代码在服务器关闭连接的情况下具有弹性：

import pscyopg2

CONN = None

def create_or_get_connection():
    global CONN
    if CONN is None or CONN.closed:
        CONN = psycopg2.connect(...)
    return CONN

【讨论】：

【解决方案3】：

我以前曾多次走这条路，而您可能正在重新发明轮子。我强烈建议您使用像 [Django][1] 这样的 ORM，或者如果您需要与数据库交互 - 它使用最佳实践为您处理所有这些事情。这是一些预先学习，但我保证会有所回报。

如果您不想使用 Django，可以使用此代码来获取或创建连接和游标的上下文管理器，以避免出现错误

import pscyopg2

CONN = None

def create_or_get_connection():
    global CONN
    if CONN is None or CONN.closed:
        CONN = psycopg2.connect(...)
    return CONN

def run_sql(sql):
    con = create_or_get_connection()
    with conn.cursor() as curs:
        return curs.execute(sql)

这将允许您简单地将 sql 语句直接运行到数据库，而无需担心连接或游标问题。

【讨论】：

【解决方案4】：

如果我将您的代码片段包装到函数定义中，我不会收到“一堆错误，因为变量 cur 和 conn 不是全局的”。为什么他们需要全球化？无论错误是什么，您在发布之前将其从代码片段中删除。

你的 try-catch 对我没有任何意义。捕获错误只是为了隐藏调用站点然后退出似乎是有益的反面。

何时连接取决于您构建事务的方式、执行它们的频率以及如果您的数据库在程序执行过程中重新启动时您想要做什么。

【讨论】：