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 Path Operations

Name: tessl/pypi-aiofiles
Author: tessl

Async versions of os.path module functions for path manipulation and information gathering. These functions provide file system path introspection and validation without blocking the asyncio event loop.

Capabilities

Path Existence and Type Checking

Check whether paths exist and determine their types (file, directory, symbolic link, etc.).

async def exists(path: str, *, loop = None, executor = None) -> bool:
    """
    Check if path exists.
    
    Parameters:
    - path: Path to check
    - loop: Event loop to use
    - executor: Thread pool executor to use
    
    Returns:
    True if path exists, False otherwise
    """

async def isfile(path: str, *, loop = None, executor = None) -> bool:
    """
    Check if path is a regular file.
    
    Parameters:
    - path: Path to check
    - loop: Event loop to use
    - executor: Thread pool executor to use
    
    Returns:
    True if path is a file, False otherwise
    """

async def isdir(path: str, *, loop = None, executor = None) -> bool:
    """
    Check if path is a directory.
    
    Parameters:
    - path: Path to check
    - loop: Event loop to use
    - executor: Thread pool executor to use
    
    Returns:
    True if path is a directory, False otherwise
    """

async def islink(path: str, *, loop = None, executor = None) -> bool:
    """
    Check if path is a symbolic link.
    
    Parameters:
    - path: Path to check
    - loop: Event loop to use
    - executor: Thread pool executor to use
    
    Returns:
    True if path is a symbolic link, False otherwise
    """

async def ismount(path: str, *, loop = None, executor = None) -> bool:
    """
    Check if path is a mount point.
    
    Parameters:
    - path: Path to check
    - loop: Event loop to use
    - executor: Thread pool executor to use
    
    Returns:
    True if path is a mount point, False otherwise
    """

Usage Example:

from aiofiles import ospath

# Check path existence and type
if await ospath.exists('example.txt'):
    if await ospath.isfile('example.txt'):
        print("example.txt is a file")
    elif await ospath.isdir('example.txt'):
        print("example.txt is a directory")
    elif await ospath.islink('example.txt'):
        print("example.txt is a symbolic link")

# Check for mount points
if await ospath.ismount('/mnt/external'):
    print("/mnt/external is a mount point")

File Size and Timestamps

Get file size and various timestamp information.

async def getsize(path: str, *, loop = None, executor = None) -> int:
    """
    Get file size in bytes.
    
    Parameters:
    - path: Path to file
    - loop: Event loop to use
    - executor: Thread pool executor to use
    
    Returns:
    File size in bytes
    """

async def getmtime(path: str, *, loop = None, executor = None) -> float:
    """
    Get file modification time.
    
    Parameters:
    - path: Path to file
    - loop: Event loop to use
    - executor: Thread pool executor to use
    
    Returns:
    Modification time as seconds since epoch
    """

async def getatime(path: str, *, loop = None, executor = None) -> float:
    """
    Get file access time.
    
    Parameters:
    - path: Path to file
    - loop: Event loop to use
    - executor: Thread pool executor to use
    
    Returns:
    Access time as seconds since epoch
    """

async def getctime(path: str, *, loop = None, executor = None) -> float:
    """
    Get file creation time (Windows) or metadata change time (Unix).
    
    Parameters:
    - path: Path to file
    - loop: Event loop to use
    - executor: Thread pool executor to use
    
    Returns:
    Creation/change time as seconds since epoch
    """

Usage Example:

import time
from aiofiles import ospath

# Get file information
size = await ospath.getsize('document.pdf')
print(f"File size: {size} bytes")

# Get timestamps
mtime = await ospath.getmtime('document.pdf')
atime = await ospath.getatime('document.pdf')
ctime = await ospath.getctime('document.pdf')

print(f"Modified: {time.ctime(mtime)}")
print(f"Accessed: {time.ctime(atime)}")
print(f"Created/Changed: {time.ctime(ctime)}")

# Check if file was modified recently
import asyncio
current_time = time.time()
if current_time - mtime < 3600:  # Less than 1 hour ago
    print("File was modified recently")

File Comparison

Compare files to determine if they refer to the same file system object.

async def samefile(path1: str, path2: str, *, loop = None, executor = None) -> bool:
    """
    Check if two paths refer to the same file.
    
    Parameters:
    - path1: First path to compare
    - path2: Second path to compare
    - loop: Event loop to use
    - executor: Thread pool executor to use
    
    Returns:
    True if paths refer to same file, False otherwise
    """

