# Server API

class aioftp.Server(users: ~collections.abc.Sequence[~aioftp.server.User] | ~aioftp.server.AbstractUserManager | None = None, *, block_size: int = 8192, socket_timeout: float | int | None = None, idle_timeout: float | int | None = None, wait_future_timeout: float | int | None = 1, path_timeout: float | int | None = None, path_io_factory: type[~aioftp.pathio.AbstractPathIO[~pathlib._local.Path]] = <class 'aioftp.pathio.PathIO'>, maximum_connections: int | None = None, read_speed_limit: int | None = None, write_speed_limit: int | None = None, read_speed_limit_per_connection: int | None = None, write_speed_limit_per_connection: int | None = None, ipv4_pasv_forced_response_address: str | None = None, data_ports: ~collections.abc.Iterable[int] | None = None, encoding: str = 'utf-8', ssl: ~ssl.SSLContext | None = None)

  Bases: `object`

  FTP server.

  Parameters:

    - users (`tuple` or `list` of `aioftp.User` or instance of `aioftp.AbstractUserManager` subclass) – list of users or user manager object

    - block_size (`int`) – bytes count for socket read operations

    - socket_timeout (`float`, `int` or `None`) – timeout for socket read and write operations

    - idle_timeout (`float`, `int` or `None`) – timeout for socket read operations, another words: how long user can keep silence without sending commands

    - wait_future_timeout (`float`, `int` or `None`) – wait for data connection to establish

    - 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»

    - maximum_connections (`int`) – Maximum command connections per server

    - read_speed_limit (`int` or `None`) – server read speed limit in bytes per second

    - write_speed_limit (`int` or `None`) – server write speed limit in bytes per second

    - read_speed_limit_per_connection (`int` or `None`) – server read speed limit per connection in bytes per second

    - write_speed_limit_per_connection (`int` or `None`) – server write speed limit per connection in bytes per second

    - ipv4_pasv_forced_response_address (`str` or `None`) – external IPv4 address for passive connections

    - data_ports (`collections.Iterable` or `None`) – port numbers that are available for passive connections

    - encoding (`str`) – encoding to use for convertion strings to bytes

    - ssl (`ssl.SSLContext`) – can be set to an `ssl.SSLContext` instance to enable TLS over the accepted connections. Please look `asyncio.loop.create_server()` docs.

  async close() → None

    `asyncio.coroutine()`

    Shutdown the server and close all connections.

  async run(host: str | None = None, port: int = 0, **kwargs: Unpack[AsyncIOStartServerKwargs]) → None

    `asyncio.coroutine()`

    Single entrypoint to start, serve and close.

    Parameters:

      - host (`str`) – ip address to bind for listening.

      - port (`int`) – port number to bind for listening.

      - kwargs – keyword arguments, they passed to `asyncio.start_server()`

  async serve_forever() → None

    `asyncio.coroutine()`

    Proxy to `asyncio.Server` serve_forever method.

  async start(host: str | None = None, port: int = 0, **kwargs: Unpack[AsyncIOStartServerKwargs]) → None

    `asyncio.coroutine()`

    Start server.

    Parameters:

      - host (`str`) – ip address to bind for listening.

      - port (`int`) – port number to bind for listening.

      - kwargs – keyword arguments, they passed to `asyncio.start_server()`

aioftp.server.worker(f: Callable[[Server, Connection, Any], Awaitable[WorkerR]]) → Callable[[Server, Connection, Any], Awaitable[WorkerR]]

  Decorator. Abortable worker. If wrapped task will be cancelled by dispatcher, decorator will send ftp codes of successful interrupt.

      >>> @worker
      ... async def worker(self, connection, rest):
      ...     ...

class aioftp.User(login: str | None = None, password: str | None = None, *, base_path: str | Path = PosixPath('.'), home_path: str | PurePosixPath = PurePosixPath('/'), permissions: Sequence[Permission] | None = None, maximum_connections: int | None = None, read_speed_limit: int | None = None, write_speed_limit: int | None = None, read_speed_limit_per_connection: int | None = None, write_speed_limit_per_connection: int | None = None)

  User description.

  Parameters:

    - login (`str`) – user login

    - password (`str`) – user password

    - base_path (`str` or `pathlib.Path`) – real user path for file io operations

    - home_path (`str` or `pathlib.PurePosixPath`) – virtual user path for client representation (must be absolute)

    - permissions (`tuple` or `list` of `aioftp.Permission`) – list of path permissions

    - maximum_connections (`int`) – Maximum connections per user

    - read_speed_limit (`int` or `None`) – read speed limit per user in bytes per second

    - write_speed_limit (`int` or `None`) – write speed limit per user in bytes per second

    - read_speed_limit_per_connection (`int` or `None`) – read speed limit per user connection in bytes per second

    - write_speed_limit_per_connection (`int` or `None`) – write speed limit per user connection in bytes per second

  async get_permissions(path: str | PurePosixPath) → Permission

    Return nearest parent permission for path.

    Parameters:

      path (`str` or `pathlib.PurePosixPath`) – path which permission you want to know

    Return type:

      `aioftp.Permission`

