Jak mogę sprawdzić połączenia zbiorcze w SQLAlchemy przed przekazaniem ich do mojego kodu aplikacji?

Question

Mamy nieco zawodny serwer bazy danych, z różnych powodów, aw konsekwencji czasami połączenia z bazą danych używane przez moją aplikację znikają pod nim. Połączenia są połączeniami SQLAlchemy 0.6.5 z bazą danych PostgreSQL w środowisku wykonawczym Pylons 1.0.

To, czego chcę, to sposób na złapanie większości z nich bez widocznego błędu; najlepiej, gdybym przetestował połączenie na poziomie basenu przed zwróceniem go z silnika. Kontroluję tworzenie silnika, więc jestem w porządku.

Jaki jest najlepszy (najbardziej idomatyczny/najczystszy) sposób, aby to osiągnąć? Zdaję sobie sprawę, że zawsze istnieje możliwość zgonu między testem a użyciem, ale to będzie dość rzadkie w tym środowisku i dlatego nie jest to dla mnie problemem.

Answer 1

Można użyć słuchacza Basen:

class ConnectionChecker(sqlalchemy.interfaces.PoolListener): 
    def checkout(self, dbapi_con, con_record, con_proxy): 
     if not is_valid_connection(dbapi_con): 
      # a new connection will be used 
      raise sqlalchemy.exc.DisconnectionError 

wyjechał do Ciebie jest jak zaimplementować is_valid_connection dla przypadku użycia.