Path abstraction layer API

class aioftp.AbstractPathIO(timeout: float | int | None = None, connection: Connection | None = None, state: list[Node] | None = None)

Abstract class for path io operations.

Parameters:

  • timeout (float, int or None) – timeout used by with_timeout decorator

  • connection (aioftp.Connection) – server connection that is accessing this PathIO

  • state – shared pathio state per server

abstractmethod async _open(path: PathType, mode: str) → BytesIO

asyncio.coroutine()

Open file. You should implement “mode” argument, which can be: “rb”, “wb”, “ab” (read, write, append. all binary). Return type depends on implementation, anyway the only place you need this file-object is in your implementation of read, write and close

Parameters:

  • path (pathlib.Path) – path to create

  • mode (str) – specifies the mode in which the file is opened (“rb”, “wb”, “ab”, “r+b” (read, write, append, read/write, all binary))

Returns:

file-object

abstractmethod async close(file: BytesIO) → None

asyncio.coroutine()

Close file

Parameters:

file – file-object from aioftp.AbstractPathIO.open

abstractmethod async exists(path: PathType) → bool

asyncio.coroutine()

Check if path exists

Parameters:

path (pathlib.Path) – path to check

Return type:

bool

abstractmethod async is_dir(path: PathType) → bool

asyncio.coroutine()

Check if path is directory

Parameters:

path (pathlib.Path) – path to check

Return type:

bool

abstractmethod async is_file(path: PathType) → bool

asyncio.coroutine()

Check if path is file

Parameters:

path (pathlib.Path) – path to check

Return type:

bool

abstractmethod list(path: PathType) → AsyncIterable[PathType]

Create instance of subclass of aioftp.AbstractAsyncLister. You should subclass and implement __anext__ method for aioftp.AbstractAsyncLister and return new instance.

Parameters:

path (pathlib.Path) – path to list

Return type:

aioftp.AbstractAsyncLister

Usage:

>>> async for p in pathio.list(path):
...     # do

or borring instance of list:

>>> paths = await pathio.list(path)
>>> paths
[path, path, path, ...]
abstractmethod async mkdir(path: PathType, *, parents: bool = False, exist_ok: bool = False) → None

asyncio.coroutine()

Make directory

Parameters:

  • path (pathlib.Path) – path to create

  • parents (bool) – create parents is does not exists

  • exist_ok (bool) – do not raise exception if directory already exists

open(*args: Unpack[tuple[PathType]], **kwargs: Unpack[_OpenKwargs]) → AsyncPathIOContext

Create instance of aioftp.pathio.AsyncPathIOContext, parameters passed to aioftp.AbstractPathIO._open()

Return type:

aioftp.pathio.AsyncPathIOContext

abstractmethod async read(file: BytesIO, block_size: int) → bytes

asyncio.coroutine()

Read some data from file

Parameters:
Return type:

bytes

abstractmethod async rename(source: PathType, destination: PathType) → PathType

asyncio.coroutine()

Rename path

Parameters:
abstractmethod async rmdir(path: PathType) → None

asyncio.coroutine()

Remove directory

Parameters:

path (pathlib.Path) – path to remove

abstractmethod async seek(file: BytesIO, offset: int, whence: int = 0) → int

asyncio.coroutine()

Change the stream position to the given byte offset. Same behaviour as io.IOBase.seek()

Parameters:
abstractmethod async stat(path: PathType) → stat_result

asyncio.coroutine()

Get path stats

Parameters:

path (pathlib.Path) – path, which stats need

Returns:

path stats. For proper work you need only this stats: st_size, st_mtime, st_ctime, st_nlink, st_mode

Return type:

same as os.stat_result

property state: list[Node] | None

Shared pathio state per server

asyncio.coroutine()

Remove file

Parameters:

path (pathlib.Path) – path to remove

abstractmethod async write(file: BytesIO, data: bytes) → int

asyncio.coroutine()

Write some data to file

Parameters:
class aioftp.pathio.AsyncPathIOContext(pathio: PathIOType, args: tuple[Any], kwargs: _OpenKwargs)

Async pathio context.

Usage:

>>> async with pathio.open(filename) as file_in:
...     async for block in file_in.iter_by_block(size):
...         # do

or borring:

>>> file = await pathio.open(filename)
... data = await file.read(size)
... await file.write(data)
... await file.close()
aioftp.pathio.universal_exception(coro: Callable[[UniversalExceptionParamSpec], Awaitable[UniversalExceptionReturnType]]) → Callable[[UniversalExceptionParamSpec], Awaitable[UniversalExceptionReturnType]]

Decorator. Reraising any exception (except CancelledError and NotImplementedError) with universal exception aioftp.PathIOError

class aioftp.PathIO(timeout: float | int | None = None, connection: Connection | None = None, state: list[Node] | None = None)

Bases: AbstractPathIO[Path]

Blocking path io. Directly based on pathlib.Path methods.

class aioftp.AsyncPathIO(timeout: float | int | None = None, connection: Connection | None = None, state: list[Node] | None = None, executor: Executor | None = None)

Bases: AbstractPathIO[Path]

Non-blocking path io. Based on asyncio.BaseEventLoop.run_in_executor() and pathlib.Path methods. It’s really slow, so it’s better to avoid usage of this path io layer.

Parameters:

executor (concurrent.futures.Executor) – executor for running blocking tasks

class aioftp.MemoryPathIO(timeout: float | int | None = None, connection: Connection | None = None, state: list[Node] | None = None, cwd: str | PurePosixPath | None = None)

Bases: AbstractPathIO[PurePosixPath]

Non-blocking path io. Based on in-memory tree. It is just proof of concept and probably not so fast as it can be.

class aioftp.PathIOError(*args: Any, reason: tuple[type[BaseException], BaseException, TracebackType] | tuple[None, None, None] | None = None, **kwargs: Any)

Universal exception for any path io errors.

>>> try:
...     # some client/server path operation
... except PathIOError as exc:
...     type, value, traceback = exc.reason
...     if isinstance(value, SomeException):
...         # handle
...     elif ...
...         # handle