Add labels support to LoadTableResponse and Table

pyiceberg/catalog/rest/__init__.py

@@ -278,6 +278,7 @@ class TableResponse(IcebergBaseModel):
     metadata: TableMetadata
     config: Properties = Field(default_factory=dict)
     storage_credentials: list[StorageCredential] = Field(alias="storage-credentials", default_factory=list)
+    labels: dict[str, Any] = Field(default_factory=dict)
 
 
 class ViewResponse(IcebergBaseModel):
@@ -805,6 +806,7 @@ def _response_to_table(self, identifier_tuple: tuple[str, ...], table_response:
             ),
             catalog=self,
             config=table_response.config,
+            labels=table_response.labels,
         )
 
     def _response_to_staged_table(self, identifier_tuple: tuple[str, ...], table_response: TableResponse) -> StagedTable:

pyiceberg/table/__init__.py

@@ -1049,6 +1049,7 @@ class Table:
     io: FileIO
     catalog: Catalog
     config: dict[str, str]
+    labels: dict[str, Any]
 
     def __init__(
         self,
@@ -1058,13 +1059,32 @@ def __init__(
         io: FileIO,
         catalog: Catalog,
         config: dict[str, str] = EMPTY_DICT,
+        labels: dict[str, Any] | None = None,
     ) -> None:
         self._identifier = identifier
         self.metadata = metadata
         self.metadata_location = metadata_location
         self.io = io
         self.catalog = catalog
         self.config = config
+        self.labels = labels or {}
+
+    @property
+    def table_labels(self) -> dict[str, str]:
+        """Table-level labels from catalog enrichment."""
+        return self.labels.get("table", {})
+
+    @property
+    def column_labels(self) -> list[dict[str, Any]]:
+        """Column-level labels from catalog enrichment."""
+        return self.labels.get("columns", [])
+
+    def get_column_label(self, field_id: int) -> dict[str, str]:
+        """Get labels for a specific column by field-id."""
+        for col in self.column_labels:
+            if col.get("field-id") == field_id:
+                return col.get("labels", {})
+        return {}
 
     def transaction(self) -> Transaction:
         """Create a new transaction object to first stage the changes, and then commit them to the catalog.