aiopath 0.4.5

📁 Async pathlib for Python

📁 Async pathlib for Python

aiopath is a complete implementation of Python's pathlib that is compatible with asyncio and Python's async/await syntax. All I/O performed by aiopath is asynchronous and awaitable.

aiopath takes advantage of Python type annotations, and it also takes advantage of libaio on Linux.

Use case

If you're writing asynchronous code and want to take advantage of pathlib's conveniences, but don't want to mix blocking and non-blocking I/O, then you can reach for aiopath.

As an example, if you're writing an asynchronous scraper, you might want to make several concurrent requests to websites, and then write the results of those requests to secondary storage:

from typing import List
from asyncio import run, gather

from aiohttp import ClientSession
from aiopath import AsyncPath


async def save_page(url: str, name: str):
  async with ClientSession() as session:
    response = await session.get(url)
    content: bytes = await response.read()

  path = AsyncPath(name)

  if not await path.exists():
    await path.write_bytes(content)


async def main():
  urls: List[str] = [
    'https://example.com',
    'https://github.com/alexdelorenzo/aiopath',
    'https://alexdelorenzo.dev'
  ]

  scrapers = (save_page(url, f"{index}.html") for index, url in enumerate(urls))
  return await gather(*scrapers)


run(main())

If you used pathlib instead of aiopath in the example above, tasks would block upon writing to the disk, and tasks that make network connections would be forced to pause while other tasks write to the disk.

By using aiopath, all I/O is non-blocking, and your script can simultaneously write to the disk and perform network operations at once.

Usage

aiopath is a direct reimplementation of CPython's pathlib.py, and aiopath's class hierarchy directly matches the one from pathlib, where Path inherits from PurePath, AsyncPath inherits from AsyncPurePath, and so on.

With aiopath, methods that perform I/O are asynchronous and awaitable, and methods that perform I/O and return iterators in pathlib now return async generators.

Examples

Running examples

To run the following examples with top-level await expressions, launch an asynchronous Python REPL using python3 -m asyncio.

You'll also need to install asynctempfile via PyPI, like so python3 -m pip install asynctempfile

Basic

All of pathlib.Path's methods that perform synchronous I/O are reimplemented as asynchronous methods. PurePath methods are not asynchronous because they don't perform I/O.

from pathlib import Path

from asynctempfile import NamedTemporaryFile
from aiopath import AsyncPath


async with NamedTemporaryFile() as temp:
  path = Path(temp.name)
  apath = AsyncPath(temp.name)

  # check existence
  ## sync
  assert path.exists()
  ## async
  assert await apath.exists()

  # check if file
  ## sync
  assert path.is_file()
  ## async
  assert await apath.is_file()

  # touch
  path.touch()
  await apath.touch()

  # PurePath methods are not async
  assert path.is_absolute() == apath.is_absolute()
  assert path.as_uri() == apath.as_uri()

  # read and write text
  text: str = "example"
  await apath.write_text(text)
  assert text == await apath.read_text()

assert not path.exists()
assert not await apath.exists()

You can convert pathlib.Path objects to aiopath.AsyncPath objects, and vice versa:

from pathlib import Path
from aiopath import AsyncPath


home: Path = Path.home()
ahome: AsyncPath = AsyncPath(home)
path: Path = Path(ahome)

assert isinstance(home, Path)
assert isinstance(ahome, AsyncPath)
assert isinstance(path, Path)
assert str(home) == str(ahome) == str(path)

Opening a file

You can get an asynchronous file-like object handle by using asynchronous context managers.

from asynctempfile import NamedTemporaryFile
from aiopath import AsyncPath


text: str = 'example'

async with NamedTemporaryFile() as temp:
  apath = AsyncPath(temp.name)

  async with apath.open(mode='w') as afile:
    await afile.write(text)

  assert text == await apath.read_text()

Globbing

aiopath implements pathlib globbing using async I/O and async generators.

from typing import List
from aiopath import AsyncPath


home: AsyncPath = await AsyncPath.home()

async for path in home.glob('*'):
  assert isinstance(path, AsyncPath)
  print(path)

downloads: AsyncPath = home / 'Downloads'

if await downloads.exists():
  # this might take a while
  paths: List[AsyncPath] = \
    [path async for path in downloads.glob('**/*')]

Installation

Dependencies

• A POSIX compliant OS, or Windows
• Python 3.7+
• requirements.txt

Linux dependencies

If you're using a 4.18 or newer kernel and have libaio installed, aiopath will use it via aiofile. You can install libaio on Debian/Ubuntu like so:

$ sudo apt install libaio1

PyPI

$ python3 -m pip install aiopath

GitHub

$ python3 -m pip install -r requirements.txt
$ python3 setup.py install

Support

Want to support this project and other open-source projects like it?

License

See LICENSE. If you'd like to use this project with a different license, please get in touch.

Credit

See CREDIT.md.