From f4c1236bfbc8bf03fbaea5e0ea7a28374500599f Mon Sep 17 00:00:00 2001
From: Anton-Tarazi <antontarazi@gmail.com>
Date: Fri, 13 Jun 2025 00:16:03 -0400
Subject: [PATCH] feat: support listing known catalogs

---
 pyiceberg/catalog/__init__.py | 4 ++++
 pyiceberg/utils/config.py     | 6 ++++++
 tests/utils/test_config.py    | 9 +++++++++
 3 files changed, 19 insertions(+)

diff --git a/pyiceberg/catalog/__init__.py b/pyiceberg/catalog/__init__.py
index cf649ba7d6..d7d4c9d050 100644
--- a/pyiceberg/catalog/__init__.py
+++ b/pyiceberg/catalog/__init__.py
@@ -262,6 +262,10 @@ def load_catalog(name: Optional[str] = None, **properties: Optional[str]) -> Cat
     raise ValueError(f"Could not initialize catalog with the following properties: {properties}")
 
 
+def list_catalogs() -> List[str]:
+    return _ENV_CONFIG.get_known_catalogs()
+
+
 def delete_files(io: FileIO, files_to_delete: Set[str], file_type: str) -> None:
     """Delete files.
 
diff --git a/pyiceberg/utils/config.py b/pyiceberg/utils/config.py
index 0bfaefdbc6..78f121a402 100644
--- a/pyiceberg/utils/config.py
+++ b/pyiceberg/utils/config.py
@@ -159,6 +159,12 @@ def get_catalog_config(self, catalog_name: str) -> Optional[RecursiveDict]:
                 return catalog_conf
         return None
 
+    def get_known_catalogs(self) -> List[str]:
+        catalogs = self.config.get(CATALOG, {})
+        if not isinstance(catalogs, dict):
+            raise ValueError("Catalog configurations needs to be an object")
+        return list(catalogs.keys())
+
     def get_int(self, key: str) -> Optional[int]:
         if (val := self.config.get(key)) is not None:
             try:
diff --git a/tests/utils/test_config.py b/tests/utils/test_config.py
index 89247d8fca..24a867f812 100644
--- a/tests/utils/test_config.py
+++ b/tests/utils/test_config.py
@@ -56,6 +56,15 @@ def test_fix_nested_objects_from_environment_variables() -> None:
     }
 
 
+@mock.patch.dict(os.environ, EXAMPLE_ENV)
+@mock.patch.dict(os.environ, {"PYICEBERG_CATALOG__DEVELOPMENT__URI": "https://dev.service.io/api"})
+def test_list_all_known_catalogs() -> None:
+    assert Config().get_known_catalogs() == [
+        "production",
+        "development",
+    ]
+
+
 def test_from_configuration_files(tmp_path_factory: pytest.TempPathFactory) -> None:
     config_path = str(tmp_path_factory.mktemp("config"))
     with open(f"{config_path}/.pyiceberg.yaml", "w", encoding=UTF8) as file: