### Install aioftp

Source: https://github.com/aio-libs/aioftp/blob/master/docs/index.md

Install the aioftp library using pip. For socks proxy support, use `pip install aioftp[socks]`.

```shell
pip install aioftp
```

--------------------------------

### Start aioftp Server

Source: https://github.com/aio-libs/aioftp/blob/master/docs/server_tutorial.md

Instantiate and start the aioftp server with default settings. This allows anonymous login with read/write permissions on the current directory.

```python
>>> server = aioftp.Server()
>>> await server.start()
```

--------------------------------

### Start aioftp Server from Command Line

Source: https://context7.com/aio-libs/aioftp/llms.txt

Launch an aioftp FTP server directly from the command line for quick testing or file sharing. Options include specifying the host, port, and serving the current directory anonymously.

```bash
# Start anonymous FTP server on port 8021 serving current directory
python -m aioftp

# Specify host and port
python -m aioftp --host 0.0.0.0 --port 2121

# Get help for all options
python -m aioftp --help

```

--------------------------------

### Configure Client with SOCKS Proxy

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

Initialize the aioftp.Client to use a SOCKS proxy for its connections. Ensure aioftp is installed with the 'socks' extra or siosocks is installed separately.

```python
client = aioftp.Client(
    socks_host="localhost",
    socks_port=9050,
    socks_version=5,
)
```

--------------------------------

### Test Custom aioftp Server and Client Interaction

Source: https://github.com/aio-libs/aioftp/blob/master/docs/developer_tutorial.md

Example test function to start a custom aioftp server, connect a client, log in, execute a custom 'collect' command, and then close both server and client. Basic logging is configured.

```python
async def test():
    server = MyServer()
    client = MyClient()
    await server.start("127.0.0.1", 8021)
    await client.connect("127.0.0.1", 8021)
    await client.login()
    collected = await client.collect(20)
    print(collected)
    await client.quit()
    await server.close()


if __name__ == "__main__":
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s [%(name)s] %(message)s",
        datefmt="[%H:%M:%S]:",
    )
    loop = asyncio.get_event_loop()
    loop.run_until_complete(test())
    print("done")
```

--------------------------------

### Implement Custom Path IO Layer in Python

Source: https://context7.com/aio-libs/aioftp/llms.txt

Create custom file system backends, such as in-memory storage or cloud storage, by implementing a custom path IO layer. This example demonstrates using an in-memory file system for testing.

```python
import asyncio
import aioftp
from aioftp import pathio

async def main():
    # Use in-memory file system (useful for testing)
    server = aioftp.Server(
        path_io_factory=pathio.MemoryPathIO,
    )

    # For client with memory path IO
    client = aioftp.Client(
        path_io_factory=pathio.MemoryPathIO,
    )

    # Available path IO implementations:
    # - pathio.PathIO: Default blocking filesystem operations
    # - pathio.AsyncPathIO: Non-blocking via run_in_executor (slower)
    # - pathio.MemoryPathIO: In-memory filesystem (for testing)

    await server.run("0.0.0.0", 2121)

asyncio.run(main())

```

--------------------------------

### Configure Bandwidth Control with Throttle in Python

Source: https://context7.com/aio-libs/aioftp/llms.txt

Utilize the Throttle and StreamThrottle classes for fine-grained bandwidth control on data streams. This example shows creating individual throttles, stream throttles, modifying limits, and cloning throttles.

```python
import asyncio
import aioftp
from aioftp.common import Throttle, StreamThrottle

# Create individual throttles
read_throttle = Throttle(limit=1024 * 100)  # 100 KB/s
write_throttle = Throttle(limit=1024 * 50)  # 50 KB/s

# Create stream throttle pair
stream_throttle = StreamThrottle.from_limits(
    read_speed_limit=1024 * 1024,    # 1 MB/s read
    write_speed_limit=512 * 1024,    # 512 KB/s write
)

# Modify throttle at runtime
read_throttle.limit = 1024 * 200  # Increase to 200 KB/s
read_throttle.limit = None        # Remove limit (unlimited)

# Clone throttle (without accumulated state)
new_throttle = stream_throttle.clone()

```

--------------------------------

### Start aioftp Server

Source: https://context7.com/aio-libs/aioftp/llms.txt

Initialize and run an aioftp server. The server can be configured for anonymous access or with custom user authentication. The `serve_forever` method runs the server indefinitely, while `run` is a convenience method that also blocks.

```python
import asyncio
import aioftp

async def main():
    # Simple anonymous server serving current directory
    server = aioftp.Server()
    await server.start("0.0.0.0", 2121)
    print(f"Server running on {server.address}")

    try:
        await server.serve_forever()
    finally:
        await server.close()

# Or use the combined run() method
async def main_simple():
    server = aioftp.Server()
    await server.run("0.0.0.0", 2121)  # Blocks until interrupted

asyncio.run(main())
```

--------------------------------

### Implement Custom User Manager in Python

Source: https://context7.com/aio-libs/aioftp/llms.txt

Extend aioftp.AbstractUserManager to fetch users from a database or external service for dynamic authentication. This example simulates a user database and handles user retrieval and authentication.

