Skip to content

File Access

Haiway provides a functional, context-aware file access interface that integrates seamlessly with the framework's state management and structured concurrency features. The file access system supports asynchronous operations with proper resource management and optional exclusive locking.

Overview

The file access system in Haiway follows the framework's core principles:

  • Functional Interface: All operations are performed through class methods on the File state
  • Context Integration: File access is scoped to the context, ensuring proper resource cleanup
  • Resource Safety: Automatic file handle cleanup even when exceptions occur
  • Cross-Platform: Works on Unix/Linux/macOS with exclusive locking, degrades gracefully on Windows

Quick Start

1. Basic File Operations

Open and work with files using the context system:

from haiway import ctx
from haiway.helpers import Files, File

async def read_config():
    # Open file for reading and writing
    async with ctx.scope(
        "read_config",
        disposables=(Files.access("config.json"),)
    ):
        # Read file contents
        content = await File.read()
        # Process the data
        print(f"Config data: {content.decode()}")

        # Write updated content
        new_config = b'{"setting": "updated"}'
        await File.write(new_config)

2. Creating New Files

Create files and their parent directories automatically:

async def create_data_file():
    async with ctx.scope(
        "create_file",
        disposables=(Files.access("data/output.txt", create=True),)
    ):
        # Write initial content
        await File.write(b"Initial data\n")

        # Read it back
        content = await File.read()
        print(f"Written: {content.decode()}")

3. Exclusive File Access

Use exclusive locking for critical operations:

async def update_shared_resource():
    # Exclusive lock prevents concurrent access
    async with ctx.scope(
        "exclusive_update",
        disposables=(Files.access("shared.json", exclusive=True),)
    ):
        # Read current state
        data = await File.read()
        parsed = json.loads(data)

        # Update state
        parsed["counter"] += 1
        parsed["last_update"] = datetime.now().isoformat()

        # Replace the file contents in-place
        await File.write(json.dumps(parsed).encode())

File Access Options

Files.access() Parameters

Configure file access behavior:

Files.access(
    path="data/file.txt",      # Path as string or Path object
    create=False,              # Create file if it doesn't exist
    exclusive=False,           # Use exclusive locking (Unix/Linux/macOS)
    mode=0o600,                # Permissions of a created file - owner only by default
)
  • path: File path to open (absolute or relative)
  • create: If True, creates the file and parent directories if needed
  • exclusive: If True, acquires an exclusive flock for the duration of the context, waiting for a conflicting holder to release it. The lock is advisory - it excludes other holders of the same lock, not every access to the file - and it is unavailable on Windows
  • mode: Permission bits applied to a file created by this call. Defaults to 0o600 - readable and writable by the current user only. Parent directories created along the way receive the same permissions extended with search access wherever read or write is granted, so mode=0o640 creates 0o750 directories. Permissions of an existing file are never modified.

Symbolic links are refused rather than followed: opening a path which is a link raises FileException, so a link planted at the target cannot redirect reads or writes to another file. Only regular files are accepted, so a fifo or a device planted at the target is refused as well. Path components are not normalized - a .. in the path still leads out of the directory preceding it.

Error Handling

File and directory helpers raise FileException for operational failures (open, read, write, traverse, close, and unlock). Original provider errors are preserved as __cause__ for debugging.

from haiway.helpers import FileException

async def safe_file_read():
    try:
        async with ctx.scope(
            "read_file",
            disposables=(Files.access("data.txt"),)
        ):
            content = await File.read()
            return content.decode()
    except FileException as e:
        print(f"File operation failed: {e}")
        return None

Advanced Usage

Working with Binary Data

async def process_binary_file():
    async with ctx.scope(
        "binary_ops",
        disposables=(Files.access("image.png"),)
    ):
        # Read binary data
        image_data = await File.read()

        # Process the data
        processed = transform_image(image_data)

        # Write back as binary
        await File.write(processed)

Text File Handling

async def update_text_file():
    async with ctx.scope(
        "text_file",
        disposables=(Files.access("document.txt"),)
    ):
        # Read as text
        content = await File.read()
        text = content.decode("utf-8")

        # Modify text
        updated_text = text.replace("old", "new")

        # Write back as UTF-8
        await File.write(updated_text.encode("utf-8"))

Configuration File Management

import json
import yaml
from datetime import datetime

async def manage_config():
    # JSON configuration
    async with ctx.scope(
        "json_config",
        disposables=(Files.access("config.json", create=True),)
    ):
        try:
            data = json.loads(await File.read())
        except (json.JSONDecodeError, FileException):
            data = {"version": "1.0", "settings": {}}

        data["settings"]["updated"] = True
        await File.write(json.dumps(data, indent=2).encode())

    # YAML configuration
    async with ctx.scope(
        "yaml_config",
        disposables=(Files.access("config.yaml", create=True),)
    ):
        try:
            content = await File.read()
            data = yaml.safe_load(content) or {}
        except FileException:
            data = {}

        data["last_run"] = datetime.now().isoformat()
        await File.write(yaml.dump(data).encode())

