def check_user_obj_privileges()

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

• 23 lines of code
• 5 McCabe index (conditional complexity)

    def check_user_obj_privileges(self, db_name: str) -> list[str]:
        errors: list[str] = []
        rows = self.local_db.sql(
            "select namespace_name from collection_postgres_user_schemas_without_privilege where database_name = $db_name",
            params={"db_name": db_name},
        ).fetchall()
        errors.extend(f"user doesn't have USAGE privilege on schema {row[0]}" for row in rows)

        rows = self.local_db.sql(
            "select schema_name, table_name from collection_postgres_user_tables_without_privilege where database_name = $db_name",
            params={"db_name": db_name},
        ).fetchall()
        errors.extend(f"user doesn't have SELECT privilege on table {row[0]}.{row[1]}" for row in rows)

        rows = self.local_db.sql(
            "select schema_name, view_name from collection_postgres_user_views_without_privilege where database_name = $db_name",
            params={"db_name": db_name},
        ).fetchall()
        errors.extend(f"user doesn't have SELECT privilege on view {row[0]}.{row[1]}" for row in rows)

        rows = self.local_db.sql(
            "select namespace_name, rel_name from collection_postgres_user_sequences_without_privilege where database_name = $db_name",
            params={"db_name": db_name},
        ).fetchall()
        errors.extend(f"user doesn't have SELECT privilege on sequence {row[0]}.{row[1]}" for row in rows)
        return errors