def _check_pglogical_privileges()

in src/dma/collector/workflows/readiness_check/_postgres/main.py [0:0]

• 21 lines of code
• 7 McCabe index (conditional complexity)

    def _check_pglogical_privileges(self, db_name: str) -> list[str]:
        err = self._check_pglogical_schema_usage_privilege(db_name)
        if err is not None:
            return [err]
        result = self.local_db.sql(
            "select has_tables_select_privilege, has_local_node_select_privilege, has_node_select_privilege, has_node_interface_select_privilege from collection_postgres_pglogical_privileges where database_name = $db_name",
            params={"db_name": db_name},
        ).fetchone()
        errors: list[str] = []
        if result is None:
            errors.append("Empty result reading pglogical privileges for the user")
        else:
            if not result[0]:
                errors.append("user doesn't have SELECT privilege on table pglogical.tables")
            if not result[1]:
                errors.append("user doesn't have SELECT privilege on table pglogical.local_node")
            if not result[2]:
                errors.append("user doesn't have SELECT privilege on table pglogical.node")
            if not result[3]:
                errors.append("user doesn't have SELECT privilege on table pglogical.node_interface")
        return errors