```python
import asyncio
import aioftp
from pathlib import Path

class DatabaseUserManager(aioftp.AbstractUserManager):
    """Custom user manager that could fetch users from a database."""

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        # Simulated user database
        self.users_db = {
            "john": {"password": "secret123", "quota": 1024*1024*100},
            "jane": {"password": "pass456", "quota": 1024*1024*500},
        }
        self.connected_users = {}

    async def get_user(self, login):
        if login in self.users_db:
            user = aioftp.User(
                login=login,
                password=self.users_db[login]["password"],
                base_path=Path(f"/srv/ftp/users/{login}"),
                home_path=f"/{login}",
                permissions=[
                    aioftp.Permission(f"/{login}", readable=True, writable=True),
                ],
            )
            return (self.GetUserResponse.PASSWORD_REQUIRED, user, "password required")
        return (self.GetUserResponse.ERROR, None, "user not found")

    async def authenticate(self, user, password):
        if user.login in self.users_db:
            return self.users_db[user.login]["password"] == password
        return False

    async def notify_logout(self, user):
        print(f"User {user.login} logged out")

async def main():
    user_manager = DatabaseUserManager()
    server = aioftp.Server(user_manager)
    await server.run("0.0.0.0", 2121)

asyncio.run(main())

```

--------------------------------

### Get Current Directory

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

Retrieves the current working directory on the FTP server using `aioftp.Client.get_current_directory()`.

```python
>>> await client.get_current_directory()
PosixPath('/public_html')
```

--------------------------------

### Get Directory Statistics

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

Retrieve detailed statistics for a directory on the FTP server, including type and size.

```python
await client.stat(".git")
```

--------------------------------

### Get Path Statistics

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

Retrieve detailed statistics for a given path on the FTP server, including size, creation time, and type.

```python
await client.stat("tmp2.py")
```

--------------------------------

### Connect aioftp Client via SOCKS Proxy

Source: https://context7.com/aio-libs/aioftp/llms.txt

Establish an FTP connection through a SOCKS proxy, such as Tor, for enhanced privacy or tunneling. Ensure `aioftp[socks]` is installed.

```python
import asyncio
import aioftp

async def main():
    # Connect through SOCKS5 proxy (e.g., Tor)
    client = aioftp.Client(
        socks_host="localhost",
        socks_port=9050,
        socks_version=5,
    )

    await client.connect("ftp.example.com")
    await client.login("anonymous", "anonymous@")

    files = await client.list()
    for path, info in files:
        print(path)

    await client.quit()

asyncio.run(main())
```

--------------------------------

### aioftp Client Directory and File Operations

Source: https://context7.com/aio-libs/aioftp/llms.txt

Perform common file system operations on the remote server, including changing directories, creating directories, getting file stats, checking path types, renaming, and removing files or directories.

```python
import asyncio
import aioftp

async def main():
    async with aioftp.Client.context("ftp.example.com", user="user", password="pass") as client:
        # Get current working directory
        cwd = await client.get_current_directory()
        print(f"Current directory: {cwd}")

        # Change directory
        await client.change_directory("/pub/data")

        # Go up one level
        await client.change_directory("..")

        # Create directory (parents created automatically)
        await client.make_directory("/new/nested/directory")

        # Get file/directory stats
        stats = await client.stat("/pub/file.txt")
        print(f"Type: {stats['type']}, Size: {stats.get('size')}, Modified: {stats.get('modify')}")

        # Check path type
        if await client.is_file("/pub/file.txt"):
            print("It's a file")
        if await client.is_dir("/pub"):
            print("It's a directory")
        if await client.exists("/pub/maybe.txt"):
            print("Path exists")

        # Rename/move file or directory
        await client.rename("/old/path.txt", "/new/path.txt")

        # Remove file or directory (recursive)
        await client.remove("/path/to/delete")

asyncio.run(main())
```

--------------------------------

### Connect and Login to FTP Server

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

Instantiate an aioftp client and connect to a specified host, then log in with provided credentials.

```python
client = aioftp.Client()
await client.connect("ftp.server.com")
await client.login("user", "pass")
```

--------------------------------

### Client.context - Async Context Manager for FTP Connections

Source: https://context7.com/aio-libs/aioftp/llms.txt

The `Client.context` classmethod provides an async context manager for automatically handling FTP connection, login, and cleanup. This is the recommended approach for most use cases.

```APIDOC
## Client.context - Context Manager for FTP Connections

### Description
Provides an async context manager that automatically handles connection, login, and cleanup for FTP operations. This is the recommended way to interact with FTP servers.

### Method
`aioftp.Client.context(host, port=21, user='anonymous', password='', *, use_ssl=False, cert_reqs=None, session=None, **connection_options)`

### Parameters
#### Connection Options (passed to `aioftp.Client` constructor)
- **host** (str) - Required - The FTP server hostname or IP address.
- **port** (int) - Optional - The FTP server port. Defaults to 21.
- **user** (str) - Optional - The username for authentication. Defaults to 'anonymous'.
- **password** (str) - Optional - The password for authentication. Defaults to an empty string.
- **use_ssl** (bool) - Optional - Whether to use SSL/TLS for the connection. Defaults to False.
- **cert_reqs** (int) - Optional - SSL certificate requirements. See `ssl.CERT_NONE`, `ssl.CERT_OPTIONAL`, `ssl.CERT_REQUIRED`.
- **session** (aiohttp.ClientSession) - Optional - An existing aiohttp client session to use.
- **socket_timeout** (float) - Optional - Timeout for socket operations in seconds.
- **read_speed_limit** (int) - Optional - Download speed limit in bytes per second.
- **write_speed_limit** (int) - Optional - Upload speed limit in bytes per second.

### Request Example
```python
import asyncio
import aioftp

