Bring back async methods for python (#1074)

* Add first python async function

* Moreeeee

* clippy

* All sessions meethods now have async versions

* lookup_tags

* Add missing tags async impls

* All branches methods now have async

* lint

* Add writable session async method

* Add rewrite manifests async

* Add expire_snapshots

* Add garbage_collection async method

* Fix bad merge

* Update repository.rs

* Wrok on splitting up resolve version

* One more

* Resolve diff correctly

* More async methods

* remaining store methods

* Add python bindigns

* add a boatload of async tests to timetravel

* More tests

* Mix in async method with gc test

* Sync with upstream virtual refs changes

* Remove unnecessary async methods

* cleanup

* More cleanup

* Cleanup as of spec versioning

* gc test cleanup prep

* Virtual refs tests

* gc async tests

* More tests

* Remove unneccsssary async chuink coordinates function

* More tests

* Fix tests

* Think thats all

* Finish up virtual tests to simplify

* More tests

* More tests

Author: mpiannucci

icechunk-python/python/icechunk/_icechunk_python.pyi

@@ -1281,6 +1281,14 @@ class PyRepository:
         authorize_virtual_chunk_access: dict[str, AnyCredential | None] | None = None,
     ) -> PyRepository: ...
     @classmethod
+    async def create_async(
+        cls,
+        storage: Storage,
+        *,
+        config: RepositoryConfig | None = None,
+        authorize_virtual_chunk_access: dict[str, AnyCredential | None] | None = None,
+    ) -> PyRepository: ...
+    @classmethod
     def open(
         cls,
         storage: Storage,
@@ -1289,23 +1297,57 @@ class PyRepository:
         authorize_virtual_chunk_access: dict[str, AnyCredential | None] | None = None,
     ) -> PyRepository: ...
     @classmethod
+    async def open_async(
+        cls,
+        storage: Storage,
+        *,
+        config: RepositoryConfig | None = None,
+        authorize_virtual_chunk_access: dict[str, AnyCredential | None] | None = None,
+    ) -> PyRepository: ...
+    @classmethod
     def open_or_create(
         cls,
         storage: Storage,
         *,
         config: RepositoryConfig | None = None,
         authorize_virtual_chunk_access: dict[str, AnyCredential | None] | None = None,
     ) -> PyRepository: ...
+    @classmethod
+    async def open_or_create_async(
+        cls,
+        storage: Storage,
+        *,
+        config: RepositoryConfig | None = None,
+        authorize_virtual_chunk_access: dict[str, AnyCredential | None] | None = None,
+    ) -> PyRepository: ...
     @staticmethod
     def exists(storage: Storage) -> bool: ...
+    @staticmethod
+    async def exists_async(storage: Storage) -> bool: ...
     @classmethod
     def from_bytes(cls, data: bytes) -> PyRepository: ...
     def as_bytes(self) -> bytes: ...
     @staticmethod
     def fetch_config(storage: Storage) -> RepositoryConfig | None: ...
+    @staticmethod
+    async def fetch_config_async(storage: Storage) -> RepositoryConfig | None: ...
     def save_config(self) -> None: ...
+    async def save_config_async(self) -> None: ...
     def config(self) -> RepositoryConfig: ...
+    def storage_settings(self) -> StorageSettings: ...
     def storage(self) -> Storage: ...
+    def reopen(
+        self,
+        *,
+        config: RepositoryConfig | None = None,
+        authorize_virtual_chunk_access: dict[str, AnyCredential | None] | None = None,
+    ) -> PyRepository: ...
+    async def reopen_async(
+        self,
+        *,
+        config: RepositoryConfig | None = None,
+        authorize_virtual_chunk_access: dict[str, AnyCredential | None] | None = None,
+    ) -> PyRepository: ...
     def set_default_commit_metadata(self, metadata: dict[str, Any]) -> None: ...
     def default_commit_metadata(self) -> dict[str, Any]: ...
     def async_ancestry(
@@ -1316,15 +1358,25 @@ class PyRepository:
         snapshot_id: str | None = None,
     ) -> AsyncIterator[SnapshotInfo]: ...
     def create_branch(self, branch: str, snapshot_id: str) -> None: ...
+    async def create_branch_async(self, branch: str, snapshot_id: str) -> None: ...
     def list_branches(self) -> set[str]: ...
+    async def list_branches_async(self) -> set[str]: ...
     def lookup_branch(self, branch: str) -> str: ...
+    async def lookup_branch_async(self, branch: str) -> str: ...
     def lookup_snapshot(self, snapshot_id: str) -> SnapshotInfo: ...
+    async def lookup_snapshot_async(self, snapshot_id: str) -> SnapshotInfo: ...
     def reset_branch(self, branch: str, snapshot_id: str) -> None: ...
+    async def reset_branch_async(self, branch: str, snapshot_id: str) -> None: ...
     def delete_branch(self, branch: str) -> None: ...
+    async def delete_branch_async(self, branch: str) -> None: ...
     def delete_tag(self, tag: str) -> None: ...
+    async def delete_tag_async(self, tag: str) -> None: ...
     def create_tag(self, tag: str, snapshot_id: str) -> None: ...
+    async def create_tag_async(self, tag: str, snapshot_id: str) -> None: ...
     def list_tags(self) -> set[str]: ...
+    async def list_tags_async(self) -> set[str]: ...
     def lookup_tag(self, tag: str) -> str: ...
