SQLAlchemy Session try~catch

簡単なサンプルで、、、
まずは、Session使用メソッドの呼び出し側で捕捉

from sqlalchemy import create_engine, text
from sqlalchemy.orm import Session

# 接続情報、username, password, host, port, dbname は設定済
engine = create_engine(f'postgresql://{username}:{password}@{host}:{port}/{dbname}')

def countquery() -> int:
    with Session(autocommit=False, autoflush=True, bind=engine) as session:
        print("---session---")
        sql = 'SELECT COUNT(*) FROM samples'
        rows = session.execute(text(sql))
        return list(rows)[0][0]

if __name__ == '__main__':
    print("---START---")
    try:
        c = countquery()
        print('count = %d' % c)
    except Exception as e:
        print(e)
    else:
        print("OK")
    finally:
        print("---END---")

正常時

---START---
---session---
count = 6
OK
---END---

DB接続異常時

---START---
---session---
(psycopg2.OperationalError) connection to server at "localhost" (::1), port 5432 failed: Connection refused (0x0000274D/10061)
	Is the server running on that host and accepting TCP/IP connections?
connection to server at "localhost" (127.0.0.1), port 5432 failed: Connection refused (0x0000274D/10061)
	Is the server running on that host and accepting TCP/IP connections?

(Background on this error at: https://sqlalche.me/e/20/e3q8)
---END---

これが示しているのは
　with Session(autocommit=False, autoflush=True, bind=engine) as session:
の後の session の execute で実行してから接続エラーを検出していることである。
SQLの誤り

---START---
---session---
(psycopg2.errors.UndefinedTable) リレーション"asamples"は存在しません
LINE 1: SELECT COUNT(*) FROM asamples
                             ^

[SQL: SELECT COUNT(*) FROM asamples]
(Background on this error at: https://sqlalche.me/e/20/f405)
---END---

今度は、with Session　～ の中で、try-catch を書いてみる。

engine = create_engine(f'postgresql://{username}:{password}@{host}:{port}/{dbname}')

def countquery() -> int:
    with Session(autocommit=False, autoflush=True, bind=engine) as session:
        print("---session---")
        try:
            sql = 'SELECT COUNT(*) FROM samples'
            rows = session.execute(text(sql))
        except Exception as e:
            print(e)
            raise
        else:
            return list(rows)[0][0]

if __name__ == '__main__':
    print("---START---")
    c = countquery()
    print('count = %d' % c)
    print("---END---")

except Exception: でraise 実行しているので、呼び出し側では try-catch を書かなければ、
呼び出し側に戻ることなく、そこで異常終了になる。