async def main():
    # Basic anonymous connection
    async with aioftp.Client.context("ftp.example.com") as client:
        # List files in root directory
        for path, info in await client.list("/"):
            print(f"{path}: {info['type']}")

    # Authenticated connection with options
    async with aioftp.Client.context(
        "ftp.example.com",
        port=21,
        user="username",
        password="secret",
        socket_timeout=30,
        read_speed_limit=1024 * 1024,  # 1 MB/s download limit
    ) as client:
        current_dir = await client.get_current_directory()
        print(f"Current directory: {current_dir}")

asyncio.run(main())
```

### Response
- **client** (aioftp.Client) - An active `aioftp.Client` instance within the `async with` block.
```

--------------------------------

### Configure Users and Permissions

Source: https://github.com/aio-libs/aioftp/blob/master/docs/server_tutorial.md

Create a custom set of users with specific home paths and permissions. Users can be granted read/write access to designated directories while restricting access to others.

```python
>>> 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()
```

--------------------------------

### FTP Directory Listing with Client.list

Source: https://context7.com/aio-libs/aioftp/llms.txt

Demonstrates both eager and lazy directory listing modes. Eager mode loads all entries into memory and allows client interaction during iteration. Lazy mode is memory-efficient for large directories but prohibits client interaction within the loop.

```python
import asyncio
import aioftp

async def main():
    async with aioftp.Client.context("ftp.example.com") as client:
        # Eager listing - loads all entries into memory
        # Use this when you need to interact with client during iteration
        for path, info in await client.list("/pub"):
            if info["type"] == "file":
                print(f"File: {path}, Size: {info.get('size', 'unknown')}")
                # Safe to call client methods here
                await client.stat(path)

        # Lazy listing - memory efficient for large directories
        # WARNING: Do not interact with client inside async for block
        async for path, info in client.list("/pub", recursive=True):
            if path.suffix == ".txt":
                print(f"Found text file: {path}")

        # Force specific LIST command (MLSD or LIST)
        entries = await client.list("/pub", raw_command="MLSD")

asyncio.run(main())
```

--------------------------------

### Create Directory

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

Creates a new directory on the FTP server using `aioftp.Client.make_directory()`.

```python
>>> await client.make_directory("folder2")
>>> await client.change_directory("folder2")
>>> await client.get_current_directory()
PosixPath('/public_html/folder2')
```

--------------------------------

### Eager List and Download

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

Perform an eager listing of the current directory and then download each listed item. This avoids blocking server interaction during iteration.

```python
for path, info in (await client.list()):
    await client.download(path, path.name)
```

--------------------------------

### Run aioftp Server

Source: https://github.com/aio-libs/aioftp/blob/master/README.rst

Sets up and runs an aioftp server. Requires user configuration and a path IO factory.

```python
import asyncio
import aioftp


async def main():
    server = aioftp.Server([user], path_io_factory=path_io_factory)
    await server.run()

asyncio.run(main())
```

--------------------------------

### Configure Client with MemoryPathIO

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

Instantiate aioftp.Client using MemoryPathIO for in-memory file system operations. This is useful for testing or scenarios where a real file system is not needed.

```python
client = aioftp.Client(path_io_factory=pathio.MemoryPathIO)
```

--------------------------------

### Async FTP Connection using Client.context

Source: https://context7.com/aio-libs/aioftp/llms.txt

Use Client.context as an async context manager for automatic connection, login, and cleanup. Supports anonymous and authenticated connections with various options like port, user, password, and socket timeout.

```python
import asyncio
import aioftp

async def main():
    # Basic anonymous connection
    async with aioftp.Client.context("ftp.example.com") as client:
        # List files in root directory
        for path, info in await client.list("/"):
            print(f"{path}: {info['type']}")

    # Authenticated connection with options
    async with aioftp.Client.context(
        "ftp.example.com",
        port=21,
        user="username",
        password="secret",
        socket_timeout=30,
        read_speed_limit=1024 * 1024,  # 1 MB/s download limit
    ) as client:
        current_dir = await client.get_current_directory()
        print(f"Current directory: {current_dir}")

asyncio.run(main())
```

--------------------------------

### Download Stream with Abort

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

Demonstrates how to download a file using `aioftp.Client.download_stream()` and abort the transfer if necessary. Do not use `async with` if aborting.

```python
>>> stream = await client.download_stream("tmp.py")
... async for block in stream.iter_by_block():
...     # do something with data
...     if something_not_interesting:
...         await client.abort()
...         stream.close()
...         break
... else:
...     await stream.finish()
```

