Python идиома «Попробуй, пока не сработает исключение» - PullRequest
7 голосов
/ 22 октября 2009

Я хочу, чтобы мой код автоматически пробовал несколько способов создать соединение с базой данных. Как только кто-то работает, код должен двигаться дальше (то есть он больше не должен пытаться искать другие пути). Если все они не удаются, сценарий может просто взорваться.

Так что - то, что я думал, было, но, скорее всего, нет - гениальный удар, я попробовал это:

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

Но, похоже, что второе и последующее, кроме операторов, больше не перехватывают OperationalErrors. Возможно, потому что Python только один раз перехватывает исключение при попытке ... кроме оператора?

Это так? Если нет: что-то еще я делаю не так? Если так: как ты тогда делаешь что-то подобное? Есть ли стандартная идиома?

(я знаю, что есть способы обойти эту проблему, например, когда пользователь указывает параметры соединения в командной строке, но это не мой вопрос, хорошо :))

EDIT:

Я принял превосходный ответ 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()

Ответы [ 2 ]

14 голосов
/ 22 октября 2009

Приблизительно:

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

Теперь, если вы не хотите звонить 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()

... хотя я думаю, что есть несколько других способов сделать то же самое.

Собираем все вместе:

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

Примечание: полностью не проверено

0 голосов
/ 22 октября 2009

Ты рядом. Вероятно, лучшее, что нужно сделать в этом случае, - это вложить вторую и последующие попытки в блок исключений. Таким образом, критическая часть вашего кода будет выглядеть так:

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?
        try:
            conn = psycopg2.connect(database='postgres', user='pgsql')
        except psycopg2.OperationalError:
            # maybe it was postgres, but on localhost? prolly need password then
            try:
                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())
Добро пожаловать на сайт PullRequest, где вы можете задавать вопросы и получать ответы от других членов сообщества.
...