Client API¶

class aioftp.Client(*, socket_timeout: float | int | None = None, connection_timeout: float | int | None = None, read_speed_limit: int | None = None, write_speed_limit: int | None = None, path_timeout: float | int | None = None, path_io_factory: type[~aioftp.pathio.AbstractPathIO[~pathlib.Path]] = <class 'aioftp.pathio.PathIO'>, encoding: str = 'utf-8', ssl: ~ssl.SSLContext | bool | None = None, parse_list_line_custom: ~aioftp.client.ParseListLineCustomCallable | None = None, parse_list_line_custom_first: bool = True, passive_commands: tuple[str, ...] = ('epsv', 'pasv'), **siosocks_asyncio_kwargs: ~typing.Any)¶

Bases: BaseClient

FTP client.

Parameters:

socket_timeout (float, int or None) – timeout for read operations
connection_timeout (float, int or None) – timeout for connection
read_speed_limit – download speed limit in bytes per second
write_speed_limit – upload speed limit in bytes per second
path_timeout (float, int or None) – timeout for path-related operations (make directory, unlink file, etc.)
path_io_factory (aioftp.AbstractPathIO) – factory of «path abstract layer»
encoding (str) – encoding to use for convertion strings to bytes
ssl (bool or ssl.SSLContext) – if given and not false, a SSL/TLS transport is created (by default a plain TCP transport is created). If ssl is a ssl.SSLContext object, this context is used to create the transport; if ssl is True, a default context returned from ssl.create_default_context() is used. Please look asyncio.loop.create_connection() docs.
parse_list_line_custom (callable) – callable, which receive exactly one argument: line of type bytes or str. Should return tuple of Path object and dictionary with fields “modify”, “type”, “size”. For more information see sources.
parse_list_line_custom_first (bool) – Should be custom parser tried first or last
**siosocks_asyncio_kwargs –
siosocks key-word only arguments

async abort(*, wait: bool = True) → None¶

asyncio.coroutine()

Request data transfer abort.

Parameters:: wait (bool) – wait for abort response [426]→226 if True

append_stream(destination: str | PurePosixPath, *, offset: int = 0) → AsyncEnterableInstanceProtocol[DataConnectionThrottleStreamIO]¶

Create stream for append (write) data to destination file.

Parameters:

destination (str or pathlib.PurePosixPath) – destination path of file on server side
offset (int) – byte offset for stream start position

Return type:

aioftp.DataConnectionThrottleStreamIO

async change_directory(path: str | PurePosixPath = '..') → None¶

asyncio.coroutine()

Change current directory. Goes «up» if no parameters passed.

Parameters:: path (str or pathlib.PurePosixPath) – new directory, goes «up» if omitted

check_codes(expected_codes: Code | tuple[Code, ...], received_code: Code, info: list[str]) → None¶

Checks if any of expected matches received.

Parameters:

expected_codes (tuple) – tuple of expected codes
received_code (aioftp.Code) – received code for matching
info (list) – list of response lines from server

Raises:

aioftp.StatusCodeError – if received code does not matches any expected code

close() → None¶: Close connection.

asyncio.coroutine()

Basic command logic.

Send command if not omitted.
Yield response until no wait code matches.
Check code for expected.

Parameters:

command (str) – command line
expected_codes (tuple of str or str) – tuple of expected codes or expected code
wait_codes (tuple of str or str) – tuple of wait codes or wait code
censor_after (None or int) – index after which the line should be censored when logging

async connect(host: str, port: int = 21) → list[str]¶

asyncio.coroutine()

Connect to server.

Parameters:

host (str) – host name for connection
port (int) – port number for connection

classmethod context(host: str, port: int = 21, user: str = 'anonymous', password: str = 'anon@', account: str = '', upgrade_to_tls: bool = False, **kwargs: Unpack[BaseClientArgs]) → AsyncGenerator[Self]¶

Classmethod async context manager. This create aioftp.Client, make async call to aioftp.Client.connect(), aioftp.Client.login() on enter and aioftp.Client.quit() on exit.

