我希望我的代码能够自动尝试多种方式来创建数据库连接.一旦工作,代码需要继续前进(即它不应该再尝试其他方式).如果它们都失败了,那么脚本就会爆炸.
所以在 - 我的想法,但很可能不是 - 天才的一击,我试过这个:
import psycopg2
from getpass import getpass
# ouch, global variable, ooh well, it's just a simple script eh
CURSOR = None
def get_cursor():
"""Create database connection and return standard cursor."""
global CURSOR
if not CURSOR:
# try to connect and get a cursor
try:
# first try the bog standard way: db postgres, user postgres and local socket
conn = psycopg2.connect(database='postgres', user='postgres')
except psycopg2.OperationalError:
# maybe user pgsql?
conn = psycopg2.connect(database='postgres', user='pgsql')
except psycopg2.OperationalError:
# maybe it was postgres, but on localhost? prolly need password then
conn = psycopg2.connect(database='postgres', user='postgres', host='localhost', password=getpass())
except psycopg2.OperationalError:
# or maybe it was pgsql and on localhost
conn = psycopg2.connect(database='postgres', user='pgsql', host='localhost', password=getpass())
# allright, nothing blew up, so we have a connection
# now make a cursor
CURSOR = conn.cursor()
# return existing or new cursor
return CURSOR
Run Code Online (Sandbox Code Playgroud)
但似乎第二个和后续的except语句不再捕获OperationalErrors了.可能是因为Python只在try ... except语句中捕获一次异常?
是这样吗?如果没有:还有什么我做错了吗?如果是这样的话:你怎么做这样的事呢?有标准的习语吗?
(我知道有很多方法可以解决这个问题,比如让用户在命令行上指定连接参数,但这不是我的问题,确定:))
编辑:
我接受了retracile的优秀答案,我接受了gnibbler对使用for..else结构的评论.最后的代码变成了(抱歉,我并没有真正遵循pep8的每行最大字符数限制):
编辑2:正如您从Cursor类的注释中看到的那样:我真的不知道如何调用这种类.它不是真正的单例(我可以有多个不同的Cursor实例)但是在调用get_cursor时我每次都会得到相同的游标对象.那就像一个单身工厂?:)
import psycopg2
from getpass import getpass
import sys
class UnableToConnectError(Exception):
pass
class Cursor:
"""Cursor singleton factory?"""
def __init__(self):
self.CURSOR = None
def __call__(self):
if self.CURSOR is None:
# try to connect and get a cursor
attempts = [
{'database': 'postgres', 'user': 'postgres'},
{'database': 'postgres', 'user': 'pgsql'},
{'database': 'postgres', 'user': 'postgres', 'host': 'localhost', 'password': None},
{'database': 'postgres', 'user': 'pgsql', 'host': 'localhost', 'password': None},
]
for attempt in attempts:
if 'password' in attempt:
attempt['password'] = getpass(stream=sys.stderr) # tty and stderr are default in 2.6, but 2.5 uses sys.stdout, which I don't want
try:
conn = psycopg2.connect(**attempt)
attempt.pop('password', None)
sys.stderr.write("Succesfully connected using: %s\n\n" % attempt)
break # no exception raised, we have a connection, break out of for loop
except psycopg2.OperationalError:
pass
else:
raise UnableToConnectError("Unable to connect: exhausted standard permutations of connection dsn.")
# allright, nothing blew up, so we have a connection
# now make a cursor
self.CURSOR = conn.cursor()
# return existing or new cursor
return self.CURSOR
get_cursor = Cursor()
Run Code Online (Sandbox Code Playgroud)
ret*_*ile 14
大约:
attempts = [
{ 'database'='postgres', 'user'='pgsql', ...},
{ 'database'='postgres', 'user'='postgres', 'host'='localhost', 'password'=getpass()},
...
]
conn = None
for attempt in attempts:
try:
conn = psycopg2.connect(**attempt)
break
except psycopg2.OperationalError:
pass
if conn is None:
raise a ruckus
CURSOR = conn.cursor()
Run Code Online (Sandbox Code Playgroud)
现在,如果你不想打电话,getpass()除非有必要,你想要检查if 'password' in attempt: attempt['password'] = getpass()左右.
现在关于全球......
class MyCursor:
def __init__(self):
self.CURSOR = None
def __call__(self):
if self.CURSOR is None:
<insert logic here>
return self.CURSOR
get_cursor = MyCursor()
Run Code Online (Sandbox Code Playgroud)
......虽然我认为还有其他几种方法可以完成同样的事情.
将它们整合在一起:
class MyCursor:
def __init__(self):
self.CURSOR = None
def __call__(self):
if self.CURSOR is None:
attempts = [
{'database'='postgres', 'user'='postgres'},
{'database'='postgres', 'user'='pgsql'},
{'database'='postgres', 'user'='postgres', 'host'='localhost', 'password'=True},
{'database'='postgres', 'user'='pgsql', 'host'='localhost', 'password'=True},
]
conn = None
for attempt in attempts:
if 'password' in attempt:
attempt['password'] = getpass()
try:
conn = psycopg2.connect(**attempt)
break # that didn't throw an exception, we're done
except psycopg2.OperationalError:
pass
if conn is None:
raise a ruckus # nothin' worked
self.CURSOR = conn.cursor()
return self.CURSOR
get_cursor = MyCursor()
Run Code Online (Sandbox Code Playgroud)
注意:完全未经测试