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.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:
class aioftp.AbstractUserManager(*, timeout: float | int | None = None)

Abstract user manager.

Parameters:

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

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

asyncio.coroutine()

Check if user can be authenticated with provided password

Parameters:
Return type:

bool

abstractmethod 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:
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):
...     ...