2009-10-21 16 views
7

Chcę, aby mój kod automatycznie próbował różnych sposobów tworzenia połączenia z bazą danych. Gdy tylko zadziała, kod musi przejść (tzn. Nie powinien już próbować na inne sposoby). Jeśli wszystko zawiedzie, skrypt może po prostu wysadzić w powietrze.Python idiom dla "Spróbuj, dopóki nie zostanie zgłoszony wyjątek"

Więc - co moim zdaniem było, ale najprawdopodobniej nie jest - genialny Próbowałem to:

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 

Ale wydaje się, że drugi i kolejny oprócz oświadczenia nie są łapanie OperationalErrors już. Prawdopodobnie dlatego, że Python przechwytuje wyjątek tylko raz w instrukcji try ... except?

Czy to prawda? Jeśli nie: czy coś jeszcze robię źle? Jeśli tak, to jak to zrobić? Czy istnieje standardowy idiom?

(wiem, że istnieją sposoby obejścia tego problemu, jak posiadanie użytkownikowi określić parametry połączenia w linii poleceń, ale to nie moja sprawa ok :))

EDIT:

Przyjąłem Doskonała odpowiedź retracile i wziąłem komentarz gnibblera za użycie konstruktu for..else. Ostateczny kod stał się (przykro mi, nie podążam za maksymalnymi znakami na limit linii od pep8):

EDYCJA 2: Jak widać z komentarza na temat klasy Cursor: tak naprawdę nie wiem jak nazwać ten rodzaj klasy. Nie jest to singleton (mogę mieć wiele różnych instancji Cursora), ale podczas wywoływania get_cursor, za każdym razem otrzymuję ten sam obiekt kursora. Więc to jest jak singletowa fabryka? :)

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

Odpowiedz

14

przybliżeniu:

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

Teraz, jeśli nie chcemy, aby zadzwonić getpass() chyba że jest to konieczne, to chciałby, aby sprawdzić if 'password' in attempt: attempt['password'] = getpass() lub tak.

Teraz o tym globalnym ....

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

... choć myślę, że istnieje kilka innych sposobów, aby osiągnąć to samo.

Bringing to wszystko razem:

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

Uwaga: całkowicie niesprawdzone

+3

Zamiast inicjalizacji 'conn' do' None', można po prostu użyć 'else' klauzulę o' for' –

+0

+1 @David: przerwie zapewnia, że ​​zwracane jest pierwsze działające połączenie. To rozwiązanie będzie skalować się z większą liczbą baz danych, a łańcuchy połączenia zostały ładnie wyodrębnione z kodu. –

+1

@gnibbler - Woah, zapomniałem Python miał tę niesamowitą funkcję (http://docs.python.org/reference/compound_stmts.html # the-for-statement) –

-1

Jesteś blisko. Prawdopodobnie najlepszą rzeczą do zrobienia w tym przypadku jest zagnieżdżanie drugiej i kolejnych prób w bloku except. Tak więc kluczowym elementem kodu wyglądałby następująco:

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

-1 Ugh, całkowicie nie pytoniczny. Sprawia, że ​​wyglądasz jak programista C++ COM. –

+0

Całkiem brzydka, tak. – grasshopper

Powiązane problemy