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

Question

我正在编写一个python程序，该程序具有使用psycopg2从postgres数据库插入和检索值等功能。问题是，每次我想要创建一个查询，我必须连接到数据库，所以以下代码片段在整个文件中多次出现:

# 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)

我的问题是:

1. 是否有一种方法，我可以只是创建一个方法来调用每次我想连接到数据库?我试过创建一个，但我得到了一堆错误，因为变量cur和conn不是全局的

2. 我可以在程序开始时连接到数据库一次，并在程序运行的整个时间保持连接打开吗?这似乎是最简单的选择，但我不确定这是否会是不好的做法(作为参考，该程序将运行24/7，所以我认为最好只在查询时连接)。

谢谢你的帮助。

Answer 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>

Answer 2

保持连接打开应该不会太糟糕。服务器本身应该负责关闭它认为存在太长时间或太不活跃的连接。然后，我们只需要使我们的代码在服务器关闭连接的情况下具有弹性:

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

Answer 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)

这将允许您直接向DB运行sql语句，而不必担心连接或游标问题。

Answer 4

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

你的试捕对我来说没有任何意义。捕获一个错误只是为了隐藏调用站点，然后退出似乎是相反的帮助。

何时连接取决于你如何构建你的事务，你多久做一次，如果你的数据库在程序执行过程中重新启动，你想做什么。