--------------------------------

### Extend aioftp Server with Custom Commands

Source: https://context7.com/aio-libs/aioftp/llms.txt

Add custom FTP commands like NOOP and SITE by subclassing aioftp.Server and registering new command handlers. Supports connection condition decorators.

```python
import asyncio
import aioftp
from aioftp import ConnectionConditions, PathConditions, PathPermissions

class CustomServer(aioftp.Server):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        # Register custom command
        self.commands_mapping["noop"] = self.noop
        self.commands_mapping["site"] = self.site_command

    async def noop(self, connection, rest):
        """Handle NOOP command - do nothing, just respond OK."""
        connection.response("200", "OK")
        return True

    @ConnectionConditions(ConnectionConditions.login_required)
    async def site_command(self, connection, rest):
        """Handle SITE command for server-specific operations."""
        if rest.upper() == "HELP":
            connection.response("214", "SITE commands: HELP, TIME")
        elif rest.upper() == "TIME":
            import time
            connection.response("200", f"Server time: {time.ctime()}")
        else:
            connection.response("502", f"Unknown SITE command: {rest}")
        return True

async def main():
    server = CustomServer()
    await server.run("0.0.0.0", 2121)

asyncio.run(main())

```

--------------------------------

### Client.list - Directory Listing

Source: https://context7.com/aio-libs/aioftp/llms.txt

Fetches directory listings from the FTP server. Supports both eager (loads all entries) and lazy (iterates entries) modes, as well as forcing specific LIST commands.

```APIDOC
## Client.list - Directory Listing

### Description
Fetches directory listings from the FTP server. Supports both eager (await) and lazy (async for) iteration modes. Allows forcing specific LIST commands like MLSD or LIST.

### Method
`client.list(path='.', *, recursive=False, raw_command=None)`

### Parameters
- **path** (str) - Optional - The directory path to list. Defaults to the current directory ('.').
- **recursive** (bool) - Optional - If True, lists directories recursively. Defaults to False.
- **raw_command** (str) - Optional - Forces the use of a specific command (e.g., 'MLSD', 'LIST'). If None, the library chooses the best available command.

### Request Example
```python
import asyncio
import aioftp

async def main():
    async with aioftp.Client.context("ftp.example.com") as client:
        # Eager listing - loads all entries into memory
        # Use this when you need to interact with client during iteration
        for path, info in await client.list("/pub"):
            if info["type"] == "file":
                print(f"File: {path}, Size: {info.get('size', 'unknown')}")
                # Safe to call client methods here
                await client.stat(path)

        # Lazy listing - memory efficient for large directories
        # WARNING: Do not interact with client inside async for block
        async for path, info in client.list("/pub", recursive=True):
            if path.suffix == ".txt":
                print(f"Found text file: {path}")

        # Force specific LIST command (MLSD or LIST)
        entries = await client.list("/pub", raw_command="MLSD")

asyncio.run(main())
```

### Response
- **list**: Returns an iterable (list for eager, async iterator for lazy) of tuples. Each tuple contains:
    - **path** (pathlib.Path) - The path of the file or directory.
    - **info** (dict) - A dictionary containing file/directory information (e.g., 'type', 'size', 'modify').
```

--------------------------------

### Configure Server Read/Write Speed Limits

Source: https://github.com/aio-libs/aioftp/blob/master/docs/server_tutorial.md

Set global read and write speed limits for the server, as well as per-connection limits.

```python
>>> 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
... )
```

--------------------------------

### Configure aioftp Server with TLS/SSL

Source: https://context7.com/aio-libs/aioftp/llms.txt

Set up the aioftp server to use encrypted connections by providing an SSL context loaded with certificate and key files. Uses the standard FTPS port.

```python
import asyncio
import ssl
import aioftp

async def main():
    # Create SSL context
    ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
    ssl_context.load_cert_chain("server.crt", "server.key")

    server = aioftp.Server(
        ssl=ssl_context,
        socket_timeout=30,
        idle_timeout=300,
    )

    await server.run("0.0.0.0", 990)  # Standard FTPS port

asyncio.run(main())

```

--------------------------------

### Client.connect and Client.login - Manual Connection

Source: https://context7.com/aio-libs/aioftp/llms.txt

Allows for manual control over the FTP connection process, including explicit steps for connecting, upgrading to TLS, and logging in. Useful for scenarios requiring operations between connection and authentication.

```APIDOC
## Client.connect and Client.login - Manual Connection

### Description
Provides manual control over the FTP connection process. This is useful when you need to perform operations between connecting and logging in, such as upgrading to TLS.

### Method
- `client.connect(host, port=21, *, use_ssl=False, **connection_options)`
- `client.login(user, password, *, acct='')`
- `client.quit()`

### Parameters
#### `connect` method:
- **host** (str) - Required - The FTP server hostname or IP address.
- **port** (int) - Optional - The FTP server port. Defaults to 21.
- **use_ssl** (bool) - Optional - Whether to use SSL/TLS for the connection. Defaults to False.
- **connection_options** - Additional keyword arguments passed to the `aioftp.Client` constructor (e.g., `socket_timeout`).

#### `login` method:
- **user** (str) - Required - The username for authentication.
- **password** (str) - Required - The password for authentication.
- **acct** (str) - Optional - The account name for authentication.

### Request Example
```python
import asyncio
import aioftp

