快捷導(dǎo)航

Python如何管理多個(gè)PostgreSQL數(shù)據(jù)庫的連接

更新時(shí)間：2024年11月12日 08:28:29 作者：jzy3711

這篇文章主要為大家詳細(xì)介紹了Python如何通過讀取配置文件中的PostgreSQL服務(wù)信息,連接到相應(yīng)的PostgreSQL數(shù)據(jù)庫,感興趣的小伙伴可以了解下

主要實(shí)現(xiàn)了通過讀取配置文件中的PostgreSQL服務(wù)信息，連接到相應(yīng)的PostgreSQL數(shù)據(jù)庫。提供了兩個(gè)連接函數(shù)，postgresql_connect 和 postgresql_connect_encode，分別是普通的和帶編碼參數(shù)的連接函數(shù)。

操作步驟

1.導(dǎo)入相關(guān)庫

from pyhive import hive
import cx_Oracle
import pymysql
import pymssql
from optparse import OptionParser
import logging
import sys
import traceback
import re
import warnings
with warnings.catch_warnings(record=True):
    import psycopg2
reload(sys)

2. 配置PostgreSQL連接信息

這里定義了一個(gè)包含多個(gè)PostgreSQL服務(wù)配置信息的列表pgconfigs。每個(gè)配置項(xiàng)由數(shù)據(jù)庫名稱、用戶、密碼、主機(jī)和端口組成。

pgconfigs = [
    {"asset_factory": {"database": "asset_factory", "user": "asset", "password": "Ro20ot16", "host": "10.251.80.202", "port": "65432"}},
    {"asset_register": {"database": "asset_register", "user": "asset", "password": "Ro20ot16", "host": "10.251.80.202", "port": "65432"}},
    {"creditdb": {"database": "creditdb", "user": "credit", "password": "hHJ98#pE40Y", "host": "10.251.101.175", "port": "18923"}},
    {"asset_portal": {"database": "asset_portal", "user": "asset", "password": "Ro20ot16", "host": "10.251.80.202", "port": "65432"}},
    {"tyjh": {"database": "tyjh", "user": "tyjh", "password": "k5y2dwoKcFm&^OsW", "host": "10.251.90.35", "port": "18922"}},
    {"cwrl": {"database": "cwrl", "user": "cwjh_yaxin", "password": "eKYy4R3&MbaaY3Zy", "host": "10.251.90.34", "port": "18921"}},
    {"dc_new_pg": {"database": "dc", "user": "dc", "password": "Ojjkcy@jVxKIeo5C", "host": "10.251.90.36", "port": "18923"}},
    {"credit_data": {"database": "credit_data", "user": "credit_data", "password": "@G*1f*3$3DzTH6%o", "host": "10.251.90.134", "port": "18923"}},
    {"new_jf_pg_cd": {"database": "accthuiju", "user": "datacenter", "password": "KridU593&%rj90", "host": "133.37.116.192", "port": "18921"}},
    {"new_accthuiju": {"database": "accthuiju", "user": "itf_ods", "password": "2dH~fZ^8", "host": "10.251.64.226", "port": "18921"}},
    {"crm3huijupg": {"database": "crm3huijupg", "user": "datacenter", "password": "lp*36^YD", "host": "10.251.65.93", "port": "18921"}},
    {"crm3hispg": {"database": "crm3hispg", "user": "datacenter", "password": "~4nD_jWG", "host": "10.251.64.196", "port": "18921"}},
    {"hana": {"database": "hana", "user": "hana", "password": "gvptHKXVNKEGw7pk", "host": "10.251.90.94", "port": "18924"}},
    {"dataos_71_pg_dev": {"database": "dacp", "user": "dacp", "password": "jxFgCKv9GJw2ohS3", "host": "10.251.110.104", "port": "18921"}}
]

3. 定義連接函數(shù)

普通連接函數(shù)

def postgresql_connect(servicename):
    global job_task_content_ori
    global job_resource_conn_info
    global job_hdfs_dir_info
    try:
        for subconfig in pgconfigs:
            if servicename in subconfig:
                pgconfig = subconfig[servicename]
                conn = psycopg2.connect(database=pgconfig["database"], user=pgconfig["user"],
                                        password=pgconfig["password"], host=pgconfig["host"], port=pgconfig["port"])
                return conn
        print("servicename %s not found in config dictionary" % servicename)
        sys.exit(6)
    except:
        traceback.print_exc()
        print("connect pg error " + pgconfig["database"])
        sys.exit(6)

上述代碼中的postgresql_connect函數(shù)：

遍歷pgconfigs列表，查找與參數(shù)servicename匹配的配置。
使用匹配的配置建立PostgreSQL連接。
如果未找到匹配的服務(wù)名，打印錯(cuò)誤信息并退出程序。

帶編碼參數(shù)的連接函數(shù)

def postgresql_connect_encode(servicename, encoding='UTF-8'):
    global job_task_content_ori
    global job_resource_conn_info
    global job_hdfs_dir_info
    try:
        for subconfig in pgconfigs:
            if servicename in subconfig:
                pgconfig = subconfig[servicename]
                conn = psycopg2.connect(database=pgconfig["database"], user=pgconfig["user"],
                                        password=pgconfig["password"], host=pgconfig["host"], port=pgconfig["port"],
                                        client_encoding=encoding)
                return conn
        print("servicename %s not found in config dictionary" % servicename)
        sys.exit(6)
    except:
        traceback.print_exc()
        print("connect pg error " + pgconfig["database"])
        sys.exit(6)

