replication.py 文件源码

def get_replication_status(db):
    result = {"result": False, "status": None}
    db_conn = connect('Destination', db_name=db)
    src_db_conn = connect('Source', db_name=db)
    result["result"] = False
    try:
        cur = db_conn.cursor(cursor_factory=extras.DictCursor)
        cur.execute("SELECT status FROM pglogical.show_subscription_status(subscription_name := 'subscription');")
        r = cur.fetchone()
        if r:
            result["result"] = True
            result["status"] = r['status']

    except psycopg2.InternalError:
        result["result"] = False
    except psycopg2.OperationalError:
        result["result"] = False
    except psycopg2.ProgrammingError:
        result["result"] = False

    return result