Durable In-Place Updates

Writes are in-place with os.write followed by fsync; they are not copy-on-write or rename atomic. Use exclusive locks plus your own temp-file strategy if you need atomic replacement.

Contents are always read in full and written in full, so a file has to fit in memory twice over - this interface is meant for configuration and small payloads rather than for streaming.

async def atomic_update():
    async with ctx.scope(
        "atomic",
        disposables=(Files.access("critical.dat", exclusive=True),)
    ):
        # Read current state
        current = await File.read()

        # Process data (may take time)
        processed = expensive_computation(current)

        # Replace file contents and fsync the result
        await File.write(processed)
        # File is automatically fsync'd to ensure durability

Testing

Mock file operations for testing:

from pathlib import Path

from haiway import ctx
from haiway.helpers import File, FileException

# Create mock implementations
class MockFileStorage:
    def __init__(self) -> None:
        self.files: dict[str, bytes] = {}

    async def read_file(self, path: str) -> bytes:
        if path not in self.files:
            raise FileException(f"File not found: {path}")
        return self.files[path]

    async def write_file(self, path: str, content: bytes) -> None:
        self.files[path] = content

# Use in tests
async def test_file_processing():
    storage = MockFileStorage()
    storage.files["input.txt"] = b"test data"

    # Create mock file state
    file_state = File(
        path=Path("input.txt"),
        reading=lambda: storage.read_file("input.txt"),
        writing=lambda content: storage.write_file("output.txt", content),
    )

    async with ctx.scope("test", file_state):
        # Test file operations
        data = await File.read()
        assert data == b"test data"

        await File.write(b"processed data")
        assert storage.files["output.txt"] == b"processed data"

File is constructed with the reading and writing protocol implementations - FileReading and FileWriting from haiway.helpers.files when you need to type them explicitly - which is what Files.access(...) injects for real files.

Best Practices

  1. Always use as disposable: Ensures file handles are properly closed
  2. Handle exceptions: Wrap file operations in try/except blocks
  3. Use exclusive locking: For critical updates that must not overlap
  4. Create parent directories: Use create=True when writing to new locations
  5. Avoid multiple concurrent opens: You can open more than one file in a scope; the newest File state shadows earlier ones. Prefer one file per scope to avoid surprises.

Platform Considerations

Unix/Linux/macOS

  • Full support for exclusive file locking via fcntl
  • In-place writes with proper fsync, and a directory fsync when a file is created, so a new entry survives a crash along with its contents
  • Created files and directories receive the requested mode (narrowed by the process umask), restricted to the owner by default
  • Symbolic links are refused through O_NOFOLLOW, irregular files through a S_ISREG check

Windows

  • No exclusive locking (fcntl not available) - exclusive=True raises FileException
  • Still provides in-place writes with fsync, but a newly created directory entry is only as durable as the platform makes it
  • Parent directories are not verified component by component, so only the file itself is checked for being a link
  • File handles properly managed

Implementation Details

The file access system uses:

  • OS-level file operations: Direct use of os.open, os.read, os.write for efficiency
  • Exclusive locking: fcntl.flock on supported platforms, acquired by polling its non-blocking variant from the event loop so that waiting for it stays cancellable
  • Durable writes: In-place writes, truncate, and fsync ensure durability
  • Async wrappers: File I/O operations run in thread pool to avoid blocking. An operation whose caller stopped waiting still runs to completion, and the file is not closed underneath it

Custom Implementations

Create custom file access implementations by implementing the protocols. File, Files, Directory, FileException, and Paths.

from pathlib import Path
from types import TracebackType

from haiway import ctx
from haiway.helpers import File, FileAccess, Files

class RemoteFileContext:
    def __init__(
        self,
        server_url: str,
        path: Path | str,
        create: bool,
        exclusive: bool,
        mode: int,
    ) -> None:
        self.server_url = server_url
        self._path = Path(path)

    @property
    def path(self) -> Path:
        return self._path

    async def __aenter__(self) -> File:
        # Open the remote resource and expose it through File reading/writing
        ...

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
        # Flush pending changes and release the remote resource
        ...

class RemoteFileAccess:
    def __init__(self, server_url: str):
        self.server_url = server_url

    def __call__(
        self,
        path: Path | str,
        create: bool,
        exclusive: bool,
        mode: int,
    ) -> FileAccess:
        # Return a context manager that provides File operations
        return RemoteFileContext(self.server_url, path, create, exclusive, mode)

# Use custom implementation
async with ctx.scope(
    "remote_files",
    Files(accessing=RemoteFileAccess("https://files.example.com"))
):
    async with ctx.scope(
        "read_remote",
        disposables=(Files.access("remote/data.txt"),)
    ):
        content = await File.read()

This allows integration with cloud storage, databases, or any custom file storage backend while maintaining the same interface.