aioftp

You can define specific users with unique credentials, home directories, and granular permissions by passing a tuple of aioftp.User objects to the aioftp.Server constructor.

Permissions are defined using aioftp.Permission objects, which map virtual paths to readable and writable boolean flags.

>>> users = (
...     aioftp.User(
...         "Guido",
...         "secret_password",
...         home_path="/Guido",
...         permissions=(
...             aioftp.Permission("/", readable=False, writable=False),
...             aioftp.Permission("/Guido", readable=True, writable=True),
...         )
...     ),
...     aioftp.User(
...         home_path="/anon",
...         permissions=(
...             aioftp.Permission("/", readable=False, writable=False),
...             aioftp.Permission("/anon", readable=True),
...         )
...     ),
... )
>>> server = aioftp.Server(users)
>>> await server.start()

Use aioftp.Client.list to list paths. It can return a list of paths or be used as an asynchronous generator for lazy iteration.

• Eager version: await client.list() returns a list of (path, info) tuples.
• Lazy version: async for path, info in client.list(...) allows iterating over results as they arrive.

Warning: If you use the lazy async for version, you cannot interact with the server (e.g., issue other commands) until you have finished iterating through the loop. If you need to interact with the server while listing, use the eager version or create a second client connection.

# Eager list (returns a list)
paths = await client.list("/")

# Lazy list (asynchronous generator)
async for path, info in client.list("/", recursive=True):
    print(path)

# Mixing lazy list with interaction (requires two clients)
async for path, info in client1.list():
    await client2.download(path, path.name)

You can limit data transfer speeds at multiple levels:

• Global (Per Server): Set via read_speed_limit and write_speed_limit in the aioftp.Server constructor.
• Per Connection: Set via read_speed_limit_per_connection and write_speed_limit_per_connection in the aioftp.Server constructor.
• Per User: Set via read_speed_limit, write_speed_limit, etc., in the aioftp.User constructor.
• Per User Connection: Set via read_speed_limit_per_connection and write_speed_limit_per_connection in the aioftp.User constructor.

# Global server throttling
>>> server = aioftp.Server(
...     read_speed_limit=1024 * 1024,
...     write_speed_limit=1024 * 1024,
...     read_speed_limit_per_connection=100 * 1024,
...     write_speed_limit_per_connection=100 * 1024
... )

# User-specific throttling
>>> users = (
...     aioftp.User(
...         read_speed_limit=1024 * 1024,
...         write_speed_limit=1024 * 1024,
...         read_speed_limit_per_connection=100 * 1024,
...         write_speed_limit_per_connection=100 * 1024
...     ),
... )
>>> server = aioftp.Server(users)

For commands that require a passive connection (like data transfers), you should use a worker to ensure the server can continue receiving commands while data is being transferred.

To implement this:

1. Define an internal async function (the worker).
2. Decorate it with @aioftp.server.worker and appropriate ConnectionConditions.
3. In the main command method, create a task for the worker using connection.loop.create_task(coro).
4. Add the task to connection.extra_workers so the dispatcher manages it.
5. Send an initial response (e.g., 150) to the client immediately.

class MyServer(aioftp.Server):
    @aioftp.ConnectionConditions(
        aioftp.ConnectionConditions.login_required,
        aioftp.ConnectionConditions.passive_server_started)
    async def coll(self, connection, rest):

        @aioftp.ConnectionConditions(
            aioftp.ConnectionConditions.data_connection_made,
            wait=True,
            fail_code="425",
            fail_info="Can't open data connection")
        @aioftp.server.worker
        async def coll_worker(self, connection, rest):
            stream = connection.data_connection
            del connection.data_connection
            async with stream:
                for i in range(count):
                    binary = i.to_bytes(8, "big")
                    await stream.write(binary)
            connection.response("200", "coll transfer done")
            return True

        count = int(rest)
        coro = coll_worker(self, connection, rest)
        task = connection.loop.create_task(coro)
        connection.extra_workers.add(task)
        connection.response("150", "coll transfer started")
        return True

The aioftp.Client provides a high-level asynchronous API for interacting with FTP servers. You can use aioftp.Client.context() to manage the connection lifecycle automatically.

Note on Directory Listings: For commands involving stats or listings, aioftp prefers MLSx-family commands (like MLSD) because they provide structured, machine-readable data. If the server does not support these, aioftp falls back to the LIST command, which lacks a standard format and may be harder to parse. You can provide a custom LIST parser if needed.

import asyncio
import aioftp


async def get_mp3(host, port, login, password):
    async with aioftp.Client.context(host, port, login, password) as client:
        for path, info in (await client.list(recursive=True)):
            if info["type"] == "file" and path.suffix == ".mp3":
                await client.download(path)


async def main():
    tasks = [
        asyncio.create_task(get_mp3("server1.com", 21, "login", "password")),
        asyncio.create_task(get_mp3("server2.com", 21, "login", "password")),
        asyncio.create_task(get_mp3("server3.com", 21, "login", "password")),
    ]
    await asyncio.wait(tasks)

asyncio.run(main())