Coverage for /usr/local/lib/python3.12/site-packages/prefect/_internal/compatibility/blocks.py: 22%

1import inspect 1 ctx1a

2from typing import Any, Union 1 ctx1a

3 

4from prefect.filesystems import NullFileSystem, WritableFileSystem 1 ctx1a

5 

6 

7async def call_explicitly_async_block_method( 1 ctx1a

8 block: Union[WritableFileSystem, NullFileSystem], 

9 method: str, 

10 args: tuple[Any, ...], 

11 kwargs: dict[str, Any], 

12) -> Any: 

13 """ 

14 TODO: remove this once we have explicit async methods on all storage blocks 

15 

16 see https://github.com/PrefectHQ/prefect/issues/15008 

17 """ 

18 if hasattr(block, f"a{method}"): # explicit async method 

19 return await getattr(block, f"a{method}")(*args, **kwargs) 

20 elif hasattr(getattr(block, method, None), "aio"): # sync_compatible 

21 return await getattr(block, method).aio(block, *args, **kwargs) 

22 else: # should not happen in prefect, but users can override impls 

23 maybe_coro = getattr(block, method)(*args, **kwargs) 

24 if inspect.isawaitable(maybe_coro): 

25 return await maybe_coro 

26 else: 

27 return maybe_coro