Python3でPostgreSQL接続(SSH)

で、選択のストアドプロシジャ実行のソースを変更

#!/usr/bin/env python3
# import sys
import psycopg2
import psycopg2.extras
import datetime

from sqlalchemy import create_engine
from sshtunnel import SSHTunnelForwarder

print('START : ' + '{0:%Y/%m/%d %H:%M:%S.%f}'.format(datetime.datetime.now()))
try:
    server = SSHTunnelForwarder((SV名, 22), ssh_username=UID, ssh_password=P/W, ssh_private_key_password=p/w, ssh_pkey=秘密鍵, remote_bind_address=('127.0.0.1', 5432))
    try:
        server.start()

        try:
            strCon = "postgresql+psycopg2://UID:P/W@127.0.0.1:" + '{0:5d}'.format(server.local_bind_port) + "/DB名"
            print(strCon)
            engine = create_engine(strCon, client_encoding="utf8")
            try:
                con = engine.raw_connection()
                ##con = psycopg2.connect(host = SV名, port = 5432, database = DB名, user = UID, password = P/W)

                try:
                    cur = con.cursor(cursor_factory = psycopg2.extras.DictCursor)

                    try:
                        cur.execute('call pr_mst_item_select_00 (0, 99, 200, 299, null, null, null, null)')
                        row = cur.fetchone()

                        for intValue in row['p_num']:
                            print('{0:+09}'.format(intValue))

                        for strValue in row['p_msg']:
                            print(strValue)

                        if row['p_result'] == 0:
                            cur.execute(f'FETCH ALL IN "{row["p_cur"]}"')
                            rows = cur.fetchall()
                        else:
                            rows = None

                        if rows is None:
                            print('データ無！')
                        else:
                            for row in rows:
                                ret = '{0:08}'.format(row['item']) + ' : ' + row['name'] + ' : ' + row['kana']
                                print(ret)

                        con.rollback()
                        print('OK')

                    except (Exception, psycopg2.DatabaseError) as err:
                        print(err)
                        con.rollback()

                    except Exception as err:
                        print(err)
                        con.rollback()

                    finally:
                        cur.close()

                except (Exception, psycopg2.DatabaseError) as err:
                    print(err)

                except Exception as err:
                    print(err)

                finally:
                    con.close()

            except Exception as err:
                print(err)

        except Exception as err:
            print(err)

        finally:
            server.close()

    except Exception as err:
        print(err)

except Exception as err:
    print(err)

finally:
    print('FINISH : ' + '{0:%Y/%m/%d %H:%M:%S.%f}'.format(datetime.datetime.now()))
 

Python3でPostgreSQL接続(SSH)

Windows10のPCからSSHトンネルでPostgreSQLデータベースへ接続

そのためにPython環境を作る

先ずはココからダウンロードして

https://www.python.org/ftp/python/3.11.1/python-3.11.1-amd64.exe

インストール

次にPIPをインストール

https://bootstrap.pypa.io/get-pip.py

を開くとソースが表示されたので、すべて選択して

テキトーなファイル名「tekito.py」で保存

コマンドプロンプトを管理者で実行しファイル保存先の

ディレクトリへ移動し、メイクする

python3 tekito.py

PIPインストールが成功したらSSH接続に必要な

モジュールをPIPでインストールする

pip install sshtunnel

pip install sqlalchemy

Python3でPostgreSQL接続(SSH)

PIPのインストール

$ sudo -i

# dnf install -y epel-release

# dnf install -y python3-pip

インストール確認

# pip --version

PIPインストールが成功したらSSH接続に必要な

モジュールをPIPでインストールする

# pip install sshtunnel

# pip install sqlalchemy

で、テキトーにソースをコーディング

print('START : ' + '{0:%Y/%m/%d %H:%M:%S.%f}'.format(datetime.datetime.now()))

try:
    server = SSHTunnelForwarder(('SV名', 22), ssh_username='UID', ssh_password='P/W', ssh_private_key_password='秘密鍵P/W', ssh_pkey='秘密鍵', remote_bind_address=('localhost', 5432))

    try:
        server.start()
        print('ローカルバインドポート : {0:5d}'.format(server.local_bind_port))
        server.close()

    except Exception as err:
        print(err)

except Exception as err:
    print(err)

finally:
    print('FINISH : ' + '{0:%Y/%m/%d %H:%M:%S.%f}'.format(datetime.datetime.now()))

名前を付けて保存して実行

$ python3 ファイル名.py

ローカルバインドポートが数値表示されるのでヨシ？