aiomegfile.filesystem.s3 module

class aiomegfile.filesystem.s3.MultiPartWriter(client: S3Client, path: str | PathLike)[source]

Bases: AioClosable

async close() → None

Flush and close the file-like object.

This method has no effect if the file is already closed.

async upload_part(part_num: int, file_obj: BytesIO) → None[source]
async upload_part_by_paths(part_num: int, paths: List[Tuple[str | PathLike, str]]) → None[source]
async upload_part_copy(part_num: int, path: str | PathLike, copy_source_range: str | None = None) → None[source]
class aiomegfile.filesystem.s3.S3FileSystem(profile_name: str | None = None, *, aws_access_key_id: str | None = None, aws_secret_access_key: str | None = None, endpoint_url: str | None = None, addressing_style: str | None = None)[source]

Bases: BaseFileSystem

Protocol for s3 operations.

async absolute(path: str) → str[source]

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

Parameters:

path – The path to make absolute.

Returns:

Absolute path string.

async access(path: str, mode: Access = Access.READ) → bool[source]

Test if path has access permission described by mode

Parameters:

mode – access mode

Returns:

bool, if the bucket of s3_url has read/write access.

build_uri(path: str) → str

Build URI for the filesystem by path part.

Parameters:

path – Path without protocol.

Returns:

Generated URI string.

async concat(src_paths: List[str | PathLike], dst_path: str | PathLike, block_size: int = 8388608) → None[source]

Concatenate s3 files to one file.

Parameters:

  • src_paths – Given source paths

  • dst_path – Given destination path

async copy(src_path: str, dst_path: str, callback: Callable[[int], None] | None = None, followlinks: bool = False) → str[source]

Copy single file, not directory

Parameters:

  • src_path – Given source path

  • dst_path – Given destination path

  • callback – Called periodically during copy, and the input parameter is the data size (in bytes) of copy since the last call

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

Returns:

Destination path after copy.

async download(src_path: str, dst_path: str, callback: Callable[[int], None] | None = None) → None[source]

download file

Parameters:

  • src_path – Given source path

  • dst_path – Given destination path

  • callback – Called periodically during download with bytes written.

Returns:

None.

async exists(path: str, followlinks: bool = False) → bool[source]

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.

classmethod from_uri(uri: str)

Return new instance of this class

Parameters:

uri – URI string.

Returns:

new instance of new path

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

Return entries whose paths match the glob pattern.

Parameters:

  • path – S3 glob pattern without protocol.

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

  • missing_ok – If False and no matches exist, raise FileNotFoundError.

  • sort – Whether to preserve lexicographic listing order.

Returns:

Async iterator of FileEntry objects.

Raises:

S3FileNotFoundError – If no matches and missing_ok is False.

async is_dir(path: str, followlinks: bool = False) → bool[source]

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 is_file(path: str, followlinks: bool = False) → bool[source]

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.

Return True if the path points to a symbolic link.

Parameters:

path – The path to check.

Returns:

True if the path is a symbolic link, otherwise False.

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

Return the MD5 checksum for an S3 path.

Parameters:

  • path – S3 path to compute MD5.

  • recalculate – If True, recompute MD5 by reading the object.

  • followlinks – If True, follow symbolic links.

Returns:

MD5 hex digest.

Return type:

str

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

Create a directory.

Parameters:

  • path – Directory path to create.

  • mode – Permission bits for the new directory.

  • parents – Whether to create missing parents.

  • exist_ok – Whether to ignore if the directory exists.

Raises:

FileExistsError – When directory exists and exist_ok is False.

async move(src_path: str, dst_path: str, overwrite: bool = True) → str[source]

Move file or directory.

Parameters:

  • src_path – Given source path.

  • dst_path – Given destination path.

  • overwrite – Whether to overwrite file when exists.

Returns:

Destination path after move.

Raises:

FileExistsError – If destination exists and overwrite is False.

