Path abstraction layer API

class aioftp.AbstractPathIO(timeout=None, connection=None, state=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

abstract async _open(path, mode)

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

abstract async close(file)

asyncio.coroutine()

Close file

Parameters

file – file-object from aioftp.AbstractPathIO.open

abstract async exists(path)

asyncio.coroutine()

Check if path exists

Parameters

path (pathlib.Path) – path to check

Return type

bool

abstract async is_dir(path)

asyncio.coroutine()

Check if path is directory

Parameters

path (pathlib.Path) – path to check

Return type

bool

abstract async is_file(path)

asyncio.coroutine()

Check if path is file

Parameters

path (pathlib.Path) – path to check

Return type

bool

abstract list(path)

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, ...]
abstract async mkdir(path, *, parents=False, exist_ok=False)

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, **kwargs)

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

Return type

aioftp.pathio.AsyncPathIOContext

abstract async read(file, block_size)

asyncio.coroutine()

Read some data from file

Parameters
Return type

bytes

abstract async rename(source, destination)

asyncio.coroutine()

Rename path

Parameters
abstract async rmdir(path)

asyncio.coroutine()

Remove directory

Parameters

path (pathlib.Path) – path to remove

abstract async seek(file, offset, whence=0)

asyncio.coroutine()

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

Parameters
abstract async stat(path)

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

Shared pathio state per server

asyncio.coroutine()

Remove file

Parameters

path (pathlib.Path) – path to remove

abstract async write(file, data)

asyncio.coroutine()

Write some data to file

Parameters
class aioftp.pathio.AsyncPathIOContext(pathio, args, kwargs)

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)

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

class aioftp.PathIO(timeout=None, connection=None, state=None)

Bases: AbstractPathIO

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

class aioftp.AsyncPathIO(*args, executor=None, **kwargs)

Bases: AbstractPathIO

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(*args, state=None, cwd=None, **kwargs)

Bases: AbstractPathIO

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, reason=None, **kwargs)

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