我们的系统运行在Ubuntu, python 3.4, postgres 9.4。X和psycopg2.
我们(将来会)使用模式在dev
, test
和prod
环境之间进行划分。我创建了一个方便的方法来创建到数据库的连接。它使用json连接配置文件来创建连接字符串。我想将连接配置为对使用返回连接的所有后续查询使用特定模式。我不希望我的查询有硬编码模式,因为我们应该能够根据我们是在开发,测试还是生产阶段/环境轻松地在它们之间切换。
当前的方便方法如下所示:
def connect(conn_config_file = 'Commons/config/conn_commons.json'):
with open(conn_config_file) as config_file:
conn_config = json.load(config_file)
conn = psycopg2.connect(
"dbname='" + conn_config['dbname'] + "' " +
"user='" + conn_config['user'] + "' " +
"host='" + conn_config['host'] + "' " +
"password='" + conn_config['password'] + "' " +
"port=" + conn_config['port'] + " "
)
cur = conn.cursor()
cur.execute("SET search_path TO " + conn_config['schema'])
return conn
只要您给它时间执行set search_path
查询,它就可以正常工作。不幸的是,如果我执行以下查询的速度太快,则会在没有设置search_path
的情况下发生竞争条件。我试图在return conn
之前强制执行conn.commit()
,但是,这会将search_path
重置为默认模式postgres
,因此它不会使用prod
。数据库或应用程序层的建议是可取的,但是,我知道我们可能也可以在操作系统级别解决这个问题,在这个方向上的任何建议也是受欢迎的。
一个json配置文件示例如下:
{
"dbname": "thedatabase",
"user": "theuser",
"host": "localhost",
"password": "theusers_secret_password",
"port": "6432",
"schema": "prod"
}
我认为更优雅的解决方案是在connect()
的options
参数中设置search_path
,如下所示:
def connect(conn_config_file = 'Commons/config/conn_commons.json'):
with open(conn_config_file) as config_file:
conn_config = json.load(config_file)
schema = conn_config['schema']
conn = psycopg2.connect(
dbname=conn_config['dbname'],
user=conn_config['user'],
host=conn_config['host'],
password=conn_config['password'],
port=conn_config['port'],
options=f'-c search_path={schema}',
)
return conn
当然,你可以使用"options"作为连接字符串的一部分。但是使用关键字参数可以避免字符串连接的所有麻烦。
我在这个psycopg2特性请求中找到了这个解决方案。至于"options"参数本身,这里已经提到了。
我认为一个更好的主意是有一些像DatabaseCursor返回游标,您使用"SET search_path…"而不是连接来执行查询。我的意思是这样的:
class DatabaseCursor(object):
def __init__(self, conn_config_file):
with open(conn_config_file) as config_file:
self.conn_config = json.load(config_file)
def __enter__(self):
self.conn = psycopg2.connect(
"dbname='" + self.conn_config['dbname'] + "' " +
"user='" + self.conn_config['user'] + "' " +
"host='" + self.conn_config['host'] + "' " +
"password='" + self.conn_config['password'] + "' " +
"port=" + self.conn_config['port'] + " "
)
self.cur = self.conn.cursor()
self.cur.execute("SET search_path TO " + self.conn_config['schema'])
return self.cur
def __exit__(self, exc_type, exc_val, exc_tb):
# some logic to commit/rollback
self.conn.close()
和
with DatabaseCursor('Commons/config/conn_commons.json') as cur:
cur.execute("...")