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
_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
close(file)

asyncio.coroutine()

Close file

Parameters:file – file-object from aioftp.AbstractPathIO.open
exists(path)

asyncio.coroutine()

Check if path exists

Parameters:path (pathlib.Path) – path to check
Return type:bool
is_dir(path)

asyncio.coroutine()

Check if path is directory

Parameters:path (pathlib.Path) – path to check
Return type:bool
is_file(path)

asyncio.coroutine()

Check if path is file

Parameters:path (pathlib.Path) – path to check
Return type:bool
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, ...]
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
read(file, block_size)

asyncio.coroutine()

Read some data from file

Parameters:
Return type:

bytes
rename(source, destination)

asyncio.coroutine()

Rename path

Parameters:
rmdir(path)

asyncio.coroutine()

Remove directory

Parameters:path (pathlib.Path) – path to remove
seek(file, offset, whence=0)

asyncio.coroutine()

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

Parameters:
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
state

Shared pathio state per server

asyncio.coroutine()

Remove file

Parameters:path (pathlib.Path) – path to remove
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: aioftp.pathio.AbstractPathIO

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

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

Bases: aioftp.pathio.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: aioftp.pathio.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