feat(storage): add pyiceberg wrapper (#1326)

Author: o-santi

src/storage/pyproject.toml

@@ -26,6 +26,7 @@ dependencies = [
   "deprecation >=2.1.0",
   "pydantic >=2.11.7",
   "yarl>=1.20.1",
+  "pyiceberg>=0.10.0",
 ]
 
 [project.urls]

src/storage/src/storage3/_async/analytics.py

@@ -1,6 +1,7 @@
 from typing import List, Optional
 
 from httpx import QueryParams
+from pyiceberg.catalog.rest import RestCatalog
 
 from ..types import (
     AnalyticsBucket,
@@ -49,3 +50,24 @@ async def delete(self, bucket_name: str) -> AnalyticsBucketDeleteResponse:
             http_method="DELETE", path=["bucket", bucket_name]
         )
         return AnalyticsBucketDeleteResponse.model_validate_json(data.content)
+
+    def catalog(
+        self, catalog_name: str, access_key_id: str, secret_access_key: str
+    ) -> RestCatalog:
+        catalog_uri = self._request._base_url
+        s3_endpoint = self._request._base_url.parent.joinpath("s3")
+        service_key = self._request.headers.get("apiKey")
+        assert service_key, "apiKey must be passed in the headers."
+        return RestCatalog(
+            catalog_name,
+            warehouse=catalog_name,
+            uri=str(catalog_uri),
+            token=service_key,
+            **{
+                "py-io-impl": "pyiceberg.io.pyarrow.PyArrowFileIO",
+                "s3.endpoint": str(s3_endpoint),
+                "s3.access-key-id": access_key_id,
+                "s3.secret-access-key": secret_access_key,
+                "s3.force-virtual-addressing": "False",
+            },
+        )

src/storage/src/storage3/_sync/analytics.py

@@ -1,6 +1,7 @@
 from typing import List, Optional
 
 from httpx import QueryParams
+from pyiceberg.catalog.rest import RestCatalog
 
 from ..types import (
     AnalyticsBucket,
@@ -47,3 +48,24 @@ def list(
     def delete(self, bucket_name: str) -> AnalyticsBucketDeleteResponse:
         data = self._request.send(http_method="DELETE", path=["bucket", bucket_name])
         return AnalyticsBucketDeleteResponse.model_validate_json(data.content)
+
+    def catalog(
+        self, catalog_name: str, access_key_id: str, secret_access_key: str
+    ) -> RestCatalog:
+        catalog_uri = self._request._base_url
+        s3_endpoint = self._request._base_url.parent.joinpath("s3")
+        service_key = self._request.headers.get("apiKey")
+        assert service_key, "apiKey must be passed in the headers."
+        return RestCatalog(
+            catalog_name,
+            warehouse=catalog_name,
+            uri=str(catalog_uri),
+            token=service_key,
+            **{
+                "py-io-impl": "pyiceberg.io.pyarrow.PyArrowFileIO",
+                "s3.endpoint": str(s3_endpoint),
+                "s3.access-key-id": access_key_id,
+                "s3.secret-access-key": secret_access_key,
+                "s3.force-virtual-addressing": "False",
+            },
+        )