如何在函数中使用.get()方法仅读取字典中存在的项,解决未配置表的报错问题
如何在函数中使用.get()方法仅读取字典中存在的项,解决未配置表的报错问题
看起来你的问题根源是:当尝试访问配置里未定义的表时,代码在self.config["...UNIQUE_TABLES"][self.table]这一行触发了KeyError,但原来的try-except块根本没包裹这段代码,所以异常捕获完全没起作用。用字典的.get()方法确实是更稳妥的方案,能直接跳过不存在的表,避免报错。
我们来修改你的代码,核心就是用.get()替代直接的键访问,让代码只处理配置里存在的表:
def DBCount(self, db_count_dict): schema = self.config["data-validation"]["database"]["schema"] # 用get()获取当前表的唯一列配置,不存在则返回None unique_cols_str = self.config["data-validation"]["RULES"]["RULE_UNIQUE_VALIDATION"]["UNIQUE_TABLES"].get(self.table) # 如果当前表没有配置唯一列,直接跳过处理 if not unique_cols_str: db_count_dict[self.table] = {} return db_count_dict unique_cols = unique_cols_str.split(",") db_count_dict[self.table] = {} for col in unique_cols: try: cnxn = SourceValidation.getDBConnection(self.env) cursor = cnxn.cursor() print(f"UNIQUE COL: {col}") unique_count_query = f"SELECT COUNT(DISTINCT {col}) as unique_count FROM {schema}.{self.table}" df = pd.read_sql(unique_count_query, cnxn) db_count_dict[self.table][col] = str(df['unique_count'].values[0]) except Exception as e: # 这里可以根据需要捕获具体异常,比如数据库连接、SQL语法错误等 print(f"处理列 {col} 时出错: {str(e)}") continue finally: # 记得关闭数据库连接,避免资源泄漏 if cnxn: cnxn.close() return db_count_dict
关键修改点说明:
- 用
.get()安全获取配置:UNIQUE_TABLES.get(self.table)会在表不存在时返回None,而不是直接抛出KeyError,这就从根源避免了报错。 - 提前判断跳过无配置的表:如果
unique_cols_str为空或None,直接给当前表设空字典并返回,不用进入后续逻辑。 - 优化循环和异常处理:把
range(len(unique_cols))改成直接遍历列名,代码更简洁;同时在finally块里关闭数据库连接,避免资源泄漏;还可以捕获更具体的异常并打印日志,方便排查问题。 - 用f-string替代字符串拼接:比原来的
str()拼接更清晰易读,也减少语法错误的可能。
这样修改后,你的函数就只会处理配置文件里存在的表,遇到未配置的表时会自动跳过,不会再报错啦。
备注:内容来源于stack exchange,提问作者noride_togo
相关产品推荐
相关产品推荐