Parameters:

host (str) – host name for connection
port (int) – port number for connection
user (str) – username
password (str) – password
account (str) – account (almost always blank)
**kwargs –
keyword arguments, which passed to aioftp.Client

>>> async with aioftp.Client.context("127.0.0.1") as client:
...     # do

async download(source: str | PurePosixPath, destination: str | PurePosixPath = '', *, write_into: bool = False, block_size: int = 8192) → None¶

asyncio.coroutine()

High level download method for downloading files and directories recursively and save them to the file system.

Parameters:

source (str or pathlib.PurePosixPath) – source path of file or directory on server side
destination (str or pathlib.Path) – destination path of file or directory on client side
write_into (bool) – write source into destination (if you want download file and change it name, as well with directories)
block_size (int) – block size for transaction

download_stream(source: str | PurePosixPath, *, offset: int = 0) → AsyncEnterableInstanceProtocol[DataConnectionThrottleStreamIO]¶

asyncio.coroutine()

Create stream for read data from source file.

Parameters:

source (str or pathlib.PurePosixPath) – source path of file on server side
offset (int) – byte offset for stream start position

Return type:

aioftp.DataConnectionThrottleStreamIO

async exists(path: str | PurePosixPath) → bool¶

asyncio.coroutine()

Check path for existence.

Parameters:: path (str or pathlib.PurePosixPath) – path to check
Return type:: bool

static format_date_time(d: datetime) → str¶

Formats dates from strptime in a consistent format

Parameters:: d (datetime) – return value from strptime
Return type:: :py:class`str`

async get_current_directory() → PurePosixPath¶

asyncio.coroutine()

Getting current working directory.

Return type:: pathlib.PurePosixPath

async get_passive_connection(conn_type: Literal['I', 'A', 'E', 'L'] = 'I', commands: Sequence[str] | None = None) → tuple[StreamReader, StreamWriter]¶

asyncio.coroutine()

Getting pair of reader, writer for passive connection with server.

Parameters:

conn_type (str) – connection type (“I”, “A”, “E”, “L”)
commands (list or None) – sequence of commands to try to initiate passive server creation. First success wins. Default is EPSV, then PASV. Overwrites the parameters passed when initializing the client.

Return type:

(asyncio.StreamReader, asyncio.StreamWriter)

get_stream(command: str | None = None, expected_codes: str | tuple[str, ...] = (), wait_codes: str | tuple[str, ...] = (), censor_after: int | None = None, conn_type: Literal['I', 'A', 'E', 'L'] = 'I', offset: int = 0) → DataConnectionThrottleStreamIO¶

asyncio.coroutine()

Create aioftp.DataConnectionThrottleStreamIO for straight read/write io.

Parameters:

command_args – arguments for aioftp.Client.command()
conn_type (str) – connection type (“I”, “A”, “E”, “L”)
offset (int) – byte offset for stream start position

Return type:

aioftp.DataConnectionThrottleStreamIO

async is_dir(path: str | PurePosixPath) → bool¶

asyncio.coroutine()

Checks if path is dir.

Parameters:: path (str or pathlib.PurePosixPath) – path to check
Return type:: bool

async is_file(path: str | PurePosixPath) → bool¶

asyncio.coroutine()

Checks if path is file.

Parameters:: path (str or pathlib.PurePosixPath) – path to check
Return type:: bool

list(path: str | PurePosixPath = '', *, recursive: bool = False, raw_command: Literal['MLSD', 'LIST'] | None = None) → AbstractAsyncLister[tuple[PurePosixPath, BasicListInfo | UnixListInfo]]¶

asyncio.coroutine()

List all files and directories in “path”. If “path” is a file, then result will be empty

Parameters:

path (str or pathlib.PurePosixPath) – directory
recursive (bool) – list recursively
raw_command (str) – optional ftp command to use in place of fallback logic (must be one of “MLSD”, “LIST”)

Return type:

list or async for context

>>> # lazy list
>>> async for path, info in client.list():
...     # no interaction with client should be here(!)

