in src/dma/collector/workflows/readiness_check/_postgres/main.py [0:0]
def _save_results(self, db_variant: PostgresVariants, db_check_results: dict[str, dict[str, list]]) -> None:
for rule, result in db_check_results.items():
for severity in [ACTION_REQUIRED, WARNING, PASS]:
output_str = ""
if rule == PGLOGICAL_INSTALLED:
if severity == ACTION_REQUIRED:
output_str = f"`pglogical` extension is not installed on the databases: {','.join(result[ACTION_REQUIRED])}."
elif severity == PASS:
output_str = f"`pglogical` is installed on the databases: {','.join(result[PASS])}."
else:
continue
if rule == PRIVILEGES:
if severity == ACTION_REQUIRED:
output_str = ";\n\n".join(result[ACTION_REQUIRED])
elif severity == PASS:
output_str = ";\n".join(result[PASS])
else:
continue
if rule == PGLOGICAL_NODE_ALREADY_EXISTS:
if severity == ACTION_REQUIRED:
output_str = (
f"pglogical provider node already exists on databases: {','.join(result[ACTION_REQUIRED])}"
)
elif severity == PASS:
output_str = f"No existing pglogical provider node on the database: {','.join(result[PASS])}"
else:
continue
if rule == TABLES_WITH_NO_PK:
if severity == WARNING:
output_str = f"Some tables have limited support. Tables without primary keys were identified and only INSERT statements will be replicated for these tables: {';'.join(result[WARNING])}"
else:
continue
if rule == UNSUPPORTED_TABLES_WITH_REPLICA_IDENTITY:
if severity == ACTION_REQUIRED:
output_str = f"Source has table(s) with both primary key and replica identity FULL or NOTHING. Please remove replica identity or change it to DEFAULT to migrate: {';'.join(result[ACTION_REQUIRED])}"
else:
continue
if len(result[severity]) > 0:
self.save_rule_result(
db_variant,
rule,
severity, # type: ignore
output_str,
)