tessl/pypi-aiofiles

File support for asyncio applications providing async/await compatible file operations.

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

OS File System Operations

Name: tessl/pypi-aiofiles
Author: tessl

Async versions of os module functions for file system operations like renaming, removing, creating directories, and getting file statistics. These functions delegate to thread pool executors to avoid blocking the asyncio event loop.

Capabilities

File Information

Get detailed information about files and directories including statistics, permissions, and metadata.

async def stat(path: str, *, loop = None, executor = None):
    """
    Get file or directory status.
    
    Parameters:
    - path: Path to file or directory
    - loop: Event loop to use
    - executor: Thread pool executor to use
    
    Returns:
    os.stat_result object with file information
    """

async def access(path: str, mode: int, *, loop = None, executor = None) -> bool:
    """
    Check access permissions for path.
    
    Parameters:
    - path: Path to check
    - mode: Access mode (os.F_OK, os.R_OK, os.W_OK, os.X_OK)
    - loop: Event loop to use  
    - executor: Thread pool executor to use
    
    Returns:
    True if access is allowed, False otherwise
    """

Usage Example:

import aiofiles.os
import os

# Get file statistics
stat_result = await aiofiles.os.stat('example.txt')
print(f"File size: {stat_result.st_size}")
print(f"Modified: {stat_result.st_mtime}")
print(f"Mode: {oct(stat_result.st_mode)}")

# Check file permissions
can_read = await aiofiles.os.access('example.txt', os.R_OK)
can_write = await aiofiles.os.access('example.txt', os.W_OK)
exists = await aiofiles.os.access('example.txt', os.F_OK)

File Operations

Basic file operations including renaming, removing, and linking files.

async def rename(src: str, dst: str, *, loop = None, executor = None):
    """
    Rename file or directory from src to dst.
    
    Parameters:
    - src: Source path
    - dst: Destination path
    - loop: Event loop to use
    - executor: Thread pool executor to use
    """

async def renames(old: str, new: str, *, loop = None, executor = None):
    """
    Recursive rename with directory creation and cleanup.
    
    Parameters:
    - old: Old path
    - new: New path (directories created as needed)
    - loop: Event loop to use
    - executor: Thread pool executor to use
    """

async def replace(src: str, dst: str, *, loop = None, executor = None):
    """
    Replace dst with src, overwriting dst if it exists.
    
    Parameters:
    - src: Source file path
    - dst: Destination file path
    - loop: Event loop to use
    - executor: Thread pool executor to use
    """

async def remove(path: str, *, loop = None, executor = None):
    """
    Remove (delete) file.
    
    Parameters:
    - path: Path to file to remove
    - loop: Event loop to use
    - executor: Thread pool executor to use
    """

async def unlink(path: str, *, loop = None, executor = None):
    """
    Remove (delete) file. Same as remove().
    
    Parameters:
    - path: Path to file to unlink
    - loop: Event loop to use
    - executor: Thread pool executor to use
    """

Usage Example:

import aiofiles.os

# Rename file
await aiofiles.os.rename('old_name.txt', 'new_name.txt')

# Replace file (overwrites destination)
await aiofiles.os.replace('source.txt', 'destination.txt')

# Remove file
await aiofiles.os.remove('unwanted_file.txt')

# Recursive rename with directory creation
await aiofiles.os.renames('old/path/file.txt', 'new/path/file.txt')

Directory Operations

Create, remove, and manage directories and their contents.

async def mkdir(path: str, mode: int = 0o777, *, loop = None, executor = None):
    """
    Create directory.
    
    Parameters:
    - path: Directory path to create
    - mode: Directory permissions (default 0o777)
    - loop: Event loop to use
    - executor: Thread pool executor to use
    """

async def makedirs(name: str, mode: int = 0o777, exist_ok: bool = False, *, loop = None, executor = None):
    """
    Create directories recursively.
    
    Parameters:
    - name: Directory path to create (including parents)
    - mode: Directory permissions (default 0o777)
    - exist_ok: Don't raise error if directory already exists
    - loop: Event loop to use
    - executor: Thread pool executor to use
    """

async def rmdir(path: str, *, loop = None, executor = None):
    """
    Remove empty directory.
    
    Parameters:
    - path: Directory path to remove
    - loop: Event loop to use
    - executor: Thread pool executor to use
    """

async def removedirs(name: str, *, loop = None, executor = None):
    """
    Remove directories recursively.
    
    Parameters:
    - name: Directory path to remove (including empty parents)
    - loop: Event loop to use
    - executor: Thread pool executor to use
    """

async def listdir(path: str = ".", *, loop = None, executor = None) -> list:
    """
    List directory contents.
    
    Parameters:
    - path: Directory path to list (defaults to current directory)
    - loop: Event loop to use
    - executor: Thread pool executor to use
    
    Returns:
    List of filenames in directory
    """