>>> # eager list
>>> for path, info in (await client.list()):
...     # interaction with client allowed, since all paths are
...     # collected already

>>> stats = await client.list()

async login(user: str = 'anonymous', password: str = 'anon@', account: str = '') → None¶

asyncio.coroutine()

Server authentication.

Parameters:

user (str) – username
password (str) – password
account (str) – account (almost always blank)

Raises:

aioftp.StatusCodeError – if unknown code received

async make_directory(path: str | PurePosixPath, *, parents: bool = True) → None¶

asyncio.coroutine()

Make directory.

Parameters:

path (str or pathlib.PurePosixPath) – path to directory to create
parents (bool) – create parents if does not exists

static parse_directory_response(s: str) → PurePosixPath¶

Parsing directory server response.

Parameters:: s (str) – response line
Return type:: pathlib.PurePosixPath

static parse_epsv_response(s: str) → tuple[None, int]¶

Parsing EPSV (message (|||port|)) response.

Parameters:: s (str) – response line
Returns:: (ip, port)
Return type:: (None, int)

async parse_line() → tuple[Code, str]¶

asyncio.coroutine()

Parsing server response line.

Returns:

(code, line)

Return type:

(aioftp.Code, str)

Raises:

ConnectionResetError – if received data is empty (this means, that connection is closed)
asyncio.TimeoutError – if there where no data for timeout period

parse_list_line(b: bytes) → tuple[PurePosixPath, BasicListInfo | UnixListInfo]¶

Parse LIST response with both Microsoft Windows® parser and UNIX parser

Parameters:: b (bytes) – response line
Returns:: (path, info)
Return type:: (pathlib.PurePosixPath, dict)

parse_list_line_unix(b: bytes) → tuple[PurePosixPath, UnixListInfo]¶

Attempt to parse a LIST line (similar to unix ls utility).

Parameters:: b (bytes) – response line
Returns:: (path, info)
Return type:: (pathlib.PurePosixPath, dict)

parse_list_line_windows(b: bytes) → tuple[PurePosixPath, BasicListInfo]¶

Parsing Microsoft Windows dir output

Parameters:: b (bytes) – response line
Returns:: (path, info)
Return type:: (pathlib.PurePosixPath, dict)

classmethod parse_ls_date(s: str, *, now: datetime | None = None) → str¶

Parsing dates from the ls unix utility. For example, “Nov 18 1958”, “Jan 03 2018”, and “Nov 18 12:29”.

Parameters:: s (str) – ls date
Return type:: str

parse_mlsx_line(b: bytes | str) → tuple[PurePosixPath, BasicListInfo]¶

Parsing MLS(T|D) response.

Parameters:: b (bytes or str) – response line
Returns:: (path, info)
Return type:: (pathlib.PurePosixPath, dict)

static parse_pasv_response(s: str) → tuple[str, int]¶

Parsing PASV server response.

Parameters:: s (str) – response line
Returns:: (ip, port)
Return type:: (str, int)

async parse_response() → tuple[Code, list[str]]¶

asyncio.coroutine()

Parsing full server response (all lines).

Returns:: (code, lines)
Return type:: (aioftp.Code, list of str)
Raises:: aioftp.StatusCodeError – if received code does not matches all already received codes

static parse_unix_mode(s: str) → int¶

Parsing unix mode strings (“rwxr-x–t”) into hexadecimal notation.

Parameters:: s (str) – mode string
Return mode:
Return type:: int

async quit() → None¶

asyncio.coroutine()

Send “QUIT” and close connection.

async remove(path: str | PurePosixPath) → None¶

asyncio.coroutine()

High level remove method for removing path recursively (file or directory).

Parameters:: path (str or pathlib.PurePosixPath) – path to remove

async remove_directory(path: str | PurePosixPath) → None¶

asyncio.coroutine()

Low level remove method for removing empty directory.

Parameters:: path (str or pathlib.PurePosixPath) – empty directory to remove

async remove_file(path: str | PurePosixPath) → None¶

