### Concatenate Files (cat) Example

Source: https://github.com/mosquito/aiofile/blob/master/README.md

A command-line utility example that reads multiple files asynchronously and prints their content to standard output, mimicking the `cat` command.

```Python
import asyncio
import sys
from argparse import ArgumentParser
from pathlib import Path

from aiofile import async_open

parser = ArgumentParser(
    description="Read files line by line using asynchronous io API"
)
parser.add_argument("file_name", nargs="+", type=Path)

async def main(arguments):
    for src in arguments.file_name:
        async with async_open(src, "r") as afp:
            async for line in afp:
                sys.stdout.write(line)


asyncio.run(main(parser.parse_args()))
```

--------------------------------

### Basic async_open Example

Source: https://github.com/mosquito/aiofile/blob/master/README.md

Demonstrates basic asynchronous file writing and reading using async_open with a context manager. It writes 'Hello world', seeks to the beginning, reads the content, writes more data, and reads lines.

```Python
import asyncio
from pathlib import Path
from tempfile import gettempdir

from aiofile import async_open

tmp_filename = Path(gettempdir()) / "hello.txt"

async def main():
    async with async_open(tmp_filename, 'w+') as afp:
        await afp.write("Hello ")
        await afp.write("world")
        afp.seek(0)

        print(await afp.read())

        await afp.write("Hello from\nasync world")
        print(await afp.readline())
        print(await afp.readline())

asyncio.run(main())
```

--------------------------------

### Copy File (cp) Example

Source: https://github.com/mosquito/aiofile/blob/master/README.md

An example demonstrating asynchronous file copying using `async_open` and `iter_chunked`. It copies content from a source file to a destination file in chunks.

```Python
import asyncio
from argparse import ArgumentParser
from pathlib import Path

from aiofile import async_open

parser = ArgumentParser(
    description="Copying files using asynchronous io API"
)
parser.add_argument("source", type=Path)
parser.add_argument("dest", type=Path)
parser.add_argument("--chunk-size", type=int, default=65535)


async def main(arguments):
    async with async_open(arguments.source, "rb") as src, \
               async_open(arguments.dest, "wb") as dest:
        async for chunk in src.iter_chunked(arguments.chunk_size):
            await dest.write(chunk)


asyncio.run(main(parser.parse_args()))
```

--------------------------------

### Writing and Reading with Specific Offsets

Source: https://github.com/mosquito/aiofile/blob/master/README.md

Illustrates writing data at a specific offset and then reading the entire file content. This example shows how to manage file pointers for independent I/O operations.

```python
import asyncio
from aiofile import AIOFile


async def main():
    async with AIOFile("/tmp/hello.txt", 'w+') as afp:
        await afp.write("Hello ")
        await afp.write("world", offset=7)
        await afp.fsync()

        print(await afp.read())


asyncio.run(main())
```

--------------------------------

### async_open Without Context Manager

Source: https://github.com/mosquito/aiofile/blob/master/README.md

Shows how to use async_open without a context manager, explicitly calling `await afp.close()`. This example writes 'Hello' to a temporary file.

```Python
import asyncio
import atexit
import os
from tempfile import mktemp

from aiofile import async_open


TMP_NAME = mktemp()
atexit.register(os.unlink, TMP_NAME)


async def main():
    afp = await async_open(TMP_NAME, "w")
    await afp.write("Hello")
    await afp.close()


asyncio.run(main())
assert open(TMP_NAME, "r").read() == "Hello"
```

--------------------------------

### aiofile async_open Helper Methods

Source: https://github.com/mosquito/aiofile/blob/master/README.md

Documentation for the async_open helper, detailing its supported asynchronous file-like methods including read, write, seek, tell, readline, and iterators.

```APIDOC
async_open Helper Methods:

* `async def read(length = -1)`
  - Reads a chunk from the file. If length is -1, reads to the end.

* `async def write(data)`
  - Writes a chunk to the file.

* `def seek(offset)`
  - Sets the file pointer position.

* `def tell()`
  - Returns the current file pointer position.

* `async def readline(size=-1, newline="\n")`
  - Reads chunks until newline or EOF. Suboptimal for small lines; use LineReader for line-by-line reading.

* `def __aiter__() -> LineReader`
  - Iterator over lines.

* `def iter_chunked(chunk_size: int = 32768) -> Reader`
  - Iterator over chunks.

* `.file` property
  - Contains the underlying AIOFile object.
```

--------------------------------

### Manually Manage caio Contexts with aiofile in Python

Source: https://github.com/mosquito/aiofile/blob/master/README.md

Demonstrates how to explicitly create and use `AsyncioContext` instances for different `caio` implementations (like `linux` and `thread`) when opening files with `aiofile`. This allows for fine-grained control over the underlying asynchronous I/O mechanism.

```python
import asyncio

from aiofile import async_open
from caio import linux_aio_asyncio, thread_aio_asyncio


async def main():
    linux_ctx = linux_aio_asyncio.AsyncioContext()
    threads_ctx = thread_aio_asyncio.AsyncioContext()

    async with async_open("/tmp/test.txt", "w", context=linux_ctx) as afp:
        await afp.write("Hello")

    async with async_open("/tmp/test.txt", "r", context=threads_ctx) as afp:
        print(await afp.read())


asyncio.run(main())
```

