From ea1b2bb29eb7f2defac02beb5ba8c597a161bc56 Mon Sep 17 00:00:00 2001 From: Matteo Morlacchi Date: Wed, 11 Mar 2026 01:21:25 +0100 Subject: [PATCH 1/3] feat: fix async method using sync interactions with local fs --- obstore/python/obstore/fsspec.py | 34 ++++++++++++++++++-------------- 1 file changed, 19 insertions(+), 15 deletions(-) diff --git a/obstore/python/obstore/fsspec.py b/obstore/python/obstore/fsspec.py index fa0d4c0e..71fc1f13 100644 --- a/obstore/python/obstore/fsspec.py +++ b/obstore/python/obstore/fsspec.py @@ -28,7 +28,7 @@ integration. """ -# ruff: noqa: ANN401, EM102, PTH123, FBT001, FBT002 +# ruff: noqa: ANN401, EM102, FBT001, FBT002 from __future__ import annotations @@ -43,8 +43,8 @@ import fsspec.asyn import fsspec.spec -from obstore import open_reader, open_writer -from obstore.store import from_url +from obstore import open_reader, open_writer, open_writer_async +from obstore.store import LocalStore, from_url if TYPE_CHECKING: import sys @@ -260,6 +260,7 @@ def __init__( # noqa: PLR0913 self._retry_config = retry_config self._config_kwargs = kwargs self._credential_provider = credential_provider + self._local_store = LocalStore() # https://stackoverflow.com/a/68550238 self._construct_store = lru_cache(maxsize=max_cache_size)(self._construct_store) @@ -462,39 +463,42 @@ async def _put_file( mode: str = "overwrite", # noqa: ARG002 **_kwargs: Any, ) -> None: - # TODO: We shouldn't use blocking file operations in async functions - if not Path(lpath).is_file(): # noqa: ASYNC240 - err_msg = f"File {lpath} not found in local" - raise FileNotFoundError(err_msg) + # Raises FileNotFoundError if is directory or file not found + _ = await self._local_store.head_async(lpath) - # TODO: convert to use async file system methods using LocalStore # Async functions should not open files with blocking methods like `open` rbucket, rpath = self._split_path(rpath) # Should construct the store instance by rbucket, which is the target path store = self._construct_store(rbucket) - with open(lpath, "rb") as f: # noqa: ASYNC230 - await store.put_async(rpath, f) + await store.put_async(rpath, Path(lpath)) async def _get_file(self, rpath: str, lpath: str, **_kwargs: Any) -> None: res = urlparse(lpath) - # TODO: We shouldn't use blocking file operations in async functions - if res.scheme or Path(lpath).is_dir(): # noqa: ASYNC240 + if ( + res.scheme + or len( + (await self._local_store.list_with_delimiter_async(lpath)).get( + "common_prefixes", + [], + ), + ) + > 0 + ): # lpath need to be local file and cannot contain scheme return - # TODO: convert to use async file system methods using LocalStore # Async functions should not open files with blocking methods like `open` rbucket, rpath = self._split_path(rpath) # Should construct the store instance by rbucket, which is the target path store = self._construct_store(rbucket) - with open(lpath, "wb") as f: # noqa: ASYNC230 + async with open_writer_async(self._local_store, lpath) as f: resp = await store.get_async(rpath) async for buffer in resp.stream(): - f.write(buffer) + await f.write(buffer) async def _info(self, path: str, **_kwargs: Any) -> dict[str, Any]: bucket, path_no_bucket = self._split_path(path) From 2393d5410903204e95496a91bf9a9d91d220e0c3 Mon Sep 17 00:00:00 2001 From: Matteo Morlacchi Date: Wed, 25 Mar 2026 23:52:28 +0100 Subject: [PATCH 2/3] feat: fix async method using sync interactions with local fs --- obstore/python/obstore/fsspec.py | 39 +++++++++++--------------------- 1 file changed, 13 insertions(+), 26 deletions(-) diff --git a/obstore/python/obstore/fsspec.py b/obstore/python/obstore/fsspec.py index 71fc1f13..e3404a54 100644 --- a/obstore/python/obstore/fsspec.py +++ b/obstore/python/obstore/fsspec.py @@ -35,7 +35,7 @@ import asyncio import warnings from collections import defaultdict -from functools import lru_cache +from functools import cached_property, lru_cache from pathlib import Path from typing import TYPE_CHECKING, Literal, overload from urllib.parse import urlparse @@ -43,8 +43,8 @@ import fsspec.asyn import fsspec.spec -from obstore import open_reader, open_writer, open_writer_async -from obstore.store import LocalStore, from_url +from obstore import open_reader, open_writer +from obstore.store import from_url if TYPE_CHECKING: import sys @@ -260,7 +260,6 @@ def __init__( # noqa: PLR0913 self._retry_config = retry_config self._config_kwargs = kwargs self._credential_provider = credential_provider - self._local_store = LocalStore() # https://stackoverflow.com/a/68550238 self._construct_store = lru_cache(maxsize=max_cache_size)(self._construct_store) @@ -272,6 +271,10 @@ def __init__( # noqa: PLR0913 batch_size=batch_size, ) + @cached_property + def _local_store(self) -> FsspecStore: + return FsspecStore("file") + def _split_path(self, path: str) -> tuple[str, str]: """Split bucket and file path. @@ -463,8 +466,9 @@ async def _put_file( mode: str = "overwrite", # noqa: ARG002 **_kwargs: Any, ) -> None: - # Raises FileNotFoundError if is directory or file not found - _ = await self._local_store.head_async(lpath) + + if not await self._local_store._exists(lpath): # noqa: SLF001 + raise FileNotFoundError(lpath) # Async functions should not open files with blocking methods like `open` rbucket, rpath = self._split_path(rpath) @@ -476,29 +480,12 @@ async def _put_file( async def _get_file(self, rpath: str, lpath: str, **_kwargs: Any) -> None: res = urlparse(lpath) - if ( - res.scheme - or len( - (await self._local_store.list_with_delimiter_async(lpath)).get( - "common_prefixes", - [], - ), - ) - > 0 - ): + if res.scheme or self._local_store.isdir(lpath): # lpath need to be local file and cannot contain scheme return - # Async functions should not open files with blocking methods like `open` - rbucket, rpath = self._split_path(rpath) - - # Should construct the store instance by rbucket, which is the target path - store = self._construct_store(rbucket) - - async with open_writer_async(self._local_store, lpath) as f: - resp = await store.get_async(rpath) - async for buffer in resp.stream(): - await f.write(buffer) + resp = await self._cat_file(rpath) + await self._local_store._pipe_file(lpath, resp) # noqa: SLF001 async def _info(self, path: str, **_kwargs: Any) -> dict[str, Any]: bucket, path_no_bucket = self._split_path(path) From 497b48ba9a03878b4de88be3d9e9ab13ec0ad4d4 Mon Sep 17 00:00:00 2001 From: Matteo Morlacchi Date: Thu, 26 Mar 2026 00:04:00 +0100 Subject: [PATCH 3/3] feat: fix async method using sync interactions with local fs --- obstore/python/obstore/fsspec.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/obstore/python/obstore/fsspec.py b/obstore/python/obstore/fsspec.py index e3404a54..6174da60 100644 --- a/obstore/python/obstore/fsspec.py +++ b/obstore/python/obstore/fsspec.py @@ -470,7 +470,6 @@ async def _put_file( if not await self._local_store._exists(lpath): # noqa: SLF001 raise FileNotFoundError(lpath) - # Async functions should not open files with blocking methods like `open` rbucket, rpath = self._split_path(rpath) # Should construct the store instance by rbucket, which is the target path @@ -480,7 +479,7 @@ async def _put_file( async def _get_file(self, rpath: str, lpath: str, **_kwargs: Any) -> None: res = urlparse(lpath) - if res.scheme or self._local_store.isdir(lpath): + if res.scheme or await self._local_store._isdir(lpath): # noqa: SLF001 # lpath need to be local file and cannot contain scheme return