+    async def lookup_tag_async(self, tag: str) -> str: ...
     def diff(
         self,
         from_branch: str | None = None,
@@ -1334,6 +1386,15 @@ class PyRepository:
         to_tag: str | None = None,
         to_snapshot_id: str | None = None,
     ) -> Diff: ...
+    async def diff_async(
+        self,
+        from_branch: str | None = None,
+        from_tag: str | None = None,
+        from_snapshot_id: str | None = None,
+        to_branch: str | None = None,
+        to_tag: str | None = None,
+        to_snapshot_id: str | None = None,
+    ) -> Diff: ...
     def readonly_session(
         self,
         branch: str | None = None,
@@ -1342,17 +1403,36 @@ class PyRepository:
         snapshot_id: str | None = None,
         as_of: datetime.datetime | None = None,
     ) -> PySession: ...
+    async def readonly_session_async(
+        self,
+        branch: str | None = None,
+        *,
+        tag: str | None = None,
+        snapshot_id: str | None = None,
+        as_of: datetime.datetime | None = None,
+    ) -> PySession: ...
     def writable_session(self, branch: str) -> PySession: ...
+    async def writable_session_async(self, branch: str) -> PySession: ...
     def expire_snapshots(
         self,
         older_than: datetime.datetime,
         *,
         delete_expired_branches: bool = False,
         delete_expired_tags: bool = False,
     ) -> set[str]: ...
+    async def expire_snapshots_async(
+        self,
+        older_than: datetime.datetime,
+        *,
+        delete_expired_branches: bool = False,
+        delete_expired_tags: bool = False,
+    ) -> set[str]: ...
     def rewrite_manifests(
         self, message: str, *, branch: str, metadata: dict[str, Any] | None = None
     ) -> str: ...
+    async def rewrite_manifests_async(
+        self, message: str, *, branch: str, metadata: dict[str, Any] | None = None
+    ) -> str: ...
     def garbage_collect(
         self,
         delete_object_older_than: datetime.datetime,
@@ -1362,13 +1442,29 @@ class PyRepository:
         max_compressed_manifest_mem_bytes: int = 512 * 1024 * 1024,
         max_concurrent_manifest_fetches: int = 500,
     ) -> GCSummary: ...
+    async def garbage_collect_async(
+        self,
+        delete_object_older_than: datetime.datetime,
+        *,
+        dry_run: bool = False,
+        max_snapshots_in_memory: int = 50,
+        max_compressed_manifest_mem_bytes: int = 512 * 1024 * 1024,
+        max_concurrent_manifest_fetches: int = 500,
+    ) -> GCSummary: ...
     def total_chunks_storage(
         self,
         *,
         max_snapshots_in_memory: int = 50,
         max_compressed_manifest_mem_bytes: int = 512 * 1024 * 1024,
         max_concurrent_manifest_fetches: int = 500,
     ) -> int: ...
+    async def total_chunks_storage_async(
+        self,
+        *,
+        max_snapshots_in_memory: int = 50,
+        max_compressed_manifest_mem_bytes: int = 512 * 1024 * 1024,
+        max_concurrent_manifest_fetches: int = 500,
+    ) -> int: ...
 
 class PySession:
     @classmethod
@@ -1386,20 +1482,30 @@ class PySession:
     def status(self) -> Diff: ...
     def discard_changes(self) -> None: ...
     def all_virtual_chunk_locations(self) -> list[str]: ...
+    async def all_virtual_chunk_locations_async(self) -> list[str]: ...
     def chunk_coordinates(
         self, array_path: str, batch_size: int
     ) -> AsyncIterator[list[list[int]]]: ...
     @property
     def store(self) -> PyStore: ...
     def merge(self, other: PySession) -> None: ...
+    async def merge_async(self, other: PySession) -> None: ...
     def commit(
         self,
         message: str,
         metadata: dict[str, Any] | None = None,
         rebase_with: ConflictSolver | None = None,
         rebase_tries: int = 1_000,
     ) -> str: ...
+    async def commit_async(
+        self,
+        message: str,
+        metadata: dict[str, Any] | None = None,
+        rebase_with: ConflictSolver | None = None,
+        rebase_tries: int = 1_000,
+    ) -> str: ...
     def rebase(self, solver: ConflictSolver) -> None: ...
+    async def rebase_async(self, solver: ConflictSolver) -> None: ...
 
 class PyStore:
     @classmethod
@@ -1437,12 +1543,27 @@ class PyStore:
         checksum: str | datetime.datetime | None,
         validate_container: bool,
     ) -> None: ...
+    async def set_virtual_ref_async(
+        self,
+        key: str,
+        location: str,
+        offset: int,
+        length: int,
+        checksum: str | datetime.datetime | None,
+        validate_container: bool = False,
+    ) -> None: ...
     def set_virtual_refs(
         self,
         array_path: str,
         chunks: list[VirtualChunkSpec],
         validate_containers: bool,
     ) -> list[tuple[int, ...]] | None: ...
+    async def set_virtual_refs_async(
+        self,
+        array_path: str,
+        chunks: list[VirtualChunkSpec],
+        validate_containers: bool,
+    ) -> list[tuple[int, ...]] | None: ...
     async def delete(self, key: str) -> None: ...
     async def delete_dir(self, prefix: str) -> None: ...
     @property