async def scandir(path: str = ".", *, loop = None, executor = None):
    """
    Return iterator of DirEntry objects for directory.
    
    Parameters:
    - path: Directory path to scan (defaults to current directory)
    - loop: Event loop to use
    - executor: Thread pool executor to use
    
    Returns:
    Iterator of os.DirEntry objects
    """

Usage Example:

import aiofiles.os

# Create single directory
await aiofiles.os.mkdir('new_directory')

# Create directory hierarchy
await aiofiles.os.makedirs('path/to/nested/directory', exist_ok=True)

# List directory contents
files = await aiofiles.os.listdir('.')
print(f"Files in current directory: {files}")

# Scan directory with detailed info
async for entry in await aiofiles.os.scandir('.'):
    if entry.is_file():
        print(f"File: {entry.name}")
    elif entry.is_dir():
        print(f"Directory: {entry.name}")

# Remove empty directory
await aiofiles.os.rmdir('empty_directory')

# Remove directory tree
await aiofiles.os.removedirs('path/to/nested/directory')

Symbolic Links

Create and manage symbolic links between files and directories.

async def symlink(src: str, dst: str, target_is_directory: bool = False, *, loop = None, executor = None):
    """
    Create symbolic link.
    
    Parameters:
    - src: Source path (link target)
    - dst: Destination path (link location)
    - target_is_directory: Whether target is directory (Windows only)
    - loop: Event loop to use
    - executor: Thread pool executor to use
    """

async def readlink(path: str, *, loop = None, executor = None) -> str:
    """
    Read symbolic link target.
    
    Parameters:
    - path: Path to symbolic link
    - loop: Event loop to use
    - executor: Thread pool executor to use
    
    Returns:
    Target path of symbolic link
    """

Usage Example:

import aiofiles.os

# Create symbolic link
await aiofiles.os.symlink('target_file.txt', 'link_to_file.txt')

# Read symbolic link
target = await aiofiles.os.readlink('link_to_file.txt')
print(f"Link points to: {target}")

# Create directory symbolic link (Windows)
await aiofiles.os.symlink('target_dir', 'link_to_dir', target_is_directory=True)

System Information

Get current working directory and system-level file operations.

async def getcwd(*, loop = None, executor = None) -> str:
    """
    Get current working directory.
    
    Parameters:
    - loop: Event loop to use
    - executor: Thread pool executor to use
    
    Returns:
    Current working directory path
    """

Platform-specific functions (available if supported by OS):

async def link(src: str, dst: str, *, loop = None, executor = None):
    """Create hard link (Unix only)."""

async def sendfile(out_fd: int, in_fd: int, offset: int, count: int, *, loop = None, executor = None) -> int:
    """Send file data between file descriptors (Unix only)."""

async def statvfs(path: str, *, loop = None, executor = None):
    """Get filesystem statistics (Unix only)."""

Usage Example:

import aiofiles.os

# Get current directory
cwd = await aiofiles.os.getcwd()
print(f"Current directory: {cwd}")

# Platform-specific operations (if available)
if hasattr(aiofiles.os, 'link'):
    await aiofiles.os.link('source.txt', 'hardlink.txt')

if hasattr(aiofiles.os, 'statvfs'):
    fs_stats = await aiofiles.os.statvfs('.')
    print(f"Free space: {fs_stats.f_bavail * fs_stats.f_frsize}")

Utility Functions

def wrap(func):
    """
    Wrap synchronous function to run in executor.
    
    Parameters:
    - func: Synchronous function to wrap
    
    Returns:
    Async function that runs the original function in thread pool executor
    """

This utility function can be used to create async versions of other os module functions not directly provided by aiofiles.

Usage Example:

import aiofiles.os
import os

# Create async version of os.getcwdb (get current directory as bytes)
async_getcwdb = aiofiles.os.wrap(os.getcwdb)
current_dir_bytes = await async_getcwdb()

Error Handling

OS operations may raise various exceptions:

FileNotFoundError: File or directory not found
FileExistsError: File or directory already exists
PermissionError: Insufficient permissions
IsADirectoryError: Expected file but got directory
NotADirectoryError: Expected directory but got file
OSError: General OS-level errors
NotImplementedError: Operation not supported on platform

Usage Example:

import aiofiles.os

try:
    await aiofiles.os.remove('nonexistent.txt')
except FileNotFoundError:
    print("File not found")
except PermissionError:
    print("Permission denied")
except OSError as e:
    print(f"OS error: {e}")

try:
    await aiofiles.os.mkdir('existing_directory')
except FileExistsError:
    print("Directory already exists")

Install with Tessl CLI