tessl/pypi-aioftp
Asynchronous FTP client and server implementation for Python's asyncio framework
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
client.mddocs/
FTP Client
High-level FTP client functionality providing comprehensive file transfer operations, directory management, and connection handling. The client supports context managers, streaming operations, SSL/TLS encryption, and various authentication methods.
Capabilities
Connection Management
Establish and manage FTP connections with support for various authentication methods, SSL/TLS encryption, and connection configuration.
class Client(BaseClient):
"""High-level FTP client with file operations and directory management."""
@classmethod
def context(host: str, port: int = 21, user: str = "anonymous",
password: str = "anon@", account: str = "",
upgrade_to_tls: bool = False, **kwargs) -> AsyncGenerator[Client]:
"""
Async context manager for FTP client connections.
Parameters:
- host: FTP server hostname or IP address
- port: FTP server port (default: 21)
- user: Username for authentication (default: "anonymous")
- password: Password for authentication (default: "anon@")
- account: Account for authentication (default: "")
- upgrade_to_tls: Whether to upgrade connection to TLS (default: False)
- **kwargs: Additional arguments passed to Client constructor
Yields:
Connected and authenticated Client instance
"""
async def connect(self, host: str, port: int = 21) -> list[str]:
"""
Connect to FTP server and return greeting message.
Parameters:
- host: FTP server hostname or IP address
- port: FTP server port (default: 21)
Returns:
List of greeting message lines from server
"""
async def login(self, user: str = "anonymous", password: str = "anon@",
account: str = "") -> None:
"""
Authenticate with the FTP server.
Parameters:
- user: Username for authentication
- password: Password for authentication
- account: Account for authentication (rarely used)
"""
async def upgrade_to_tls(self, sslcontext: ssl.SSLContext = None) -> None:
"""
Upgrade connection to TLS encryption.
Parameters:
- sslcontext: SSL context for encryption (uses default if None)
"""
async def quit(self) -> None:
"""Gracefully close FTP connection."""
def close(self) -> None:
"""Force close FTP connection immediately."""File Transfer Operations
Upload and download files with support for streaming, resuming transfers, and various transfer modes.
class Client:
async def upload(self, source, destination: str = "", write_into: bool = False,
block_size: int = 8192) -> None:
"""
Upload a file to the FTP server.
Parameters:
- source: Local file path (str/Path) or file-like object to upload
- destination: Remote destination path (uses source name if empty)
- write_into: If True, write into existing remote directory
- block_size: Transfer block size in bytes
"""
async def download(self, source, destination: str = "", write_into: bool = False,
block_size: int = 8192) -> None:
"""
Download a file from the FTP server.
Parameters:
- source: Remote file path to download
- destination: Local destination path (uses source name if empty)
- write_into: If True, write into existing local directory
- block_size: Transfer block size in bytes
"""
def upload_stream(self, destination: str, offset: int = 0) -> AsyncEnterableInstanceProtocol[DataConnectionThrottleStreamIO]:
"""
Get upload stream for writing data to remote file.
Parameters:
- destination: Remote file path to upload to
- offset: Byte offset to start upload (for resume)
Returns:
Async context manager yielding writable stream
"""
def download_stream(self, source: str, offset: int = 0) -> AsyncEnterableInstanceProtocol[DataConnectionThrottleStreamIO]:
"""
Get download stream for reading data from remote file.
Parameters:
- source: Remote file path to download from
- offset: Byte offset to start download (for resume)
Returns:
Async context manager yielding readable stream
"""
def append_stream(self, destination: str, offset: int = 0) -> AsyncEnterableInstanceProtocol[DataConnectionThrottleStreamIO]:
"""
Get append stream for appending data to remote file.
Parameters:
- destination: Remote file path to append to
- offset: Byte offset to start append
Returns:
Async context manager yielding writable stream
"""Directory Operations
Navigate and manipulate remote directory structure including creating, removing, and changing directories.
class Client:
async def get_current_directory(self) -> PurePosixPath:
"""
Get current working directory on remote server.
Returns:
Current directory path as PurePosixPath
"""
async def change_directory(self, path: str = "..") -> None:
"""
Change current working directory on remote server.
Parameters:
- path: Directory path to change to (default: ".." for parent)
"""
async def make_directory(self, path: str, parents: bool = True) -> None:
"""
Create directory on remote server.
Parameters:
- path: Directory path to create
- parents: Create parent directories if they don't exist
"""
async def remove_directory(self, path: str) -> None:
"""
Remove directory from remote server.
Parameters:
- path: Directory path to remove (must be empty)
"""Directory Listing
List and inspect remote directory contents with support for detailed file information and recursive listing.
class Client:
def list(self, path: str = "", recursive: bool = False,
raw_command: str = None) -> AbstractAsyncLister:
"""
List directory contents on remote server.
Parameters:
- path: Directory path to list (current directory if empty)
- recursive: List subdirectories recursively
- raw_command: Raw FTP command to use instead of default
Returns:
Async iterator yielding (path, info) tuples
"""
async def stat(self, path: str) -> Union[BasicListInfo, UnixListInfo]:
"""
Get detailed information about remote path.
Parameters:
- path: Remote path to inspect
Returns:
Dictionary with file/directory information
"""
async def exists(self, path: str) -> bool:
"""
Check if remote path exists.
Parameters:
- path: Remote path to check
Returns:
True if path exists, False otherwise
"""
async def is_file(self, path: str) -> bool:
"""
Check if remote path is a file.
Parameters:
- path: Remote path to check
Returns:
True if path is a file, False otherwise
"""
async def is_dir(self, path: str) -> bool:
"""
Check if remote path is a directory.
Parameters:
- path: Remote path to check
Returns:
True if path is a directory, False otherwise
"""File Operations
Manipulate individual files including renaming, deleting, and generic removal operations.
class Client:
async def rename(self, source: str, destination: str) -> None:
"""
Rename/move remote file or directory.
Parameters:
- source: Current path of file/directory
- destination: New path for file/directory
"""
async def remove_file(self, path: str) -> None:
"""
Remove file from remote server.
Parameters:
- path: File path to remove
"""
async def remove(self, path: str) -> None:
"""
Remove file or directory from remote server.
Parameters:
- path: Path to remove (file or empty directory)
"""Connection Control
Advanced connection operations including abort and low-level protocol access.
class Client:
async def abort(self, wait: bool = True) -> None:
"""
Abort current FTP operation.
Parameters:
- wait: Wait for abort confirmation from server
"""Usage Examples
Basic File Transfer
import aioftp
import asyncio
async def basic_transfer():
async with aioftp.Client.context("ftp.example.com", user="username", password="password") as client:
# Upload a file
await client.upload("local_file.txt", "remote_file.txt")
# Download a file
await client.download("remote_file.txt", "downloaded_file.txt")
# Check if file exists
if await client.exists("remote_file.txt"):
print("File exists on server")
asyncio.run(basic_transfer())Directory Management
import aioftp
import asyncio
async def directory_operations():
async with aioftp.Client.context("ftp.example.com") as client:
# Get current directory
current = await client.get_current_directory()
print(f"Current directory: {current}")
# Create a new directory
await client.make_directory("new_folder")
# Change to the new directory
await client.change_directory("new_folder")
# List contents
async for path, info in client.list():
file_type = "DIR" if info.get("type") == "dir" else "FILE"
size = info.get("size", "unknown")
print(f"{file_type}: {path} ({size} bytes)")
asyncio.run(directory_operations())Streaming Transfers
import aioftp
import asyncio
async def streaming_transfer():
async with aioftp.Client.context("ftp.example.com") as client:
# Stream upload
async with client.upload_stream("remote_file.txt") as stream:
await stream.write(b"Hello, ")
await stream.write(b"World!")
# Stream download
async with client.download_stream("remote_file.txt") as stream:
data = await stream.read()
print(f"Downloaded: {data.decode()}")
asyncio.run(streaming_transfer())SSL/TLS Connection
import aioftp
import asyncio
import ssl
async def secure_connection():
# Create SSL context
context = ssl.create_default_context()
async with aioftp.Client.context(
"ftps.example.com",
port=990, # FTPS implicit port
user="username",
password="password",
ssl=context
) as client:
await client.upload("local_file.txt", "secure_file.txt")
asyncio.run(secure_connection())Types
# Type aliases and protocols
PathWithInfo = tuple[PurePosixPath, Union[BasicListInfo, UnixListInfo]]
class BasicListInfo(TypedDict):
"""Basic file information from directory listing."""
type: str # "file" or "dir"
size: int # File size in bytes
modify: str # Modification time string
class UnixListInfo(BasicListInfo):
"""Extended Unix-style file information."""
permissions: str # Permission string (e.g., "rwxr-xr-x")
owner: str # Owner name
group: str # Group name
links: int # Number of hard links
class DataConnectionThrottleStreamIO:
"""Throttled stream for FTP data connections."""
async def read(count: int = -1) -> bytes:
"""Read data from stream."""
async def write(data: bytes) -> None:
"""Write data to stream."""
async def finish(expected_codes: str = "2xx", wait_codes: str = "1xx") -> None:
"""Finish data transfer and wait for server confirmation."""