Source code for aiida.cmdline.params.types.path

###########################################################################
# Copyright (c), The AiiDA team. All rights reserved.                     #
# This file is part of the AiiDA code.                                    #
#                                                                         #
# The code is hosted on GitHub at https://github.com/aiidateam/aiida-core #
# For further information on the license, see the LICENSE.txt file        #
# For further information please visit http://www.aiida.net               #
###########################################################################
"""Click parameter types for paths."""
import os

import click

__all__ = ('AbsolutePathParamType', 'FileOrUrl', 'PathOrUrl')

URL_TIMEOUT_SECONDS = 10


[docs] def check_timeout_seconds(timeout_seconds): """Raise if timeout is not within range [0;60]""" try: timeout_seconds = int(timeout_seconds) except ValueError: raise TypeError(f'timeout_seconds should be an integer but got: {type(timeout_seconds)}') if timeout_seconds < 0 or timeout_seconds > 60: raise ValueError('timeout_seconds needs to be in the range [0;60].') return timeout_seconds
[docs] class AbsolutePathParamType(click.Path): """The ParamType for identifying absolute Paths (derived from click.Path).""" name = 'AbsolutePath'
[docs] def convert(self, value, param, ctx): value = os.path.expanduser(value) newval = super().convert(value, param, ctx) if not os.path.isabs(newval): raise click.BadParameter('path must be absolute') return newval
[docs] def __repr__(self): return 'ABSOLUTEPATH'
[docs] class AbsolutePathOrEmptyParamType(AbsolutePathParamType): """The ParamType for identifying absolute Paths, accepting also empty paths.""" name = 'AbsolutePathEmpty'
[docs] def convert(self, value, param, ctx): if not value: return value return super().convert(value, param, ctx)
[docs] def __repr__(self): return 'ABSOLUTEPATHEMPTY'
[docs] class PathOrUrl(click.Path): """Extension of click's Path-type to include URLs. A PathOrUrl can either be a `click.Path`-type or a URL. :param int timeout_seconds: Maximum timeout accepted for URL response. Must be an integer in the range [0;60]. """ name = 'PathOrUrl'
[docs] def __init__(self, timeout_seconds=URL_TIMEOUT_SECONDS, **kwargs): super().__init__(**kwargs) self.timeout_seconds = check_timeout_seconds(timeout_seconds)
[docs] def convert(self, value, param, ctx): """Overwrite `convert` Check first if `click.Path`-type, then check if URL.""" try: return super().convert(value, param, ctx) except click.exceptions.BadParameter: return self.checks_url(value, param, ctx)
[docs] def checks_url(self, url, param, ctx): """Check whether URL is reachable within timeout.""" import socket import urllib.error import urllib.request try: with urllib.request.urlopen(url, timeout=self.timeout_seconds): pass except (urllib.error.URLError, urllib.error.HTTPError, socket.timeout): self.fail(f'{self.name} "{url}" could not be reached within {self.timeout_seconds} s.\n', param, ctx) return url
[docs] class FileOrUrl(click.File): """Extension of click's File-type to include URLs. Returns handle either to local file or to remote file fetched from URL. :param int timeout_seconds: Maximum timeout accepted for URL response. Must be an integer in the range [0;60]. """ name = 'FileOrUrl'
[docs] def __init__(self, timeout_seconds=URL_TIMEOUT_SECONDS, **kwargs): super().__init__(**kwargs) self.timeout_seconds = check_timeout_seconds(timeout_seconds)
[docs] def convert(self, value, param, ctx): """Return file handle.""" try: return super().convert(value, param, ctx) except click.exceptions.BadParameter: handle = self.get_url(value, param, ctx) return handle
[docs] def get_url(self, url, param, ctx): """Retrieve file from URL.""" import socket import urllib.error import urllib.request try: return urllib.request.urlopen(url, timeout=self.timeout_seconds) except (urllib.error.URLError, urllib.error.HTTPError, socket.timeout): self.fail(f'{self.name} "{url}" could not be reached within {self.timeout_seconds} s.\n', param, ctx)