aiomegfile.smart module

class aiomegfile.smart.SmartCacher(path: str, cache_path: str | None = None, mode: str = 'r')

Bases: AioClosable

Async smart cache files in local filesystem.

cache_path = ''

async close() → None

Upload cached content if needed and remove cache file.

property mode: str

Return the cache mode.

property name: str

Return the original path.

async aiomegfile.smart.smart_abspath(path: str | PathLike) → str

Return the absolute path of the given path.

Parameters:

path – Given path.

Returns:

Absolute path string.

Return type:

str

aiomegfile.smart.smart_cache(path: str | ~os.PathLike, cacher: ~typing.Type[~aiomegfile.lib.cacher.SmartCacher] = <class 'aiomegfile.lib.cacher.SmartCacher'>, **options: ~typing.Any) → AsyncContextManager[str]

Return an async cacher for non-local paths.

Examples:

>>> import asyncio
>>> from aiomegfile import smart_cache
>>> async def main():
...     async with smart_cache(
...         "s3://mybucket/myfile.mp4",
...         mode="r",
...     ) as cache_path:
...         print(cache_path)
>>> asyncio.run(main())

Parameters:

• path – Path to cache.

• cacher – Cacher class for non-local paths.

• options – Optional arguments for cacher.

Returns:

Async cacher instance.

Return type:

T.AsyncContextManager[str]

aiomegfile.smart.smart_combine_open(path_glob: str, mode: str = 'rb', open_func=<function smart_open>) → AioCombineReader

Open a unified reader that supports multi-file reading.

Parameters:

• path_glob – Path pattern that may contain shell wildcard characters.

• mode – Mode to open file, supports ‘rb’.

• open_func – Callable to open each file.

Returns:

Combined async reader.

Return type:

AioCombineReader

async aiomegfile.smart.smart_concat(src_paths: List[str | PathLike], dst_path: str | PathLike) → None

Concatenate files in src_paths into a single file at dst_path.

Parameters:

• src_paths – List of source file paths to concatenate.

• dst_path – Destination path for the concatenated file.

async aiomegfile.smart.smart_copy(src_path: str | PathLike, dst_path: str | PathLike, callback: Callable[[int], None] | None = None, followlinks: bool = False, overwrite: bool = True) → str

Copy a file or directory and return the destination path string.

Parameters:

• src_path – Source path to copy.

• dst_path – Destination path.

• callback – Called periodically during copy with bytes written.

• followlinks – Whether to follow symbolic links.

• overwrite – Whether to overwrite destination when it exists.

Returns:

Destination path string.

Return type:

str

async aiomegfile.smart.smart_exists(path: str | PathLike, *, followlinks: bool = False) → bool

Return whether the path points to an existing file or directory.

Parameters:

• path – Path to check.

• followlinks – Whether to follow symbolic links.

Returns:

True if the path exists, otherwise False.

async aiomegfile.smart.smart_getmd5(path: str | PathLike, recalculate: bool = False, followlinks: bool = False) → str

Get the MD5 value of a file or directory.

Parameters:

• path – File path to compute MD5.

• recalculate – Calculate MD5 in real-time or return cached value when supported by the filesystem.

• followlinks – If True, follow symbolic links when calculating MD5.

Returns:

MD5 hex digest.

Return type:

str

async aiomegfile.smart.smart_glob(path: str | PathLike, recursive: bool = True, missing_ok: bool = True) → List[str]

Return paths whose paths match the glob pattern.

Parameters:

• path – Base path to search under.

• recursive – If False, ** will not search directory recursively.

• missing_ok – If False and target path doesn’t match any file, raise FileNotFoundError.

Returns:

List of matching path strings.

Return type:

T.List[str]

Raises:

FileNotFoundError – If no matches and missing_ok is False.

async aiomegfile.smart.smart_glob_stat(pathname: str | PathLike, recursive: bool = True, missing_ok: bool = True, *, sort: bool = False) → AsyncIterator[FileEntry]

