# 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._local.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.

  async command(command: str | None = None, expected_codes: str | tuple[str, ...] = (), wait_codes: str | tuple[str, ...] = (), censor_after: int | None = None) → tuple[Code, list[str]] | None

    `asyncio.coroutine()`

    Basic command logic.

    1. Send command if not omitted.

    2. Yield response until no wait code matches.

    3. 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'] | Literal['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.