async def main():
    client = aioftp.Client(socket_timeout=10)

    try:
        # Connect to server
        await client.connect("ftp.example.com", port=21)

        # Optional: Upgrade to TLS before login
        await client.upgrade_to_tls()

        # Login with credentials
        await client.login("username", "password")

        # Perform operations
        files = await client.list()
        for path, info in files:
            print(f"{path.name}: {info.get('size', 'N/A')} bytes")

    finally:
        # Always close the connection
        await client.quit()

asyncio.run(main())
```

### Response
- **connect**: None
- **login**: None
- **quit**: None
```

--------------------------------

### Manual FTP Connection with Client.connect and Client.login

Source: https://context7.com/aio-libs/aioftp/llms.txt

Manually connect and login to an FTP server for more control, such as upgrading to TLS before authentication. Ensure the connection is always closed using a finally block.

```python
import asyncio
import aioftp

async def main():
    client = aioftp.Client(socket_timeout=10)

    try:
        # Connect to server
        await client.connect("ftp.example.com", port=21)

        # Optional: Upgrade to TLS before login
        await client.upgrade_to_tls()

        # Login with credentials
        await client.login("username", "password")

        # Perform operations
        files = await client.list()
        for path, info in files:
            print(f"{path.name}: {info.get('size', 'N/A')} bytes")

    finally:
        # Always close the connection
        await client.quit()

asyncio.run(main())
```

--------------------------------

### Connect and Upgrade to TLS

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

Connect to an FTP server and then upgrade the connection to a secure TLS session. This is essential for sensitive data transfer.

```python
client = aioftp.Client()
await client.connect("ftp.server.com")
await client.upgrade_to_tls()
await client.login("user", "pass")
```

--------------------------------

### aioftp Command-Line Help

Source: https://github.com/aio-libs/aioftp/blob/master/README.rst

Displays the help message for the aioftp command-line interface, useful for understanding available options.

```shell
python -m aioftp --help
```

--------------------------------

### Use MemoryPathIO for File System Abstraction

Source: https://github.com/aio-libs/aioftp/blob/master/docs/server_tutorial.md

Configure the server to use an in-memory file system implementation. This is useful for testing or scenarios where a real file system is not required.

```python
>>> server = aioftp.Server(path_io_factory=aioftp.MemoryPathIO)
>>> await server.start()
```

--------------------------------

### Configure User Read/Write Speed Limits

Source: https://github.com/aio-libs/aioftp/blob/master/docs/server_tutorial.md

Define read and write speed limits for individual users, including per-connection limits.

```python
>>> 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)
```

--------------------------------

### Download Stream with Offset

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

Initiates a download from a specific byte offset using the `offset` argument in `aioftp.Client.download_stream()`. Useful for resuming interrupted downloads.

```python
>>> async with client.download_stream("tmp.py", offset=256) as stream:
...     async for block in stream.iter_by_block():
...         # do something with data
```

--------------------------------

### Implement NOOP Command for aioftp Server

Source: https://github.com/aio-libs/aioftp/blob/master/docs/developer_tutorial.md

Extend aioftp.Server to add a custom NOOP command. The method receives connection state and the rest of the command string. It should return True on success.

```python
class MyServer(aioftp.Server):

    async def noop(self, connection, rest):
        connection.response("200", "boring")
        return True
```

--------------------------------

### Set Read Speed Limit

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

Initializes an `aioftp.Client` with a specific read speed limit (e.g., 100 Kib/s) during client creation.

```python
>>> client = aioftp.Client(read_speed_limit=100 * 1024)  # 100 Kib/s
```

--------------------------------

### Configure Users and Permissions in aioftp

Source: https://context7.com/aio-libs/aioftp/llms.txt

Define multiple users with specific login credentials, base paths, home directories, permissions, and speed limits. Supports anonymous access.

```python
import asyncio
import aioftp
from pathlib import Path

async def main():
    # Define users with permissions
    users = [
        # Admin user with full access
        aioftp.User(
            login="admin",
            password="admin_secret",
            base_path=Path("/srv/ftp"),
            home_path="/",
            permissions=[
                aioftp.Permission("/", readable=True, writable=True),
            ],
            maximum_connections=5,
        ),
        # Regular user with restricted access
        aioftp.User(
            login="user1",
            password="user_pass",
            base_path=Path("/srv/ftp/users/user1"),
            home_path="/home",
            permissions=[
                aioftp.Permission("/", readable=False, writable=False),
                aioftp.Permission("/home", readable=True, writable=True),
                aioftp.Permission("/shared", readable=True, writable=False),
            ],
            read_speed_limit=1024 * 1024,        # 1 MB/s total for user
            write_speed_limit=512 * 1024,        # 512 KB/s total
            read_speed_limit_per_connection=256 * 1024,  # Per connection limit
        ),
        # Anonymous user (login=None)
        aioftp.User(
            base_path=Path("/srv/ftp/public"),
            home_path="/public",
            permissions=[
                aioftp.Permission("/public", readable=True, writable=False),
            ],
        ),
    ]

    server = aioftp.Server(
        users,
        maximum_connections=100,
        read_speed_limit=10 * 1024 * 1024,  # 10 MB/s server-wide
    )

    await server.run("0.0.0.0", 21)

asyncio.run(main())

```