--------------------------------

### Special Files with Custom Context

Source: https://github.com/mosquito/aiofile/blob/master/README.md

Demonstrates handling special files (like /proc/cpuinfo) that may not support standard asynchronous I/O by using a custom context provided by `caio.thread_aio_asyncio.AsyncioContext`.

```Python
import asyncio
from aiofile import async_open
from caio import thread_aio_asyncio
from contextlib import AsyncExitStack


async def main():
    async with AsyncExitStack() as stack:

        # Custom context should be reused
        ctx = await stack.enter_async_context(
            thread_aio_asyncio.AsyncioContext()
        )

        # Open special file with custom context
        src = await stack.enter_async_context(
            async_open("/proc/cpuinfo", "r", context=ctx)
        )

        # Open regular file with default context
        dest = await stack.enter_async_context(
            async_open("/tmp/cpuinfo", "w")
        )

        # Copying file content line by line
        async for line in src:
            await dest.write(line)


asyncio.run(main())
```

--------------------------------

### Opening Already Open File Pointer

Source: https://github.com/mosquito/aiofile/blob/master/README.md

Illustrates using `async_open` with an already opened standard Python file object (`IO[Any]`). It writes and reads lines asynchronously to the provided file pointer.

```Python
import asyncio
from typing import IO, Any
from aiofile import async_open


async def main(fp: IO[Any]):
    async with async_open(fp) as afp:
        await afp.write("Hello from\nasync world")
        print(await afp.readline())


with open("test.txt", "w+") as fp:
    asyncio.run(main(fp))
```

--------------------------------

### Concurrent Writes with Offsets using AIOFile

Source: https://github.com/mosquito/aiofile/blob/master/README.md

Demonstrates writing multiple data chunks to a file concurrently at specified byte offsets using the `AIOFile.write` method. Includes `fsync` for data persistence and verification of read content.

```python
import asyncio
from aiofile import AIOFile


async def main():
    async with AIOFile("hello.txt", 'w+') as afp:
        payload = "Hello world\n"

        await asyncio.gather(
            *[afp.write(payload, offset=i * len(payload)) for i in range(10)]
        )

        await afp.fsync()

        assert await afp.read(len(payload) * 10) == payload * 10

asyncio.run(main())
```

--------------------------------

### Line-by-Line Reading with LineReader

Source: https://github.com/mosquito/aiofile/blob/master/README.md

Demonstrates using `LineReader` to read a file line by line asynchronously. This is efficient for processing text files where lines are delimited by newline characters.

```python
import asyncio
from aiofile import AIOFile, LineReader, Writer


async def main():
    async with AIOFile("/tmp/hello.txt", 'w+') as afp:
        writer = Writer(afp)

        await writer("Hello")
        await writer(" ")
        await writer("World")
        await writer("\n")
        await writer("\n")
        await writer("From async world")
        await afp.fsync()

        async for line in LineReader(afp):
            print(line)


asyncio.run(main())
```

--------------------------------

### Sequential Reading and Writing with Reader/Writer

Source: https://github.com/mosquito/aiofile/blob/master/README.md

Utilizes `Reader` and `Writer` classes for sequential file operations. The `Writer` writes data, and the `Reader` iterates over the file content in chunks.

```python
import asyncio
from aiofile import AIOFile, Reader, Writer


async def main():
    async with AIOFile("/tmp/hello.txt", 'w+') as afp:
        writer = Writer(afp)
        reader = Reader(afp, chunk_size=8)

        await writer("Hello")
        await writer(" ")
        await writer("World")
        await afp.fsync()

        async for chunk in reader:
            print(chunk)


asyncio.run(main())
```

--------------------------------

### AsyncDictReader for CSV Files

Source: https://github.com/mosquito/aiofile/blob/master/README.md

Provides an asynchronous dictionary reader for CSV files, leveraging `LineReader` and Python's `csv.DictReader`. It handles reading lines, parsing them into dictionaries, and managing internal buffers.

```python
import asyncio
import io
from csv import DictReader

from aiofile import AIOFile, LineReader


class AsyncDictReader:
    def __init__(self, afp, **kwargs):
        self.buffer = io.BytesIO()
        self.file_reader = LineReader(
            afp, line_sep=kwargs.pop('line_sep', '\n'),
            chunk_size=kwargs.pop('chunk_size', 4096),
            offset=kwargs.pop('offset', 0),
        )
        self.reader = DictReader(
            io.TextIOWrapper(
                self.buffer,
                encoding=kwargs.pop('encoding', 'utf-8'),
                errors=kwargs.pop('errors', 'replace'),
            ), **kwargs,
        )
        self.line_num = 0

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self.line_num == 0:
            header = await self.file_reader.readline()
            self.buffer.write(header)

        line = await self.file_reader.readline()

        if not line:
            raise StopAsyncIteration

        self.buffer.write(line)
        self.buffer.seek(0)

        try:
            result = next(self.reader)
        except StopIteration as e:
            raise StopAsyncIteration from e

        self.buffer.seek(0)
        self.buffer.truncate(0)
        self.line_num = self.reader.line_num

        return result


async def main():
    async with AIOFile('sample.csv', 'rb') as afp:
        async for item in AsyncDictReader(afp):
            print(item)


asyncio.run(main())
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.