async def sameopenfile(fp1: int, fp2: int, *, loop = None, executor = None) -> bool:
    """
    Check if two file descriptors refer to the same file.
    
    Parameters:
    - fp1: First file descriptor
    - fp2: Second file descriptor
    - loop: Event loop to use
    - executor: Thread pool executor to use
    
    Returns:
    True if file descriptors refer to same file, False otherwise
    """

Usage Example:

from aiofiles import ospath

# Check if paths refer to same file (handles symlinks, hardlinks)
same = await ospath.samefile('file.txt', 'link_to_file.txt')
if same:
    print("Paths refer to the same file")

# Compare open file descriptors
with open('file1.txt') as f1, open('file2.txt') as f2:
    same_fd = await ospath.sameopenfile(f1.fileno(), f2.fileno())
    if same_fd:
        print("File descriptors refer to same file")

Path Manipulation

Transform and normalize file system paths.

async def abspath(path: str, *, loop = None, executor = None) -> str:
    """
    Get absolute path.
    
    Parameters:
    - path: Relative or absolute path
    - loop: Event loop to use
    - executor: Thread pool executor to use
    
    Returns:
    Absolute path string
    """

Usage Example:

from aiofiles import ospath

# Convert relative path to absolute
relative_path = '../documents/file.txt'
absolute_path = await ospath.abspath(relative_path)
print(f"Absolute path: {absolute_path}")

# Get absolute path of current directory file
current_file_abs = await ospath.abspath('data.txt')
print(f"Current file absolute path: {current_file_abs}")

Practical Usage Patterns

File Validation Pipeline

from aiofiles import ospath
import aiofiles.os

async def validate_file(path: str) -> dict:
    """Comprehensive file validation."""
    result = {
        'exists': await ospath.exists(path),
        'is_file': False,
        'is_dir': False,
        'is_link': False,
        'size': 0,
        'modified': 0
    }
    
    if result['exists']:
        result['is_file'] = await ospath.isfile(path)
        result['is_dir'] = await ospath.isdir(path)
        result['is_link'] = await ospath.islink(path)
        
        if result['is_file']:
            result['size'] = await ospath.getsize(path)
            result['modified'] = await ospath.getmtime(path)
    
    return result

# Usage
file_info = await validate_file('example.txt')
print(f"File info: {file_info}")

Directory Processing

from aiofiles import ospath
import aiofiles.os

async def process_directory(dir_path: str):
    """Process all files in directory."""
    if not await ospath.exists(dir_path):
        print(f"Directory {dir_path} does not exist")
        return
        
    if not await ospath.isdir(dir_path):
        print(f"{dir_path} is not a directory")
        return
    
    files = await aiofiles.os.listdir(dir_path)
    
    for filename in files:
        file_path = f"{dir_path}/{filename}"
        
        if await ospath.isfile(file_path):
            size = await ospath.getsize(file_path)
            print(f"File: {filename} ({size} bytes)")
        elif await ospath.isdir(file_path):
            print(f"Directory: {filename}")
        elif await ospath.islink(file_path):
            print(f"Link: {filename}")

# Usage
await process_directory('./data')

File Change Detection

import time
from aiofiles import ospath

async def file_changed_since(path: str, timestamp: float) -> bool:
    """Check if file was modified since given timestamp."""
    if not await ospath.exists(path):
        return False
        
    if not await ospath.isfile(path):
        return False
    
    mtime = await ospath.getmtime(path)
    return mtime > timestamp

# Usage
last_check = time.time()
# ... later ...
if await file_changed_since('config.txt', last_check):
    print("Config file has been modified")

Utility Functions

def wrap(func):
    """
    Wrap synchronous os.path function to run in executor.
    
    Parameters:
    - func: Synchronous os.path 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.path functions not directly provided by aiofiles.

Usage Example:

from aiofiles import ospath
import os.path

# Create async version of os.path.commonpath
async_commonpath = ospath.wrap(os.path.commonpath)
common = await async_commonpath(['/usr/lib', '/usr/local/lib'])

Error Handling

Path operations may raise various exceptions:

FileNotFoundError: Path does not exist
PermissionError: Insufficient permissions to access path
OSError: General OS-level errors
ValueError: Invalid path format

Usage Example:

from aiofiles import ospath

try:
    size = await ospath.getsize('nonexistent.txt')
except FileNotFoundError:
    print("File not found")
except PermissionError:
    print("Permission denied")
except OSError as e:
    print(f"OS error: {e}")

# Safe existence check with error handling
async def safe_exists(path: str) -> bool:
    """Safely check if path exists."""
    try:
        return await ospath.exists(path)
    except (OSError, ValueError):
        return False

exists = await safe_exists('potentially/invalid/path')

Install with Tessl CLI