--------------------------------

### Implement Simple NOOP Command in aioftp Client

Source: https://github.com/aio-libs/aioftp/blob/master/docs/developer_tutorial.md

Inherit from aioftp.Client and override the noop method to implement the NOOP command. This is suitable for commands that do not require extra connections.

```python
class MyClient(aioftp.Client):
    async def noop(self):
        await self.command("NOOP", "2xx")
```

--------------------------------

### FTP Client Context Manager

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

Use the aioftp.Client.context manager for automatic connection, login, and disconnection from the FTP server.

```python
async with aioftp.Client.context("ftp.server.com", user="user", password="pass") as client:
    # do
```

--------------------------------

### Download MP3 Files Recursively

Source: https://github.com/aio-libs/aioftp/blob/master/README.rst

Connects to multiple FTP servers and downloads all MP3 files recursively. Requires server details and credentials.

```python
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())
```

--------------------------------

### Configure Client with Path Timeout and AsyncPathIO

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

Set a timeout specifically for non-blocking path I/O operations when using an asynchronous path I/O layer like AsyncPathIO.

```python
client = aioftp.Client(
    path_timeout=1,
    path_io_factory=pathio.AsyncPathIO
)
```

--------------------------------

### Change Directory

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

Navigates to a different directory on the FTP server using `aioftp.Client.change_directory()`. Pass no arguments to return to the parent directory.

```python
>>> await client.change_directory("folder1")
>>> await client.get_current_directory()
PosixPath('/public_html/folder1')
>>> await client.change_directory()
>>> await client.get_current_directory()
PosixPath('/public_html')
```

--------------------------------

### Iterate Through Recursive Directory Listing

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

Asynchronously iterate through a recursive listing of a directory on the FTP server, printing each path.

```python
async for path, info in client.list("/", recursive=True):
    print(path)
```

--------------------------------

### Download Directory

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

Download a directory from the FTP server to the current local directory.

```python
await client.download("folder2")
```

--------------------------------

### aioftp Client with Speed Throttling

Source: https://context7.com/aio-libs/aioftp/llms.txt

Control bandwidth usage by setting read and write speed limits. Limits can be configured during client initialization or modified dynamically after the client is created.

```python
import asyncio
import aioftp

async def main():
    # Set speed limits at initialization
    client = aioftp.Client(
        read_speed_limit=100 * 1024,   # 100 KB/s download
        write_speed_limit=50 * 1024,   # 50 KB/s upload
    )

    await client.connect("ftp.example.com")
    await client.login()

    # Modify throttle after creation
    client.throttle.read.limit = 200 * 1024   # Increase to 200 KB/s
    client.throttle.write.limit = None        # Unlimited upload

    await client.download("/large_file.zip")
    await client.quit()

asyncio.run(main())
```

--------------------------------

### FTP File Transfer with Client.download and Client.upload

Source: https://context7.com/aio-libs/aioftp/llms.txt

High-level methods for downloading and uploading files and directories recursively. Use `write_into=True` to specify the destination name or path.

```python
import asyncio
import aioftp
from pathlib import Path

async def main():
    async with aioftp.Client.context("ftp.example.com", user="user", password="pass") as client:
        # Download a single file to current directory
        await client.download("/remote/file.txt")

        # Download file with new name
        await client.download("/remote/file.txt", "local_copy.txt", write_into=True)

        # Download entire directory recursively
        await client.download("/remote/folder", "local_folder", write_into=True)

        # Upload a file
        await client.upload("local_file.txt", "/remote/uploads/")

        # Upload with different name
        await client.upload("local_file.txt", "/remote/renamed.txt", write_into=True)

        # Upload directory recursively
        await client.upload("./my_folder", "/remote/backup/", write_into=True)

asyncio.run(main())
```

--------------------------------

### List Directory Contents

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

List the contents of a specified directory on the FTP server. Returns a list of paths and their information.

```python
await client.list("/")
```

--------------------------------

### Download Stream with Iteration

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

Downloads a file in chunks using `aioftp.Client.download_stream()` and iterates over the data blocks. Suitable for processing data on the fly.

```python
>>> async with client.download_stream("tmp.py") as stream:
...     async for block in stream.iter_by_block():
...         # do something with data
```

--------------------------------

### Upload Directory with New Name/Path

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

Upload a local directory to a specified destination path on the FTP server, using 'write_into=True' to preserve directory structure.

```python
await client.upload("folder1", "folder2", write_into=True)
```

--------------------------------

### Check for Non-existent Path

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

Verify that a path does not exist on the FTP server.

```python
await client.exists("naked-guido.png")
```

--------------------------------

### Stream Download and Upload with aioftp

Source: https://context7.com/aio-libs/aioftp/llms.txt

Use streaming methods for on-the-fly data processing during transfers without saving to disk. Supports resuming interrupted downloads using the offset parameter.