asyncio.coroutine()

Low level remove method for removing file.

Parameters:: path (str or pathlib.PurePosixPath) – file to remove

async rename(source: str | PurePosixPath, destination: str | PurePosixPath) → None¶

asyncio.coroutine()

Rename (move) file or directory.

Parameters:

source (str or pathlib.PurePosixPath) – path to rename
destination (str or pathlib.PurePosixPath) – path new name

async stat(path: str | PurePosixPath) → BasicListInfo | UnixListInfo¶

asyncio.coroutine()

Getting path stats.

Parameters:: path (str or pathlib.PurePosixPath) – path for getting info
Returns:: path info
Return type:: dict

property stream: ThrottleStreamIO¶

async upgrade_to_tls(sslcontext: SSLContext | None = None) → None¶

asyncio.coroutine()

Attempts to upgrade the connection to TLS (explicit TLS). Downgrading via the CCC or REIN commands is not supported. Both the command and data channels will be encrypted after using this command. You may call this command at any point during the connection, but not every FTP server will allow a connection upgrade after logging in.

Parameters:: sslcontext (ssl.SSLContext) – custom ssl context

async upload(source: str | Path, destination: str | PurePosixPath = '', *, write_into: bool = False, block_size: int = 8192) → None¶

asyncio.coroutine()

High level upload method for uploading files and directories recursively from file system.

Parameters:

source (str or pathlib.Path) – source path of file or directory on client side
destination (str or pathlib.PurePosixPath) – destination path of file or directory on server side
write_into (bool) – write source into destination (if you want upload file and change it name, as well with directories)
block_size (int) – block size for transaction

upload_stream(destination: str | PurePosixPath, *, offset: int = 0) → AsyncEnterableInstanceProtocol[DataConnectionThrottleStreamIO]¶

Create stream for write data to destination file.

Parameters:

destination (str or pathlib.PurePosixPath) – destination path of file on server side
offset (int) – byte offset for stream start position

Return type:

aioftp.DataConnectionThrottleStreamIO

class aioftp.DataConnectionThrottleStreamIO(client: BaseClient, reader: StreamReader, writer: StreamWriter, throttles: dict[str, StreamThrottle], *, timeout: float | int | None = None, read_timeout: float | int | None = None, write_timeout: float | int | None = None)¶

Bases: ThrottleStreamIO

Add finish method to aioftp.ThrottleStreamIO, which is specific for data connection. This requires client.

Parameters:

client (aioftp.BaseClient) – client class, which have aioftp.Client.command()
*args –
positional arguments passed to aioftp.ThrottleStreamIO
**kwargs –
keyword arguments passed to aioftp.ThrottleStreamIO

async finish(expected_codes: str | tuple[str, ...] = '2xx', wait_codes: str | tuple[str, ...] = '1xx') → None¶

asyncio.coroutine()

Close connection and wait for expected_codes response from server passing wait_codes.

Parameters:

expected_codes (tuple of str or str) – tuple of expected codes or expected code
wait_codes (tuple of str or str) – tuple of wait codes or wait code

class aioftp.Code¶

Bases: str

Representation of server status code.

matches(mask: str) → bool¶

Parameters:: mask (str) – Template for comparision. If mask symbol is not digit then it passes.

>>> Code("123").matches("1")
True
>>> Code("123").matches("1x3")
True

class aioftp.StatusCodeError(expected_codes: tuple[Code, ...] | Code, received_codes: tuple[Code, ...] | Code, info: list[str] | str)¶

Raised for unexpected or “bad” status codes.

Parameters:

expected_codes (tuple of aioftp.Code or aioftp.Code) – tuple of expected codes or expected code
received_codes (tuple of aioftp.Code or aioftp.Code) – tuple of received codes or received code
info (list of str) – list of lines with server response

>>> try:
...     # something with aioftp
... except StatusCodeError as e:
...     print(e.expected_codes, e.received_codes, e.info)
...     # analyze state

Exception members are tuples, even for one code.

Client API¶

Navigation

Related Topics