| | |
| | import os |
| | import tempfile |
| | from contextlib import contextmanager |
| | from pathlib import Path |
| | from typing import Generator, Union |
| | from urllib.request import urlopen |
| |
|
| | from .base import BaseStorageBackend |
| |
|
| |
|
| | class HTTPBackend(BaseStorageBackend): |
| | """HTTP and HTTPS storage bachend.""" |
| |
|
| | def get(self, filepath: str) -> bytes: |
| | """Read bytes from a given ``filepath``. |
| | |
| | Args: |
| | filepath (str): Path to read data. |
| | |
| | Returns: |
| | bytes: Expected bytes object. |
| | |
| | Examples: |
| | >>> backend = HTTPBackend() |
| | >>> backend.get('http://path/of/file') |
| | b'hello world' |
| | """ |
| | return urlopen(filepath).read() |
| |
|
| | def get_text(self, filepath, encoding='utf-8') -> str: |
| | """Read text from a given ``filepath``. |
| | |
| | Args: |
| | filepath (str): Path to read data. |
| | encoding (str): The encoding format used to open the ``filepath``. |
| | Defaults to 'utf-8'. |
| | |
| | Returns: |
| | str: Expected text reading from ``filepath``. |
| | |
| | Examples: |
| | >>> backend = HTTPBackend() |
| | >>> backend.get_text('http://path/of/file') |
| | 'hello world' |
| | """ |
| | return urlopen(filepath).read().decode(encoding) |
| |
|
| | @contextmanager |
| | def get_local_path( |
| | self, filepath: str) -> Generator[Union[str, Path], None, None]: |
| | """Download a file from ``filepath`` to a local temporary directory, |
| | and return the temporary path. |
| | |
| | ``get_local_path`` is decorated by :meth:`contxtlib.contextmanager`. It |
| | can be called with ``with`` statement, and when exists from the |
| | ``with`` statement, the temporary path will be released. |
| | |
| | Args: |
| | filepath (str): Download a file from ``filepath``. |
| | |
| | Yields: |
| | Iterable[str]: Only yield one temporary path. |
| | |
| | Examples: |
| | >>> backend = HTTPBackend() |
| | >>> # After existing from the ``with`` clause, |
| | >>> # the path will be removed |
| | >>> with backend.get_local_path('http://path/of/file') as path: |
| | ... # do something here |
| | """ |
| | try: |
| | f = tempfile.NamedTemporaryFile(delete=False) |
| | f.write(self.get(filepath)) |
| | f.close() |
| | yield f.name |
| | finally: |
| | os.remove(f.name) |
| |
|