aiomegfile.smart_path module

class aiomegfile.smart_path.SmartPath(uri: str | PathLike)[source]

Bases: PathLike

async absolute() SmartPath[source]

Make the path absolute, without normalization or resolving symlinks. Returns a new path object

Returns:

Absolute SmartPath without symlink resolution.

async abspath() str[source]

Return a normalized absolute version of the path.

Returns:

Absolute path string.

Return type:

str

async access(mode: Access = Access.READ) bool[source]

Test if path has access permission described by mode

Parameters:

mode (Access, optional) – access mode, defaults to Access.READ

Raises:

NotImplementedError – If the filesystem does not support access checks.

Returns:

True if the path has the specified access permission, False otherwise.

Return type:

bool

property anchor: str
async as_posix() str[source]

Return a string representation of the path with forward slashes (/)

async as_uri() str[source]

Return the path with its protocol prefix (e.g., file:///root).

async chmod(mode: int, *, follow_symlinks: bool = True) None[source]

Change the permission bits of the path.

Parameters:
  • mode – New permission bits.

  • follow_symlinks – Whether to follow symbolic links.

Raises:

NotImplementedError – If protocol is not file.

async copy(target: str | PathLike, *, callback: Callable[[int], None] | None = None, follow_symlinks: bool = False, overwrite: bool = True) SmartPath[source]

copy file

Parameters:
  • target – Given destination path

  • callback – Called periodically during copy with bytes written.

  • follow_symlinks – whether or not follow symbolic link

  • overwrite – whether or not overwrite file when exists

Returns:

Target SmartPath.

async copy_file(target: str | PathLike, callback: Callable[[int], None] | None = None) SmartPath[source]

copy file only

Parameters:
  • target – Given destination path

  • callback – Called periodically during copy with bytes written.

Returns:

Target SmartPath.

async copy_into(target_dir: str | PathLike, *, callback: Callable[[int], None] | None = None, follow_symlinks: bool = False, overwrite: bool = True) SmartPath[source]

copy file or directory into dst directory

Parameters:
  • target_dir – Given destination path

  • callback – Called periodically during copy with bytes written.

  • follow_symlinks – whether or not follow symbolic link

  • overwrite – whether or not overwrite file when exists

Returns:

Target SmartPath.

async cwd() SmartPath[source]

Return current working directory.

Returns:

Current working directory SmartPath.

Return type:

SmartPath

Raises:

NotImplementedError – If protocol is not file.

async exists(*, followlinks: bool = False) bool[source]

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

Parameters:

followlinks – Whether to follow symbolic links.

Returns:

True if the path exists, otherwise False.

async expanduser() SmartPath[source]

Return a new path with expanded ~ and ~user constructs.

Returns:

Expanded SmartPath.

Return type:

SmartPath

Raises:

NotImplementedError – If protocol is not file.

classmethod from_uri(uri: str | PathLike) SmartPath[source]

Return new instance of this class

Parameters:

uri – new path

Returns:

new instance of new path

Return type:

“SmartPath”

async full_match(pattern: str, *, case_sensitive: bool | None = None) bool[source]

Return a function that matches the entire path against the provided glob-style pattern.

Parameters:
  • pattern (str) – The glob-style pattern to match against.

  • case_sensitive (Optional[bool]) – Whether the matching should be case-sensitive. If None, the default behavior of fnmatch is used.

Returns:

Returns True if it matches the pattern, False otherwise.

Return type:

bool

async getmtime(*, follow_symlinks: bool = False) float[source]

Return the time of last modification of the file as a timestamp.

async getsize(*, follow_symlinks: bool = False) int[source]

Return the size of the file in bytes.

async glob(pattern: str, recursive: bool = True, missing_ok: bool = True, sort: bool = False) List[SmartPath][source]

Return files whose paths match the glob pattern.

Parameters:
  • pattern – Glob pattern to match relative to this path.

  • 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:

List of matching SmartPath instances.

async glob_stat(pattern: str, recursive: bool = True, missing_ok: bool = True, sort: bool = False) AsyncIterator[FileEntry][source]

Return entries whose paths match the glob pattern with stats.

Parameters:
  • pattern – Glob pattern to match relative to this path.

  • 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 matching FileEntry objects.

Return type:

T.AsyncIterator[FileEntry]

Raises:

FileNotFoundError – If no matches and missing_ok is False.

async group() str[source]

Return the name of the group owning the file.

Returns:

Group name.

Return type:

str

Raises:

NotImplementedError – If protocol is not file or platform does not support group lookups.

Make this path a hard link to the same file as target.

Parameters:

target – Existing path to hard link to.

Raises:

NotImplementedError – If protocol does not support hard links.

async home() SmartPath[source]

Return the home directory path.

Returns:

Home directory SmartPath.

Return type:

SmartPath

Raises:

NotImplementedError – If protocol is not file.

async iglob(pattern: str, recursive: bool = True, missing_ok: bool = True, sort: bool = False) AsyncIterator[SmartPath][source]

Return an iterator of files whose paths match the glob pattern.

Parameters:
  • pattern – Glob pattern to match relative to this path.

  • 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 matching SmartPath objects.

async is_absolute() bool[source]

Return True if the path is absolute.

Returns:

True if path is absolute, otherwise False.

Return type:

bool

async is_dir(followlinks: bool = False) bool[source]

Return True if the path points to a directory.

Parameters:

followlinks – Whether to follow symbolic links.

Returns:

True if the path is a directory, otherwise False.

async is_file(followlinks: bool = False) bool[source]

Return True if the path points to a regular file.

Parameters:

followlinks – Whether to follow symbolic links.

Returns:

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

async is_relative_to(other: str | PathLike) bool[source]

Return True if this path is relative to the given path.

Parameters:

other – Target path to compare against.

Returns:

True if relative, otherwise False.

Return True if the path points to a symbolic link.

Returns:

True if the path is a symlink, otherwise False.

async iterdir() AsyncIterator[SmartPath][source]

Get all contents of given fs path. The result is in ascending alphabetical order.

Returns:

All contents have in the path in ascending alphabetical order

async joinpath(*other_paths: str | PathLike) SmartPath[source]

Calling this method is equivalent to combining the path with each of the other arguments in turn

Parameters:

other_paths – Additional path components to join.

Returns:

A new SmartPath representing the combined path.

async lchmod(mode: int) None[source]

Change permissions of a symbolic link without following it.

Parameters:

mode – New permission bits.

async listdir() List[str][source]

Return the names of the entries in the directory this path points to.

Returns:

List of entry names.

Return type:

T.List[str]

async load() BinaryIO[source]

Read all content in binary into memory.

Caller is responsible for closing the returned BinaryIO.

Returns:

BinaryIO containing file content.

Return type:

T.BinaryIO

async lstat() StatResult[source]

Like stat() but, if the path points to a symbolic link, return the symbolic link’s information rather than its target’s.

Returns:

StatResult for the link itself.

async match(pattern: str, *, case_sensitive: bool | None = None) bool[source]

Match this path against the provided glob-style pattern. Return True if matching is successful, False otherwise.

This method is similar to full_match(), but the recursive wildcard “**” isn’t supported (it acts like non-recursive “*”)

Parameters:
  • pattern – Glob pattern to match against the full URI.

  • case_sensitive – Whether matching should be case sensitive.

Returns:

True if the path matches the pattern, otherwise False.

async md5(recalculate: bool = False, followlinks: bool = False) str[source]

Return the MD5 checksum for the path.

If the filesystem provides an optimized md5 implementation, it will be used. Otherwise, this method reads file contents to compute the checksum, matching the behavior of megfile’s FSPath.md5.

Parameters:
  • recalculate – Whether to force recalculation when filesystem supports cached MD5 metadata.

  • followlinks – Whether to follow symbolic links when supported.

Returns:

MD5 hex digest.

Return type:

str

async mkdir(mode: int = 511, parents: bool = False, exist_ok: bool = False) None[source]

Create a directory.

Parameters:
  • mode – Permission bits for the new directory.

  • parents – Whether to create parents as needed.

  • exist_ok – Whether to ignore if the directory exists.

async move(target: str | PathLike, overwrite: bool = True) SmartPath[source]

move file

Parameters:
  • target – Given destination path

  • overwrite – whether or not overwrite file when exists

Returns:

Destination SmartPath after move.

async move_into(target_dir: str | PathLike) SmartPath[source]

move file or directory into dst directory

Parameters:

target_dir – Given destination path

Returns:

Destination SmartPath inside the target directory.

property name: str

A string representing the final path component, excluding the drive and root

open(mode: str = 'r', buffering: int = -1, encoding: str | None = None, errors: str | None = None, newline: str | None = None, **kwargs: Any) AsyncContextManager[source]

Open the file with mode.

Parameters:
  • 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.

async owner() str[source]

Return the name of the user owning the file.

Returns:

Owner username.

Return type:

str

Raises:

NotImplementedError – If protocol is not file or platform does not support user lookups.

property parent: SmartPath

The logical parent of the path

property parents: URIPathParents

An immutable sequence providing access to the logical ancestors of the path

property parts: Tuple[str, ...]

A tuple giving access to the path’s various components

property protocol: str

Return the protocol for this path, preferring aliases when configured.

Returns:

Protocol string for display.

Return type:

str

async read_bytes() bytes[source]

Return the binary contents of the pointed-to file as a bytes object.

Returns:

File content in bytes.

async read_text(encoding: str | None = None, errors: str | None = None, newline: str | None = None) str[source]

Return the decoded contents of the pointed-to file as a string.

Parameters:
  • encoding – Optional text encoding.

  • errors – Optional error handling strategy.

  • newline – Optional newline handling policy.

Returns:

File content as text.

Return a new path representing the symbolic link’s target.

async realpath() str[source]

Return the canonical path of the path.

Returns:

Canonical path string.

Return type:

str

async relative_to(other: str | PathLike) str[source]

Compute a version of this path relative to the path represented by other. If it’s impossible, ValueError is raised.

Parameters:

other – Target path to compute the relative path against.

Returns:

Relative path string.

Raises:
  • TypeError – If other is missing.

  • ValueError – If this path is not under the given other path.

async relpath(start: str | PathLike) str[source]

Return the relative path from start to this path.

Parameters:

start – Base path to compute the relative path against.

Returns:

Relative path string.

Return type:

str

Raises:
  • TypeError – If start is not provided.

  • ValueError – If this path is not under start.

async remove(missing_ok: bool = False) None[source]

Remove (delete) the file or directory.

Parameters:

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

async rename(target: str | PathLike, overwrite: bool = True) SmartPath[source]

rename file

Parameters:
  • target – Given destination path

  • overwrite – whether or not overwrite file when exists

Returns:

Target SmartPath after rename.

Raises:

FileExistsError – If destination exists and overwrite is False.

async replace(target: str | PathLike, overwrite: bool = True) SmartPath[source]

move file

Parameters:
  • target – Given destination path

  • overwrite – whether or not overwrite file when exists

Returns:

Destination SmartPath after replace.

async resolve(strict: bool = False) SmartPath[source]

Alias of realpath.

Parameters:

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

Returns:

Resolved absolute SmartPath.

Raises:

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

async rglob(pattern: str, recursive: bool = True, sort: bool = False) List[SmartPath][source]

This is like calling Path.glob() with **/ added in front of the given relative pattern

Parameters:
  • pattern – Glob pattern to match recursively.

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

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

Returns:

List of matching SmartPath instances.

async rmdir() None[source]

Remove (delete) the empty directory.

Raises:

NotADirectoryError – If the target is not a directory.

property root: str

Return the protocol root for this path.

Returns:

Protocol root string.

Return type:

str

async samefile(other_path: str | PathLike) bool[source]

Return whether this path points to the same file

Parameters:

other_path – Path to compare.

Returns:

True if both represent the same file.

async save(file_object: BinaryIO) None[source]

Write an opened binary stream to the path.

The input stream will not be closed.

Parameters:

file_object – Stream to be read.

async scan(missing_ok: bool = True, followlinks: bool = False, sort: bool = False) AsyncIterator[str][source]

Iteratively traverse only files in the given directory. Every iteration on the generator yields a path string.

If path is a file path, yields the file only If path is a non-existent path, return an empty generator If path is a bucket path, return all file paths in the bucket

Parameters:
  • missing_ok – If False and there’s no file in the directory, raise FileNotFoundError.

  • followlinks – Whether to follow symbolic links.

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

Raises:

FileNotFoundError – If no matches and missing_ok is False.

Returns:

Async iterator of file path strings.

Return type:

T.AsyncIterator[str]

async scan_stat(missing_ok: bool = True, followlinks: bool = False, sort: bool = False) AsyncIterator[FileEntry][source]

Iteratively traverse only files in the given directory. Every iteration on the generator yields a tuple of path string and file stat.

Parameters:
  • missing_ok – If False and there’s no file in the directory, raise FileNotFoundError.

  • followlinks – Whether to follow symbolic links.

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

Raises:

FileNotFoundError – If no matches and missing_ok is False.

Returns:

Async iterator of FileEntry objects.

Return type:

T.AsyncIterator[FileEntry]

scandir() AsyncContextManager[AsyncIterator[FileEntry]][source]

Return an async context manager for directory entries.

Returns:

Async context manager yielding FileEntry items.

Return type:

T.AsyncContextManager[T.AsyncIterator[FileEntry]]

async stat(*, follow_symlinks: bool = False) StatResult[source]

Get the status of the path.

Parameters:

follow_symlinks – Whether to follow symbolic links when resolving.

Returns:

StatResult for the path.

property stem: str

The final path component, without its suffix

property suffix: str

The file extension of the final component

property suffixes: List[str]

A list of the path’s file extensions

Create a symbolic link pointing to this path named dst_path.

Parameters:

dst_path – Path of the symlink to create.

Raises:

TypeError – If filesystems differ.

Make this path a symbolic link to target. symlink_to’s arguments is the reverse of symlink’s.

Parameters:
  • target – Destination the new link should point to.

  • target_is_directory – Compatibility argument, ignored.

async touch(exist_ok: bool = True) None[source]

Create the file if missing, optionally raising on existence.

Parameters:

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

Remove (delete) the file.

Parameters:

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

Raises:

IsADirectoryError – If the target is a directory.

async utime(atime: float | int, mtime: float | int) None[source]

Set the access and modified times of the file.

Parameters:
  • atime – The access time to be set.

  • mtime – The modification time to be set.

Raises:

NotImplementedError – If protocol is not file.

async walk(follow_symlinks: bool = False) AsyncIterator[Tuple[str, List[str], List[str]]][source]

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

Parameters:

follow_symlinks – Whether to traverse symbolic links to directories.

Returns:

Async iterator of (root, dirs, files).

async with_name(name: str) SmartPath[source]

Return a new path with the name changed.

Parameters:

name – New file or directory name.

Returns:

SmartPath with the name changed.

async with_stem(stem: str) SmartPath[source]

Return a new path with the stem changed.

Parameters:

stem – New stem (basename without suffix).

Returns:

SmartPath with updated stem.

async with_suffix(suffix: str) SmartPath[source]

Return a new path with the suffix changed.

Parameters:

suffix – New suffix including leading dot.

Returns:

SmartPath with the suffix changed.

async write_bytes(data: bytes)[source]

Open the file pointed to in bytes mode, write data to it, and close the file

Parameters:

data – Bytes to write to the file.

Returns:

Number of bytes written.

async write_text(data: str, encoding: str | None = None, errors: str | None = None, newline: str | None = None)[source]

Open the file pointed to in text mode, write data to it, and close the file. The optional parameters have the same meaning as in open().

Parameters:
  • data – Text content to write.

  • encoding – Optional text encoding.

  • errors – Optional error handling strategy.

  • newline – Optional newline handling policy.

Returns:

Number of characters written.

class aiomegfile.smart_path.URIPathParents(path: SmartPath)[source]

Bases: Sequence