Real asynchronous file operations with asyncio support.
Status
Development - Stable
Features
Since version 2.0.0 using caio, is contain linux libaio and two thread-based implementations (c-based and pure-python).
AIOFile has no internal pointer. You should pass offset and chunk_size for each operation or use helpers (Reader or Writer).
For Linux using implementation based on libaio.
For POSIX (MacOS X and optional Linux) using implementation using on threadpool.
Otherwise using pure-python thread-based implementation.
Implementation chooses automatically depending on system compatibility.
Code examples
All code examples requires python 3.6+.
High-level API
async_open helper
The async_open helper creates file like object with file-like methods:
import asyncio
from aiofile import async_open
async def main():
async with async_open("/tmp/hello.txt", 'w+') as afp:
await afp.write("Hello ")
await afp.write("world")
afp.seek(0)
print(await afp.read())
loop = asyncio.get_event_loop()
loop.run_until_complete(main())Supported methods:
async def read(length = -1) - reading chunk from file, when length is -1 will read whole file.
async def write(data) - write chunk to file
def seek(offset) - set file pointer position
def tell() - returns current file pointer position
Reader and Writer
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)
loop = asyncio.get_event_loop()
loop.run_until_complete(main())LineReader - read file line by line
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)
loop = asyncio.get_event_loop()
loop.run_until_complete(main())Low-level API
Following API is just little bit sugared caio API.
Write and Read
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())
loop = asyncio.get_event_loop()
loop.run_until_complete(main())Read file line by line
import asyncio
from aiofile import AIOFile, LineReader, Writer
async def main():
async with AIOFile("/tmp/hello.txt", 'w') as afp:
writer = Writer(afp)
for i in range(10):
await writer("%d Hello World\n" % i)
await writer("Tail-less string")
async with AIOFile("/tmp/hello.txt", 'r') as afp:
async for line in LineReader(afp):
print(line[:-1])
loop = asyncio.get_event_loop()
loop.run_until_complete(main())More examples
Useful examples with aiofile
Async CSV Dict Reader
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, line_sep='\r'):
print(item)
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
loop.run_until_complete(main())
finally:
# Shutting down and closing file descriptors after interrupt
loop.run_until_complete(loop.shutdown_asyncgens())
loop.close()1 maintainer
Metadata
1 maintainer
Release files for aiofile 3.2.0
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| aiofile-3.2.0.tar.gz | 10.1 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| aiofile-3.2.0-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 26.4 kB
Release files / aiofile-3.2.0.tar.gz
| Download URL | aiofile-3.2.0.tar.gz |
|---|---|
| Size | 10.1 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
1b044e3b84bacc58c7eb84d85cab79de982b11ce942d33edada04d0a675a6fa8
|
|
BLAKE2b-256 checksum How to use checksums |
4fce0273a0e86a1b32ff6778d57ec1a48ed1010da85b28d96169b351a2f2cdf7
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/3.1.1 pkginfo/1.5.0.1 requests/2.23.0 setuptools/41.2.0 requests-toolbelt/0.9.1 tqdm/4.45.0 CPython/3.8.3
|
Release files / aiofile-3.2.0-py3-none-any.whl
| Download URL | aiofile-3.2.0-py3-none-any.whl |
|---|---|
| Size | 16.2 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
5e3793e1e267ef2e7e142ef3c5c3932433d7109d0b896eb9105b46aee4525fd0
|
|
BLAKE2b-256 checksum How to use checksums |
384022dee20f733c445aae3f9f5367a0b92a0e13ce0a0c2bd1735fcbb2f8ff32
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/3.1.1 pkginfo/1.5.0.1 requests/2.23.0 setuptools/41.2.0 requests-toolbelt/0.9.1 tqdm/4.45.0 CPython/3.8.3
|
