sqlalchemy通過ssh連接遠程mysql服務器
阿新 • • 發佈:2018-03-30
mat ssh連接 post try 繼承 listen IT raise mysql數據庫
首先需要一個模塊sshtunnel,如果沒有直接pip install sshtunnel
from sshtunnel import SSHTunnelForwarder from sqlalchemy import Column, String, Integer, create_engine, event from sqlalchemy.orm import sessionmaker from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.exc import DisconnectionError ssh_host = "" # 堡壘機ip地址或主機名 ssh_port = 22 # 堡壘機連接mysql服務器的端口號,一般都是22,必須是數字 ssh_user = "" # 這是你在堡壘機上的用戶名 ssh_password = "" # 這是你在堡壘機上的用戶密碼 mysql_host = "" # 這是你mysql服務器的主機名或ip地址 mysql_port = 3306 # 這是你mysql服務器上的端口,3306,mysql就是3306,必須是數字 mysql_user = "" # 這是你mysql數據庫上的用戶名 mysql_password = "" # 這是你mysql數據庫的密碼 mysql_db = "" # mysql服務器上的數據庫名 Base = declarative_base() class Phones(Base): __tablename__ = ‘phones‘ id = Column(Integer, primary_key=True) name = Column(String(32)) def checkout_listener(dbapi_con, con_record, con_proxy): try: try: dbapi_con.ping(False) except TypeError: dbapi_con.ping() except dbapi_con.OperationalError as exc: if exc.args[0] in (2006, 2013, 2014, 2045, 2055): raise DisconnectionError() else: raise with SSHTunnelForwarder( (ssh_host, ssh_port), ssh_username=ssh_user, ssh_password=ssh_password, remote_bind_address=(mysql_host, mysql_port) ) as server: server.start() # ssh通道服務啟動 local_port = str(server.local_bind_port) engine = create_engine( ‘mysql+pymysql://{}:{}@{}:{}/{}?charset=utf8‘.format(mysql_user, mysql_password, ‘127.0.0.1‘, local_port,mysql_db), pool_size=100, pool_recycle=3600) event.listen(engine, ‘checkout‘, checkout_listener) # 防止報連接池相關的錯誤 Base.metadata.create_all(engine) # 檢測文件中所有繼承了Base類的類,在mysqld中建立所有的表,類就是表 Session = sessionmaker(bind=engine) session = Session()
sqlalchemy通過ssh連接遠程mysql服務器