Return entries whose paths match the glob pattern.

Parameters:

• pathname – Path pattern that may contain shell wildcard characters.

• recursive – If False, ** will not search directory recursively.

• missing_ok – If False and target path doesn’t match any file, raise FileNotFoundError.

• sort – Whether to request sorted traversal when supported by the filesystem.

Returns:

Async iterator of FileEntry objects.

Return type:

T.AsyncIterator[FileEntry]

Raises:

FileNotFoundError – If no matches and missing_ok is False.

async aiomegfile.smart.smart_iglob(path: str | PathLike, recursive: bool = True, missing_ok: bool = True) → AsyncIterator[str]

Yield paths whose paths match the glob pattern.

Parameters:

• path – Base path to search under.

• recursive – If False, ** will not search directory recursively.

• missing_ok – If False and target path doesn’t match any file, raise FileNotFoundError.

Returns:

Async iterator of matching path strings.

Return type:

T.AsyncIterator[str]

async aiomegfile.smart.smart_isabs(path: str | PathLike) → bool

Test whether a path is absolute.

Parameters:

path – Given path.

Returns:

True if a path is absolute, else False.

Return type:

bool

async aiomegfile.smart.smart_isdir(path: str | PathLike, *, followlinks: bool = False) → bool

Return True if the path points to a directory.

Parameters:

• path – Path to check.

• followlinks – Whether to follow symbolic links.

Returns:

True if the path is a directory, otherwise False.

async aiomegfile.smart.smart_isfile(path: str | PathLike, *, followlinks: bool = False) → bool

Return True if the path points to a regular file.

Parameters:

• path – Path to check.

• followlinks – Whether to follow symbolic links.

Returns:

True if the path is a regular file, otherwise False.

async aiomegfile.smart.smart_islink(path: str | PathLike) → bool

Return True if the path points to a symbolic link.

Parameters:

path – Path to check.

Returns:

True if the path is a symlink, otherwise False.

async aiomegfile.smart.smart_listdir(path: str | PathLike) → List[str]

Return names of entries in the given directory.

Parameters:

path – Directory path to list.

Returns:

List of entry names.

Return type:

T.List[str]

async aiomegfile.smart.smart_load_content(path: str | PathLike, start: int | None = None, stop: int | None = None) → bytes

Get specified file range in bytes.

Parameters:

• path – Specified path.

• start – Start index.

• stop – Stop index (exclusive).

Returns:

Bytes content in range [start, stop).

Return type:

bytes

Raises:

ValueError – If stop is less than start.

async aiomegfile.smart.smart_load_from(path: str | PathLike) → BinaryIO

Read content in binary from the specified path into memory.

Caller is responsible for closing the returned BinaryIO.

Parameters:

path – Specified path to read.

Returns:

BinaryIO containing file content.

Return type:

T.BinaryIO

async aiomegfile.smart.smart_load_text(path: str | PathLike) → str

Read text content from the specified path.

Parameters:

path – Path to read.

Returns:

File content as text.

Return type:

str

async aiomegfile.smart.smart_makedirs(path: str | PathLike, *, mode: int = 511, exist_ok: bool = False) → None

Create a directory and any missing parents.

Parameters:

• path – Directory path to create.

• mode – Permission bits for the new directory.

• exist_ok – Whether to ignore if the directory exists.

Raises:

FileExistsError – When directory exists and exist_ok is False.

async aiomegfile.smart.smart_move(src_path: str | PathLike, dst_path: str | PathLike, overwrite: bool = True) → str

Move a file or directory and return the destination path string.

Parameters:

• src_path – Source path to move.

• dst_path – Destination path.

• overwrite – Whether to overwrite destination when it exists.

Returns:

Destination path string.

Return type:

str

aiomegfile.smart.smart_open(path: str | PathLike, mode: str = 'r', buffering: int = -1, encoding: str | None = None, errors: str | None = None, newline: str | None = None, **kwargs: Any) → AsyncContextManager

Open the file with mode.

Parameters:

• path – File path to open.

• mode – File open mode.

• buffering – Buffering policy.

• encoding – Text encoding in text mode.

• errors – Error handling strategy.

• newline – Newline handling policy in text mode.

• kwargs – Extra open options for compatibility with megfile.

Returns:

Async file context manager.

Return type:

T.AsyncContextManager

async aiomegfile.smart.smart_path_join(path: str | PathLike, *paths: str | PathLike) → str

Join path components and return the combined path string.

Parameters:

• path – Base path.

• paths – Additional path components to join.

Returns:

Combined path string.

Return type:

str

async aiomegfile.smart.smart_readlink(path: str | PathLike) → str

Return the target path string of a symbolic link.

Parameters:

path – Path to the symbolic link.

Returns:

Target path string.

Return type:

str

async aiomegfile.smart.smart_realpath(path: str | PathLike, *, strict: bool = False) → str

Resolve symlinks and return the absolute path string.

Parameters:

• path – Path to resolve.

• strict – Whether to raise if a symlink points to itself.

Returns:

Resolved absolute path string.

Return type:

str

Raises:

OSError – If a symlink points to itself and strict is True.

async aiomegfile.smart.smart_relpath(path: str | PathLike, start: str | PathLike) → str

Compute a relative path from start to path.

Parameters:

• path – Target path.

• start – Base path to compute the relative path against.

Returns:

Relative path string.

Return type:

str

Raises:

ValueError – If path is not under the given start path.

async aiomegfile.smart.smart_remove(path: str | PathLike, missing_ok: bool = False) → None

Remove (delete) the file or directory.

Parameters:

• path – Path to remove.

• missing_ok – If False, raise when the path does not exist.

Raises:

FileNotFoundError – When missing_ok is False and the path is absent.

async aiomegfile.smart.smart_rename(src_path: str | PathLike, dst_path: str | PathLike, overwrite: bool = True) → str

Rename a file or directory and return the destination path string.

Parameters:

• src_path – Source path to rename.

• dst_path – Destination path.

• overwrite – Whether to overwrite destination when it exists.

Returns:

Destination path string.

Return type:

str

async aiomegfile.smart.smart_save_as(file_object: BinaryIO, path: str | PathLike) → None

Write an opened binary stream to the specified path.

The input stream will not be closed.

Parameters:

• file_object – Stream to be read.

• path – Target path to save the stream content.

async aiomegfile.smart.smart_save_content(path: str | PathLike, content: bytes) → None

Save bytes content to the specified path.

Parameters:

• path – Path to save content.

• content – Bytes content to write.

async aiomegfile.smart.smart_save_text(path: str | PathLike, text: str) → None

Save text to the specified path.

Parameters:

• path – Path to save text.

• text – Text content to write.

async aiomegfile.smart.smart_scan(path: str | PathLike, *, missing_ok: bool = True, followlinks: bool = False, sort: bool = False) → AsyncIterator[str]

Iteratively traverse only files in the given path.

If the path is a file, yields that file only.

Parameters:

• path – Given path.

• missing_ok – If False and the path is missing, raise FileNotFoundError.

• followlinks – Whether to follow symbolic links.

• sort – Whether to request sorted traversal when supported by the filesystem.

Returns:

Async iterator of file paths.

Return type:

T.AsyncIterator[str]

async aiomegfile.smart.smart_scan_stat(path: str | PathLike, *, missing_ok: bool = True, followlinks: bool = False, sort: bool = False) → AsyncIterator[FileEntry]

Iteratively traverse only files in the given path with stats.

If the path is a file, yields that file entry only.

Parameters:

• path – Given path.

• missing_ok – If False and the path is missing, raise FileNotFoundError.

• followlinks – Whether to follow symbolic links.

• sort – Whether to request sorted traversal when supported by the filesystem.

Returns:

Async iterator of FileEntry objects.

Return type:

T.AsyncIterator[FileEntry]

Raises:

FileNotFoundError – If no matches and missing_ok is False.