```python
import asyncio
import aioftp
import hashlib

async def main():
    async with aioftp.Client.context("ftp.example.com") as client:
        # Stream download - calculate checksum without saving file
        md5 = hashlib.md5()
        async with client.download_stream("/remote/largefile.bin") as stream:
            async for block in stream.iter_by_block(8192):
                md5.update(block)
        print(f"MD5: {md5.hexdigest()}")

        # Stream upload - generate data on the fly
        async with client.upload_stream("/remote/generated.txt") as stream:
            for i in range(100):
                line = f"Line {i}: Generated content\n"
                await stream.write(line.encode())

        # Resume interrupted download using offset
        existing_size = 1024  # bytes already downloaded
        async with client.download_stream("/remote/file.bin", offset=existing_size) as stream:
            async for block in stream.iter_by_block():
                # Append to existing file
                pass

asyncio.run(main())
```

--------------------------------

### Implement COLL Command with Data Transfer in aioftp

Source: https://github.com/aio-libs/aioftp/blob/master/docs/developer_tutorial.md

Implement the COLL command which requires a data connection. Use ConnectionConditions decorators for login and passive server status. A worker function handles the data transfer, sending binary data chunks. Tasks are managed via connection.extra_workers.

```python
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
```

--------------------------------

### Download File with New Name/Path

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

Download a file from the FTP server to a local path, using 'write_into=True' to specify the destination file name.

```python
await client.download("tmp/test.py", "foo.py", write_into=True)
```

--------------------------------

### Upload File with New Name/Path

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

Upload a local file to a specified destination path on the FTP server, using 'write_into=True' to create intermediate directories if needed.

```python
await client.upload("test.py", "tmp/test.py", write_into=True)
```

--------------------------------

### List Current Working Directory

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

List the contents of the current working directory on the FTP server when no path is specified.

```python
await c.list()
```

--------------------------------

### Implement COLL Command with Extra Connection in aioftp Client

Source: https://github.com/aio-libs/aioftp/blob/master/docs/developer_tutorial.md

Extend aioftp.Client to implement a command that retrieves data via an extra connection using get_stream. This method is useful for commands that need to transfer data beyond simple command responses.

```python
class MyClient(aioftp.Client):

    async def collect(self, count):
        collected = []
        async with self.get_stream("COLL " + str(count), "1xx") as stream:
            async for block in stream.iter_by_block(8):
                i = int.from_bytes(block, "big")
                print("received:", block, i)
                collected.append(i)
        return collected
```

--------------------------------

### Concurrent Download with Two Clients

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

Use two separate client connections to download files from one client while iterating through the list from another, preventing blocking issues.

```python
async for path, info in client1.list():
    await client2.download(path, path.name)
```

--------------------------------

### List Directory Contents Recursively

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

Recursively list all contents of a directory on the FTP server, including subdirectories.

```python
await client.list("/", recursive=True)
```

--------------------------------

### Configure Client with Socket Timeout

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

Set a global timeout for socket I/O operations on the aioftp.Client. This prevents operations from hanging indefinitely.

```python
client = aioftp.Client(socket_timeout=1)  # 1 second socket timeout
```

--------------------------------

### Client.download and Client.upload - File Transfer

Source: https://context7.com/aio-libs/aioftp/llms.txt

High-level methods for downloading files and directories from the FTP server and uploading local files and directories to the server, supporting recursive operations.

```APIDOC
## Client.download and Client.upload - File Transfer

### Description
Provides high-level methods for downloading files and directories from the FTP server and uploading local files and directories to the server. Supports recursive transfers.

### Method
- `client.download(remotepath, localpath='.', *, write_into=False)`
- `client.upload(localpath, remotepath, *, write_into=False)`

### Parameters
#### `download` method:
- **remotepath** (str or pathlib.Path) - Required - The path on the FTP server to download.
- **localpath** (str or pathlib.Path) - Optional - The local path to save the downloaded file or directory. Defaults to the current directory.
- **write_into** (bool) - Optional - If True, and `localpath` is a directory, the contents of the remote path will be written directly into `localpath`. If False, the remote path will be created as a subdirectory within `localpath`. Defaults to False.

#### `upload` method:
- **localpath** (str or pathlib.Path) - Required - The local path to upload.
- **remotepath** (str or pathlib.Path) - Required - The destination path on the FTP server.
- **write_into** (bool) - Optional - If True, and `remotepath` is a directory, the contents of `localpath` will be written directly into `remotepath`. If False, `localpath` will be created as a subdirectory within `remotepath`. Defaults to False.

### Request Example
```python
import asyncio
import aioftp
from pathlib import Path

async def main():
    async with aioftp.Client.context("ftp.example.com", user="user", password="pass") as client:
        # Download a single file to current directory
        await client.download("/remote/file.txt")

        # Download file with new name
        await client.download("/remote/file.txt", "local_copy.txt", write_into=True)

        # Download entire directory recursively
        await client.download("/remote/folder", "local_folder", write_into=True)

        # Upload a file
        await client.upload("local_file.txt", "/remote/uploads/")

        # Upload with different name
        await client.upload("local_file.txt", "/remote/renamed.txt", write_into=True)

        # Upload directory recursively
        await client.upload("./my_folder", "/remote/backup/", write_into=True)

