如何通过 Python 中的 YAML 配置文件检查与数据库的连接?
How to check connection to database via YAML configuration file in Python?
我的项目需要一个 YAML 配置文件来检查与数据库的连接,但我遇到了一些问题:如果特定于主机的是 none,它将自动成为默认值 (localhost)。
我已经用样本文件
测试过了
但好像我的代码有问题,主机为空时不显示默认值作为loacalhost,否则显示none。谁能给我建议?我哪里错了?
我的样本文件data_source.yml:
database:
dbopt:
host:
port: 5432
dbname: db1
user: username
password: 1234
client_encoding: utf-8
connect_timeout: 60
sslmode: none
query:
select * from manufacturing_product
query:
select * from manufacturing_product
我的代码:
import yaml
class InvalidConfigError(Exception):
pass
class DB:
def __init__(self, dbconf):
self._dbconf = dict(dbconf)
# checking for database type
dbtype = self.get_db_type()
if dbtype != 'sqlite' and dbtype != 'postgres':
raise InvalidConfigError(
'E01001', 'Invalid database type, should be sqlite or postgres.')
else:
self.dbtype = dbtype
#checking db option
dbopt = self.__get_dbopt()
if dbopt is None:
raise InvalidConfigError(
'E01002', 'Invalid database options.')
else:
self.dbopt = dbopt
#checking host
host = dbopt.get('host')
if host is None or len(host) <= 0:
self.host = 'localhost'
else:
self.host = host
#checking dbname
dbname = dbopt.get('dbname')
if dbname is None or len(dbname) <= 0:
raise Exception("Database name is required.")
else:
self.dbname = dbname
def get_db_type(self):
return self._dbconf['db']
def __get_dbopt(self):
return self._dbconf.get('dbopt')
def get_db_host(self):
return self.host
def get_db_name(self):
return self.dbname
with open('data_source.yml') as file:
data = yaml.full_load(file)
for item, doc in data.items():
print(item, ":", doc)
db = DB(data)
输出:
database:
dbopt:
host:
port: 5432
dbname: db1
user: username
password: 1234
client_encoding: utf-8
connect_timeout: 60
sslmode: none
query:
select * from manufacturing_product
你的代码很好,只是你没有正确检索它,我检查并得到主机值 'localhost'
with open('data_source.yml') as file:
data = yaml.full_load(file)
for item, doc in data.items():
print(item, ":", doc)
db = DB(data)
print("default host value:", db.get_db_host()) # added to test default host value
输出:
db : postgres
dbopt : {'host': 'None', 'port': 1234, 'dbname': 'xyz', 'user': 'abc', 'password': 'pass', 'client_encoding': 'utf-8', 'connect_timeout': 100, 'sslmode': 'none'}
insopt : {'table': 'tries', 'out': 'qubna', 'bl': 'tghqua'}
我的项目需要一个 YAML 配置文件来检查与数据库的连接,但我遇到了一些问题:如果特定于主机的是 none,它将自动成为默认值 (localhost)。 我已经用样本文件
测试过了但好像我的代码有问题,主机为空时不显示默认值作为loacalhost,否则显示none。谁能给我建议?我哪里错了?
我的样本文件data_source.yml:
database:
dbopt:
host:
port: 5432
dbname: db1
user: username
password: 1234
client_encoding: utf-8
connect_timeout: 60
sslmode: none
query:
select * from manufacturing_product
query:
select * from manufacturing_product
我的代码:
import yaml
class InvalidConfigError(Exception):
pass
class DB:
def __init__(self, dbconf):
self._dbconf = dict(dbconf)
# checking for database type
dbtype = self.get_db_type()
if dbtype != 'sqlite' and dbtype != 'postgres':
raise InvalidConfigError(
'E01001', 'Invalid database type, should be sqlite or postgres.')
else:
self.dbtype = dbtype
#checking db option
dbopt = self.__get_dbopt()
if dbopt is None:
raise InvalidConfigError(
'E01002', 'Invalid database options.')
else:
self.dbopt = dbopt
#checking host
host = dbopt.get('host')
if host is None or len(host) <= 0:
self.host = 'localhost'
else:
self.host = host
#checking dbname
dbname = dbopt.get('dbname')
if dbname is None or len(dbname) <= 0:
raise Exception("Database name is required.")
else:
self.dbname = dbname
def get_db_type(self):
return self._dbconf['db']
def __get_dbopt(self):
return self._dbconf.get('dbopt')
def get_db_host(self):
return self.host
def get_db_name(self):
return self.dbname
with open('data_source.yml') as file:
data = yaml.full_load(file)
for item, doc in data.items():
print(item, ":", doc)
db = DB(data)
输出:
database:
dbopt:
host:
port: 5432
dbname: db1
user: username
password: 1234
client_encoding: utf-8
connect_timeout: 60
sslmode: none
query:
select * from manufacturing_product
你的代码很好,只是你没有正确检索它,我检查并得到主机值 'localhost'
with open('data_source.yml') as file:
data = yaml.full_load(file)
for item, doc in data.items():
print(item, ":", doc)
db = DB(data)
print("default host value:", db.get_db_host()) # added to test default host value
输出:
db : postgres
dbopt : {'host': 'None', 'port': 1234, 'dbname': 'xyz', 'user': 'abc', 'password': 'pass', 'client_encoding': 'utf-8', 'connect_timeout': 100, 'sslmode': 'none'}
insopt : {'table': 'tries', 'out': 'qubna', 'bl': 'tghqua'}