Error Handling

class AIOFTPException(Exception):
    """
    Base exception for all aioftp errors.
    
    This is the root exception class that all other aioftp exceptions inherit from.
    Catching this exception will catch all aioftp-specific errors.
    """

class StatusCodeError(AIOFTPException):
    """
    Raised when FTP server returns unexpected status code.
    
    This exception is raised when the server returns a status code that doesn't
    match what was expected for a particular FTP command. It includes detailed
    information about both expected and received codes.
    """
    
    expected_codes: tuple[Code, ...]
    """Status codes that were expected from the server."""
    
    received_codes: tuple[Code, ...]  
    """Status codes actually received from the server."""
    
    info: Union[list[str], str]
    """Additional information from the server response."""
    
    def __init__(expected_codes, received_codes, info):
        """
        Initialize StatusCodeError.
        
        Parameters:
        - expected_codes: Codes that were expected
        - received_codes: Codes actually received  
        - info: Server response information
        """

class PathIsNotAbsolute(AIOFTPException):
    """
    Raised when path is not absolute but should be.
    
    Some operations require absolute paths. This exception is raised when
    a relative path is provided where an absolute path is required.
    """
    
    def __init__(path):
        """
        Initialize PathIsNotAbsolute error.
        
        Parameters:
        - path: The invalid relative path that was provided
        """

class PathIOError(AIOFTPException):
    """
    Universal exception for path I/O operations.
    
    This exception wraps filesystem-related errors and provides additional
    context about the operation that failed. It preserves the original
    exception information for debugging.
    """
    
    reason: Union[tuple, None]
    """Original exception information from the wrapped error."""
    
    def __init__(reason):
        """
        Initialize PathIOError.
        
        Parameters:
        - reason: Original exception info (usually from sys.exc_info())
        """

class NoAvailablePort(AIOFTPException, OSError):
    """
    Raised when no data ports are available for FTP data connections.
    
    This exception is raised when the server cannot find an available port
    from the configured data port range for establishing data connections.
    Inherits from both AIOFTPException and OSError for compatibility.
    """
    
    def __init__(message="No available ports"):
        """
        Initialize NoAvailablePort error.
        
        Parameters:
        - message: Error message describing the port availability issue
        """

import aioftp
import asyncio

async def handle_basic_errors():
    try:
        async with aioftp.Client.context("ftp.example.com") as client:
            await client.upload("local_file.txt", "remote_file.txt")
    
    except aioftp.StatusCodeError as e:
        print(f"FTP protocol error: {e}")
        print(f"Expected: {e.expected_codes}, Got: {e.received_codes}")
        print(f"Server message: {e.info}")
    
    except aioftp.PathIOError as e:
        print(f"File system error: {e}")
        if e.reason:
            print(f"Original error: {e.reason}")
    
    except aioftp.AIOFTPException as e:
        print(f"General FTP error: {e}")
    
    except Exception as e:
        print(f"Unexpected error: {e}")

asyncio.run(handle_basic_errors())

import aioftp
import asyncio

async def handle_specific_errors():
    try:
        async with aioftp.Client.context("ftp.example.com") as client:
            # This might raise StatusCodeError if file doesn't exist
            await client.download("nonexistent_file.txt", "local_file.txt")
    
    except aioftp.StatusCodeError as e:
        if "550" in str(e.received_codes):
            print("File not found on server")
        elif "426" in str(e.received_codes):
            print("Connection closed during transfer")
        else:
            print(f"Other protocol error: {e}")
    
    except aioftp.PathIsNotAbsolute as e:
        print(f"Path must be absolute: {e}")
    
    except aioftp.NoAvailablePort as e:
        print(f"Server has no available data ports: {e}")

asyncio.run(handle_specific_errors())

import aioftp
import asyncio
from pathlib import Path

async def server_error_handling():
    try:
        users = [aioftp.User(
            login="test", 
            password="test",
            base_path=Path("/nonexistent/path")  # This might cause issues
        )]
        
        server = aioftp.Server(users=users)
        await server.run(host="localhost", port=21)
    
    except aioftp.PathIOError as e:
        print(f"Server path error: {e}")
        print("Check that base paths exist and are accessible")
    
    except aioftp.NoAvailablePort as e:
        print(f"Port binding error: {e}")
        print("Try a different port or check port availability")
    
    except aioftp.AIOFTPException as e:
        print(f"Server startup error: {e}")

asyncio.run(server_error_handling())

import aioftp
import asyncio
import logging

async def robust_ftp_operation():
    """Example with retry logic and comprehensive error handling."""
    
    max_retries = 3
    retry_delay = 1.0
    
    for attempt in range(max_retries):
        try:
            async with aioftp.Client.context("ftp.example.com") as client:
                await client.upload("important_file.txt", "backup.txt")
                print("Upload successful!")
                return
        
        except aioftp.StatusCodeError as e:
            if "550" in str(e.received_codes):
                # Permission denied or file not found - don't retry
                print(f"Permanent error: {e}")
                break
            elif "4" in str(e.received_codes)[0]:
                # Temporary error (4xx codes) - retry
                print(f"Temporary error (attempt {attempt + 1}): {e}")
                if attempt < max_retries - 1:
                    await asyncio.sleep(retry_delay)
                    continue
        
        except aioftp.PathIOError as e:
            print(f"Local file error: {e}")
            break  # Don't retry file system errors
        
        except (ConnectionError, OSError) as e:
            # Network errors - retry
            print(f"Network error (attempt {attempt + 1}): {e}")
            if attempt < max_retries - 1:
                await asyncio.sleep(retry_delay)
                continue
        
        except aioftp.AIOFTPException as e:
            print(f"FTP error: {e}")
            break
    
    print("All retry attempts failed")

asyncio.run(robust_ftp_operation())

import aioftp
import asyncio
import logging

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

async def monitored_ftp_operation():
    """Example with comprehensive error logging."""
    
    try:
        async with aioftp.Client.context("ftp.example.com") as client:
            # Log successful connection
            logger.info("Connected to FTP server")
            
            await client.upload("data.txt", "remote_data.txt") 
            logger.info("File uploaded successfully")
            
    except aioftp.StatusCodeError as e:
        logger.error(
            "FTP protocol error: expected %s, got %s, info: %s",
            e.expected_codes, e.received_codes, e.info
        )
        # Could send alert to monitoring system here
    
    except aioftp.PathIOError as e:
        logger.error("File system error: %s", e)
        if e.reason:
            logger.debug("Original exception: %s", e.reason)
    
    except aioftp.NoAvailablePort as e:
        logger.critical("No available ports for FTP data connection: %s", e)
        # Critical infrastructure issue
    
    except aioftp.AIOFTPException as e:
        logger.error("General FTP error: %s", e)
    
    except Exception as e:
        logger.exception("Unexpected error in FTP operation")

asyncio.run(monitored_ftp_operation())