asyncio.run(main())
```

### Response
- **download**: None
- **upload**: None
```

--------------------------------

### Upload File to Current Directory

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

Upload a local file to the current working directory on the FTP server.

```python
await client.upload("test.py")
```

--------------------------------

### Upload Directory without write_into

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

Upload a local directory without 'write_into=True', which may result in unexpected path structures on the server.

```python
await client.upload("folder1", "folder2")
```

--------------------------------

### Remove File or Directory

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

Use `aioftp.Client.remove()` to delete files or directories recursively. No prior checks are needed.

```python
>>> await client.remove("tmp.py")
>>> await client.remove("folder1")
```

--------------------------------

### Upload Stream with Offset/Resume

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

Resumes an upload process from a specific byte offset using `aioftp.Client.upload_stream()` with the `offset` argument. Handles `ConnectionResetError` for retries.

```python
>>> while True:
...     try:
...         async with aioftp.Client.context(HOST, PORT) as client:
...             if await client.exists(filename):
...                 stat = await client.stat(filename)
...                 size = int(stat["size"])
...             else:
...                 size = 0
...             file_in.seek(size)
...             async with client.upload_stream(filename, offset=size) as stream:
...                 while True:
...                     data = file_in.read(block_size)
...                     if not data:
...                         break
...                     await stream.write(data)
...         break
...     except ConnectionResetError:
...         pass
```

--------------------------------

### Rename or Move Path

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

Use `aioftp.Client.rename()` to change the name of a file or directory, effectively moving it within the current directory structure.

```python
>>> await client.list()
[(PosixPath('test.py'), {'modify': '20150423090041', 'type': 'file', ...
>>> await client.rename("test.py", "foo.py")
>>> await client.list()
[(PosixPath('foo.py'), {'modify': '20150423090041', 'type': 'file', ...
```

--------------------------------

### Set Maximum Server Connections

Source: https://github.com/aio-libs/aioftp/blob/master/docs/server_tutorial.md

Limit the total number of concurrent connections allowed for the entire server.

```python
>>> server = aioftp.Server(maximum_connections=3)
```

--------------------------------

### Upload File without write_into

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

Upload a local file without 'write_into=True', which may result in unexpected path structures on the server.

```python
await client.upload("test.py", "tmp/test.py")
```

--------------------------------

### Check if Path Exists

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

Check if a given path exists on the FTP server.

```python
await client.exists("test.py")
```

--------------------------------

### Set Maximum User Connections

Source: https://github.com/aio-libs/aioftp/blob/master/docs/server_tutorial.md

Define the maximum number of concurrent connections allowed for a specific user.

```python
>>> users = (aioftp.User(maximum_connections=3),)
>>> server = aioftp.Server(users)
```

--------------------------------

### Configure aioftp Data Port Range

Source: https://context7.com/aio-libs/aioftp/llms.txt

Restrict passive mode data connection ports by specifying a range for firewall configurations. Also sets an external IP for NAT scenarios.

```python
import asyncio
import aioftp

async def main():
    server = aioftp.Server(
        data_ports=range(30000, 30100),  # 100 ports for passive connections
        ipv4_pasv_forced_response_address="203.0.113.10",  # External IP for NAT
    )

    await server.run("0.0.0.0", 21)

asyncio.run(main())

```

--------------------------------

### Set Write Speed Limit

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

Modifies the write speed limit of an existing `aioftp.Client` instance after its creation.

```python
>>> client.throttle.write.limit = 250 * 1024
```

--------------------------------

### Check if Path is a File

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

Check if a given path on the FTP server points to a file.

```python
await client.is_file("/public_html")
```

--------------------------------

### Check if Path is a Directory

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

Check if a given path on the FTP server points to a directory.

```python
await client.is_dir("/public_html")
```

--------------------------------

### Stop aioftp Server

Source: https://github.com/aio-libs/aioftp/blob/master/docs/server_tutorial.md

Gracefully close the aioftp server and release its resources.

```python
>>> await server.close()
```

--------------------------------

### Close FTP Connection

Source: https://github.com/aio-libs/aioftp/blob/master/docs/client_tutorial.md

Gracefully closes the FTP connection by sending the 'QUIT' command using `aioftp.Client.quit()`.

```python
>>> await client.quit()
```

--------------------------------

### Use Thread-Safe Locale Context Manager in Python

Source: https://context7.com/aio-libs/aioftp/llms.txt

Employ the setlocale context manager for temporarily changing locale in multithreaded applications. This ensures thread safety when parsing date strings that rely on specific locale formats, like C locale month abbreviations.

```python
import asyncio
from aioftp.common import setlocale
from datetime import datetime

def parse_ftp_date(date_string):
    """Parse FTP date strings that use C locale month abbreviations."""
    with setlocale("C"):
        # Thread-safe locale change
        return datetime.strptime(date_string, "%b %d %H:%M")

# Example usage
parsed = parse_ftp_date("Jan 15 14:30")
print(parsed)

```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.