open(path: str, 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:

  • path – File path to open.

  • mode – File open mode.

  • buffering – Buffering policy. Not used in s3.

  • encoding – Text encoding when using text modes.

  • errors – Error handling strategy for encoding/decoding.

  • newline – Newline handling in text mode.

  • kwargs – Extra open options for compatibility with megfile.

Returns:

Async file context manager.

parse_uri(uri: str) → str[source]

Parse the path part from a URI.

Parameters:

uri – URI string.

Returns:

Path part string.

protocol = 's3'

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

Parameters:

path – The symbolic link path.

Returns:

Target path of the symbolic link.

async remove(path: str, missing_ok: bool = False) → None[source]

Remove (delete) the file or directory.

If path is a file, remove it directly. If path is a directory, remove it and all its contents recursively.

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.

same_endpoint(other_filesystem: BaseFileSystem) → bool[source]

Return whether this filesystem points to the same endpoint.

Parameters:

other_filesystem – Filesystem to compare.

Returns:

True if both represent the same endpoint.

scandir(path: str) → AsyncContextManager[AsyncIterator[FileEntry]][source]
Return an iterator of FileEntry objects corresponding to the entries

in the directory given by path.

Parameters:

path (str) – Directory path to scan.

Returns:

Async context manager yielding an async iterator of FileEntry objects.

Return type:

T.AsyncContextManager[T.AsyncIterator[FileEntry]]

scanfile(path, sort: bool = False) → AsyncContextManager[AsyncIterator[FileEntry]][source]

Iteratively traverse only files in given directory. Every iteration on generator yields FileEntry object.

Parameters:

  • path – Directory or file path to traverse.

  • sort – Whether to preserve lexicographic listing order.

Returns:

Async context manager yielding an async iterator of FileEntry objects.

Return type:

T.AsyncContextManager[T.AsyncIterator[FileEntry]]

async stat(path: str, followlinks: bool = False) → StatResult[source]

Get the status of the path.

Parameters:

  • path – Path to stat.

  • followlinks – Whether to follow symbolic links.

Raises:

FileNotFoundError – If the path does not exist.

Returns:

Populated StatResult for the path.

Create a symbolic link pointing to self named dst_path.

Parameters:

  • src_path – The source path the symbolic link points to.

  • dst_path – The symbolic link path.

async upload(src_path: str, dst_path: str, callback: Callable[[int], None] | None = None) → None[source]

Upload a local file to S3.

Parameters:

  • src_path – Local source file path.

  • dst_path – S3 destination path.

  • callback – Called periodically during upload with bytes written.

Returns:

None.

aiomegfile.filesystem.s3.get_access_token(profile_name: str | None = None, config: dict | None = None) → Tuple[str | None, str | None, str | None][source]
aiomegfile.filesystem.s3.get_config_in_file(profile_name: str | None = None, session: AioSession | None = None) → dict[source]
aiomegfile.filesystem.s3.get_endpoint_url(profile_name: str | None = None, config: dict | None = None) → str[source]

Get the endpoint url of S3

Returns:

S3 endpoint url

aiomegfile.filesystem.s3.get_env_var(env_name: str, profile_name: str | None = None) → str | None[source]
async aiomegfile.filesystem.s3.get_s3_client(profile_name: str | None = None, *, aws_access_key_id: str | None = None, aws_secret_access_key: str | None = None, aws_session_token: str | None = None, endpoint_url: str | None = None, addressing_style: str | None = None) → S3Client[source]

Get a cached S3 client bound to the current event loop.

Parameters:

  • profile_name (T.Optional[str]) – Optional AWS profile name.

  • aws_access_key_id (T.Optional[str]) – Optional AWS access key ID override.

  • aws_secret_access_key (T.Optional[str]) – Optional AWS secret access key override.

  • aws_session_token (T.Optional[str]) – Optional AWS session token override.

  • endpoint_url (T.Optional[str]) – Optional custom S3 endpoint URL.

  • addressing_style (T.Optional[str]) – Optional S3 addressing style.

Returns:

An initialized S3 client bound to the current loop.

Return type:

S3Client

aiomegfile.filesystem.s3.get_session(profile_name: str | None = None) → AioSession[source]
aiomegfile.filesystem.s3.is_s3(path: str | PathLike) → bool[source]

  1. According to aws-cli , test if a path is s3 path.

  2. megfile also support the path like s3[+profile_name]://bucket/key

Parameters:

path – Path to be tested

Returns:

True if path is s3 path, else False

aiomegfile.filesystem.s3.parse_s3_path(s3_path: str | PathLike) → tuple[str, str][source]