aiopathlib: Pathlib support for asyncio
aiopathlib is written in Python, for handling local disk files in asyncio applications.
Base on aiofiles and just like pathlib, but use await.
with open('filename', 'w') as fp:
fp.write('My file contents')
text = await aiopathlib.AsyncPath('filename').read_text()
print(text)
'My file contents'
content = await aiopathlib.AsyncPath(Path('filename')).read_bytes()
print(content)
b'My file contents'Asynchronous interface to create folder.
from aiopathlib import AsyncPath
apath = AsyncPath('dirname/subpath')
if not await apath.exists():
await apath.mkdir(parents=True)Features
- a file API very similar to Python's standard package
pathlib, blocking API - support for buffered and unbuffered binary files, and buffered text files
- support for
async/await(:PEP:492) constructs
Installation
To install aiopathlib, simply:
$ pip install aiopathlibUsage
These functions are awaitable
read_textread_bytesread_jsonwrite_textwrite_byteswrite_jsonmkdirtouchexistsrenameunlinkrmdirremovestatlstatis_fileis_diris_symlinkis_fifois_mountis_block_deviceis_char_deviceis_socket
Example
Some common using cases:
from pashlib import Path
from aiopathlib import AsyncPath
filename = 'test.json'
ap = AsyncPath(filename)
p = Path(filename)
assert (await ap.exists()) == p.exists() == False
await ap.touch() # Create a empty file
assert (await ap.is_file()) == p.is_file() == True
assert (await ap.is_dir()) == p.is_dir() == False
assert (await ap.is_symlink()) == p.is_symlink() == False
for func in ('is_fifo', 'is_mount', 'is_block_device', 'is_char_device', 'is_socket'):
assert (await getattr(ap, func)()) == getattr(p, func)()
d = {'key': 'value'}
await ap.write_json(d) # == p.write_text(json.dumps(d))
assert (await ap.read_json()) == d # == json.loads(p.read_text())
assert (await ap.read_bytes()) == p.read_bytes() # b'{"key": "value"}'
assert (await ap.stat()) == p.stat()
assert (await ap.lstat()) == p.lstat()
ap = await ap.rename('test_dir') # == AsyncPath(p.rename('test_dir'))
await ap.remove() # == await ap.unlink() == p.unlink()
await ap.mkdir() # == p.mkdir()
# Synchronization functions
[Path(i) for i in ap.glob('*')] == list(p.glob('*'))
[Path(i) for i in ap.rglob('*')] == list(p.rglob('*'))
ap / 'filename' == ap.joinpath('filename') == AsyncPath(f'{ap}/filename')
str(AsyncPath('string-or-Path-or-AsyncPath')) == str(Path('string-or-Path-or-AsyncPath'))
ap.stem == p.stem
ap.suffix == p.suffix
Path(ap.with_name('xxx')) == p.with_name('xxx')
Path(ap.parent) == p.parent
Path(ap.resolve()) == p.resolve()
...
History
0.3.1 (2022-02-20)
- Return content size after write local file
- Upgrade dependencies
0.3.0 (2021-12-16)
- Support Python3.7
- Clear
dev_requirements.txtto be only package name and version
0.2.3 (2021-10-16)
- Make
touchpass test for py39. - Remove support for pypy3 from docs.
0.2.2 (2021-09-20)
- Make
touch/stat/is_file/... be awaitable. - Use
super().__new__for initial.
0.2.0 (2021-08-29)
- Make
AsyncPathbe subclass ofpathlib.Path. - Add github action to show test coverage.
0.1.3 (2021-08-28)
- Add makefile.
- Test all functions.
- Fix rename method error.
- Support sync pathlib methods.
0.1.0 (2021-06-14)
- Introduced a changelog.
- Publish at gitee.
Contributing
Contributions are very welcome.