aiomegfile.smart.smart_scandir(path: str | PathLike) → AsyncContextManager[AsyncIterator[FileEntry]]

Return an async context manager for iterating directory entries.

Parameters:

path – Directory path to scan.

Returns:

Async context manager producing FileEntry items.

Return type:

T.AsyncContextManager[T.AsyncIterator[FileEntry]]

async aiomegfile.smart.smart_stat(path: str | PathLike, *, follow_symlinks: bool = False) → StatResult

Get the status of the path.

Parameters:

• path – Path to stat.

• follow_symlinks – Whether to follow symbolic links when resolving.

Returns:

StatResult for the path.

Return type:

StatResult

async aiomegfile.smart.smart_symlink(src_path: str | PathLike, dst_path: str | PathLike) → None

Create a symbolic link at dst_path pointing to src_path.

Parameters:

• src_path – Target path the link should point to.

• dst_path – Path of the symlink to create.

Raises:

TypeError – If src_path and dst_path are on different filesystems.

async aiomegfile.smart.smart_sync(src_path: str | PathLike, dst_path: str | PathLike, callback: Callable[[str, int], None] | None = None, followlinks: bool = False, callback_after_copy_file: Callable[[str, str], None] | None = None, force: bool = False, overwrite: bool = True, *, worker: int = -1) → None

Sync file or directory to the destination path.

Note

When the parameter is file, this function behaves like smart_copy.

If file and directory of same name and same level, sync considers it as file first.

Parameters:

• src_path – Given source path.

• dst_path – Given destination path.

• callback – Called periodically during copy with source path and bytes written.

• followlinks – False if regard symlink as file, else True.

• callback_after_copy_file – Called after copy success, and the input parameter is src file path and dst file path.

• worker – Maximum number of concurrent workers for copy tasks.

• force – Sync file forcible, do not ignore same files, priority is higher than overwrite.

• overwrite – Whether to overwrite files when they already exist.

Raises:

FileNotFoundError – If source path does not exist.

async aiomegfile.smart.smart_sync_with_progress(src_path: str | PathLike, dst_path: str | PathLike, callback: Callable[[str, int], None] | None = None, followlinks: bool = False, force: bool = False, overwrite: bool = True, *, callback_after_copy_file: Callable[[str, str], None] | None = None, worker: int = -1) → None

Sync file or directory with progress bars.

Parameters:

• src_path – Given source path.

• dst_path – Given destination path.

• callback – Called periodically during copy with source path and bytes written.

• followlinks – False if regard symlink as file, else True.

• force – Sync file forcible, do not ignore same files, priority is higher than overwrite.

• overwrite – Whether to overwrite files when they already exist.

• callback_after_copy_file – Called after copy success, and the input parameter is src file path and dst file path.

• worker – Maximum number of concurrent workers for copy tasks.

Raises:

• FileNotFoundError – If source path does not exist.

• ImportError – If tqdm is not available.

async aiomegfile.smart.smart_touch(path: str | PathLike, exist_ok: bool = True) → None

Create the file if missing, optionally raising on existence.

Parameters:

• path – Path to create.

• exist_ok – Whether to skip raising if the file already exists.

async aiomegfile.smart.smart_unlink(path: str | PathLike, missing_ok: bool = False) → None

Remove (delete) the file.

Parameters:

• path – Path to remove.

• missing_ok – If False, raise when the path does not exist.

Raises:

• FileNotFoundError – When missing_ok is False and the file is absent.

• IsADirectoryError – If the target is a directory.

async aiomegfile.smart.smart_walk(path: str | PathLike, *, followlinks: bool = False) → AsyncIterator[Tuple[str, List[str], List[str]]]

Generate the file names in a directory tree by walking the tree.

Parameters:

• path – Root directory to walk.

• followlinks – Whether to traverse symbolic links to directories.

Returns:

Async iterator of (root, dirs, files).

Return type:

T.AsyncIterator[T.Tuple[str, T.List[str], T.List[str]]]