class aioftp.Permission(path: str | PurePosixPath = '/', *, readable: bool = True, writable: bool = True)

  Path permission

  Parameters:

    - path (`str` or `pathlib.PurePosixPath`) – path

    - readable (`bool`) – is readable

    - writable (`bool`) – is writable

class aioftp.AbstractUserManager(*, timeout: float | int | None = None)

  Abstract user manager.

  Parameters:

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

  abstract async authenticate(user: User, password: str) → bool

    `asyncio.coroutine()`

    Check if user can be authenticated with provided password

    Parameters:

      - user (`aioftp.User`) – user

      - password (`str`) – password

    Return type:

      `bool`

  abstract async get_user(login: str) → tuple[GetUserResponse, User | None, str]

    `asyncio.coroutine()`

    Get user and response for USER call

    Parameters:

      login (`str`) – user’s login

  async notify_logout(user: User) → None

    `asyncio.coroutine()`

    Called when user connection is closed if user was initiated

    Parameters:

      user (`aioftp.User`) – user

class aioftp.server.MemoryUserManager(users: Sequence[User] | None, *, timeout: float | int | None = None)

  A built-in user manager that keeps predefined set of users in memory.

  Parameters:

    users (`list`, `tuple`, etc. of `aioftp.User`) – container of users

  async authenticate(user: User, password: str) → bool

    `asyncio.coroutine()`

    Check if user can be authenticated with provided password

    Parameters:

      - user (`aioftp.User`) – user

      - password (`str`) – password

    Return type:

      `bool`

  async get_user(login: str) → tuple[GetUserResponse, User | None, str]

    `asyncio.coroutine()`

    Get user and response for USER call

    Parameters:

      login (`str`) – user’s login

  async notify_logout(user: User) → None

    `asyncio.coroutine()`

    Called when user connection is closed if user was initiated

    Parameters:

      user (`aioftp.User`) – user

class aioftp.Connection(**kwargs: Any)

  Bases: `defaultdict`[`str`, `Future`[`Any`]]

  Connection state container for transparent work with futures for async wait

  Parameters:

    kwargs – initialization parameters

  Container based on `collections.defaultdict`, which holds `asyncio.Future` as default factory. There is two layers of abstraction:

  - Low level based on simple dictionary keys to attributes mapping and

      available at Connection.future.

  - High level based on futures result and dictionary keys to attributes

      mapping and available at Connection.

  To clarify, here is groups of equal expressions

      >>> connection.future.foo
      >>> connection["foo"]

      >>> connection.foo
      >>> connection["foo"].result()

      >>> del connection.future.foo
      >>> del connection.foo
      >>> del connection["foo"]

class aioftp.AvailableConnections(value: int | None = None)

  Semaphore-like object. Have no blocks, only raises ValueError on bounds crossing. If value is `None` have no limits (bounds checks).

  Parameters:

    value (`int` or `None`)

  acquire() → None

    Acquire, decrementing the internal counter by one.

  locked() → bool

    Returns True if semaphore-like can not be acquired.

    Return type:

      `bool`

  release() → None

    Release, incrementing the internal counter by one.

class aioftp.ConnectionConditions(*fields: tuple[str, str], wait: bool = False, fail_code: str = '503', fail_info: str | None = None)

  Decorator for checking connection keys for existence or wait for them. Available options:

  Parameters:

    - fields –

      - ConnectionConditions.user_required — required “user” key, user already identified

      - ConnectionConditions.login_required — required “logged” key, user already logged in.

      - ConnectionConditions.passive_server_started — required “passive_server” key, user already send PASV and server awaits incomming connection

      - ConnectionConditions.data_connection_made — required “data_connection” key, user already connected to passive connection

      - ConnectionConditions.rename_from_required — required “rename_from” key, user already tell filename for rename

    - wait (`bool`) – Indicates if should wait for parameters for connection.wait_future_timeout

    - fail_code (`str`) – return code if failure

    - fail_info (`str`) – return information string if failure. If `None`, then use default string

      >>> @ConnectionConditions(
      ...     ConnectionConditions.login_required,
      ...     ConnectionConditions.passive_server_started,
      ...     ConnectionConditions.data_connection_made,
      ...     wait=True)
      ... def foo(self, connection, rest):
      ...     ...

class aioftp.PathConditions(*conditions: tuple[str, bool, str])

  Decorator for checking paths. Available options:

  - path_must_exists

  - path_must_not_exists

  - path_must_be_dir

  - path_must_be_file

      >>> @PathConditions(
      ...     PathConditions.path_must_exists,
      ...     PathConditions.path_must_be_dir)
      ... def foo(self, connection, path):
      ...     ...

class aioftp.PathPermissions(*permissions: str)

  Decorator for checking path permissions. There is two permissions right now:

  - PathPermissions.readable

  - PathPermissions.writable

  Decorator will check the permissions and return proper code and information to client if permission denied

      >>> @PathPermissions(
      ...     PathPermissions.readable,
      ...     PathPermissions.writable)
      ... def foo(self, connection, path):
      ...     ...