postgresql_connect_encode函數(shù)與前一個(gè)函數(shù)類似，唯一的區(qū)別是它接受一個(gè)可選的編碼參數(shù)encoding，并將其傳遞給psycopg2.connect以設(shè)定客戶端編碼。

4.解釋關(guān)鍵點(diǎn)

配置管理：pgconfigs列表包含多個(gè)數(shù)據(jù)庫配置，便于根據(jù)服務(wù)名快速查找匹配的配置項(xiàng)。

錯(cuò)誤處理：在連接失敗時(shí)，打印堆棧跟蹤以幫助調(diào)試并退出程序。

編碼管理：對(duì)于需要特定編碼的連接，提供了額外的函數(shù)來設(shè)置編碼。

5.使用示例

假設(shè)你要連接到名為 dataos_71_pg_dev 的PostgreSQL服務(wù)，可以像下面這樣使用這些函數(shù)：

# 普通連接
conn = postgresql_connect("dataos_71_pg_dev")

# 帶編碼參數(shù)的連接
conn_with_encoding = postgresql_connect_encode("dataos_71_pg_dev", encoding='UTF-8')

完整python

#!/data/apps/python2715/bin/python
# -*- coding:utf-8 -*-

from pyhive import hive
import cx_Oracle
import pymysql
import pymssql
from optparse import OptionParser
import logging
import sys
import traceback
import re
import warnings
with warnings.catch_warnings(record=True):
    import psycopg2
reload(sys)

pgconfigs = [
    {"asset_factory": {"database": "asset_factory", "user": "asset", "password": "Ro20ot16", "host": "10.251.80.202", "port": "65432"}},
    {"asset_register": {"database": "asset_register", "user": "asset", "password": "Ro20ot16", "host": "10.251.80.202", "port": "65432"}},
    {"creditdb": {"database": "creditdb", "user": "credit", "password": "hHJ98#pE40Y", "host": "10.251.101.175", "port": "18923"}},
    {"asset_portal": {"database": "asset_portal", "user": "asset", "password": "Ro20ot16", "host": "10.251.80.202", "port": "65432"}},
    {"tyjh": {"database": "tyjh", "user": "tyjh", "password": "k5y2dwoKcFm&^OsW", "host": "10.251.90.35", "port": "18922"}},
    {"cwrl": {"database": "cwrl", "user": "cwjh_yaxin", "password": "eKYy4R3&MbaaY3Zy", "host": "10.251.90.34", "port": "18921"}},
    {"dc_new_pg": {"database": "dc", "user": "dc", "password": "Ojjkcy@jVxKIeo5C", "host": "10.251.90.36", "port": "18923"}},
    {"credit_data": {"database": "credit_data", "user": "credit_data", "password": "@G*1f*3$3DzTH6%o", "host": "10.251.90.134", "port": "18923"}},
    {"new_jf_pg_cd": {"database": "accthuiju", "user": "datacenter", "password": "KridU593&%rj90", "host": "133.37.116.192", "port": "18921"}},
    {"new_accthuiju": {"database": "accthuiju", "user": "itf_ods", "password": "2dH~fZ^8", "host": "10.251.64.226", "port": "18921"}},
    {"crm3huijupg": {"database": "crm3huijupg", "user": "datacenter", "password": "lp*36^YD", "host": "10.251.65.93", "port": "18921"}},
    {"crm3hispg": {"database": "crm3hispg", "user": "datacenter", "password": "~4nD_jWG", "host": "10.251.64.196", "port": "18921"}},
    {"hana": {"database": "hana", "user": "hana", "password": "gvptHKXVNKEGw7pk", "host": "10.251.90.94", "port": "18924"}},
    {"dataos_71_pg_dev": {"database": "dacp", "user": "dacp", "password": "jxFgCKv9GJw2ohS3", "host": "10.251.110.104", "port": "18921"}}
]


def postgresql_connect(servicename):
    global job_task_content_ori
    global job_resource_conn_info
    global job_hdfs_dir_info
    try:
        for subconfig in pgconfigs:
           if subconfig.has_key(servicename):
                pgconfig = subconfig[servicename]
                conn = psycopg2.connect(database=pgconfig["database"], user=pgconfig["user"],
                                        password=pgconfig["password"], host=pgconfig["host"], port=pgconfig["port"])
                return conn
        print("servinamename %s not found in config dictionary" % servicename)
        sys.exit(6)
    except:
        traceback.print_exc()
        print("connect pg error " + pgconfig["database"])
        sys.exit(6)


def postgresql_connect_encode(servicename, encoding='UTF-8'):
    global job_task_content_ori
    global job_resource_conn_info
    global job_hdfs_dir_info
    try:
        for subconfig in pgconfigs:
            if subconfig.has_key(servicename):
                pgconfig = subconfig[servicename]
                conn = psycopg2.connect(database=pgconfig["database"], user=pgconfig["user"],
                                        password=pgconfig["password"], host=pgconfig["host"], port=pgconfig["port"],
                                        client_encoding=encoding)
                return conn
        print("servinamename %s not found in config dictionary" % servicename)
        sys.exit(6)
    except:
        traceback.print_exc()
        print("connect pg error " + pgconfig["database"])
        sys.exit(6)