How to use _check_kwargs method in pandera

Best Python code snippet using pandera_python

file_io.py

Source:file_io.py

...37    protocol. It routes I/O for a generic URI which may look like "protocol://*"38    or a canonical filepath "/foo/bar/baz".39    """40    _strict_kwargs_check = True41    def _check_kwargs(self, kwargs: Dict[str, Any]) -> None:42        """43        Checks if the given arguments are empty. Throws a ValueError if strict44        kwargs checking is enabled and args are non-empty. If strict kwargs45        checking is disabled, only a warning is logged.46        Args:47            kwargs (Dict[str, Any])48        """49        if self._strict_kwargs_check:50            if len(kwargs) > 0:51                raise ValueError("Unused arguments: {}".format(kwargs))52        else:53            logger = logging.getLogger(__name__)54            for k, v in kwargs.items():55                logger.warning(56                    "[PathManager] {}={} argument ignored".format(k, v)57                )58    def _get_supported_prefixes(self) -> List[str]:59        """60        Returns:61            List[str]: the list of URI prefixes this PathHandler can support62        """63        raise NotImplementedError()64    def _get_local_path(self, path: str, **kwargs: Any) -> str:65        """66        Get a filepath which is compatible with native Python I/O such as `open`67        and `os.path`.68        If URI points to a remote resource, this function may download and cache69        the resource to local disk. In this case, this function is meant to be70        used with read-only resources.71        Args:72            path (str): A URI supported by this PathHandler73        Returns:74            local_path (str): a file path which exists on the local file system75        """76        raise NotImplementedError()77    def _open(78            self, path: str, mode: str = "r", buffering: int = -1, **kwargs: Any79    ) -> Union[IO[str], IO[bytes]]:80        """81        Open a stream to a URI, similar to the built-in `open`.82        Args:83            path (str): A URI supported by this PathHandler84            mode (str): Specifies the mode in which the file is opened. It defaults85                to 'r'.86            buffering (int): An optional integer used to set the buffering policy.87                Pass 0 to switch buffering off and an integer >= 1 to indicate the88                size in bytes of a fixed-size chunk buffer. When no buffering89                argument is given, the default buffering policy depends on the90                underlying I/O implementation.91        Returns:92            file: a file-like object.93        """94        raise NotImplementedError()95    def _copy(96            self,97            src_path: str,98            dst_path: str,99            overwrite: bool = False,100            **kwargs: Any,101    ) -> bool:102        """103        Copies a source path to a destination path.104        Args:105            src_path (str): A URI supported by this PathHandler106            dst_path (str): A URI supported by this PathHandler107            overwrite (bool): Bool flag for forcing overwrite of existing file108        Returns:109            status (bool): True on success110        """111        raise NotImplementedError()112    def _exists(self, path: str, **kwargs: Any) -> bool:113        """114        Checks if there is a resource at the given URI.115        Args:116            path (str): A URI supported by this PathHandler117        Returns:118            bool: true if the path exists119        """120        raise NotImplementedError()121    def _isfile(self, path: str, **kwargs: Any) -> bool:122        """123        Checks if the resource at the given URI is a file.124        Args:125            path (str): A URI supported by this PathHandler126        Returns:127            bool: true if the path is a file128        """129        raise NotImplementedError()130    def _isdir(self, path: str, **kwargs: Any) -> bool:131        """132        Checks if the resource at the given URI is a directory.133        Args:134            path (str): A URI supported by this PathHandler135        Returns:136            bool: true if the path is a directory137        """138        raise NotImplementedError()139    def _ls(self, path: str, **kwargs: Any) -> List[str]:140        """141        List the contents of the directory at the provided URI.142        Args:143            path (str): A URI supported by this PathHandler144        Returns:145            List[str]: list of contents in given path146        """147        raise NotImplementedError()148    def _mkdirs(self, path: str, **kwargs: Any) -> None:149        """150        Recursive directory creation function. Like mkdir(), but makes all151        intermediate-level directories needed to contain the leaf directory.152        Similar to the native `os.makedirs`.153        Args:154            path (str): A URI supported by this PathHandler155        """156        raise NotImplementedError()157    def _rm(self, path: str, **kwargs: Any) -> None:158        """159        Remove the file (not directory) at the provided URI.160        Args:161            path (str): A URI supported by this PathHandler162        """163        raise NotImplementedError()164class NativePathHandler(PathHandler):165    """166    Handles paths that can be accessed using Python native system calls. This167    handler uses `open()` and `os.*` calls on the given path.168    """169    def _get_local_path(self, path: str, **kwargs: Any) -> str:170        self._check_kwargs(kwargs)171        return path172    def _open(173            self,174            path: str,175            mode: str = "r",176            buffering: int = -1,177            encoding: Optional[str] = None,178            errors: Optional[str] = None,179            newline: Optional[str] = None,180            closefd: bool = True,181            opener: Optional[Callable] = None,182            **kwargs: Any,183    ) -> Union[IO[str], IO[bytes]]:184        """185        Open a path.186        Args:187            path (str): A URI supported by this PathHandler188            mode (str): Specifies the mode in which the file is opened. It defaults189                to 'r'.190            buffering (int): An optional integer used to set the buffering policy.191                Pass 0 to switch buffering off and an integer >= 1 to indicate the192                size in bytes of a fixed-size chunk buffer. When no buffering193                argument is given, the default buffering policy works as follows:194                    * Binary files are buffered in fixed-size chunks; the size of195                    the buffer is chosen using a heuristic trying to determine the196                    underlying deviceâs âblock sizeâ and falling back on197                    io.DEFAULT_BUFFER_SIZE. On many systems, the buffer will198                    typically be 4096 or 8192 bytes long.199            encoding (Optional[str]): the name of the encoding used to decode or200                encode the file. This should only be used in text mode.201            errors (Optional[str]): an optional string that specifies how encoding202                and decoding errors are to be handled. This cannot be used in binary203                mode.204            newline (Optional[str]): controls how universal newlines mode works205                (it only applies to text mode). It can be None, '', '\n', '\r',206                and '\r\n'.207            closefd (bool): If closefd is False and a file descriptor rather than208                a filename was given, the underlying file descriptor will be kept209                open when the file is closed. If a filename is given closefd must210                be True (the default) otherwise an error will be raised.211            opener (Optional[Callable]): A custom opener can be used by passing212                a callable as opener. The underlying file descriptor for the file213                object is then obtained by calling opener with (file, flags).214                opener must return an open file descriptor (passing os.open as opener215                results in functionality similar to passing None).216            See https://docs.python.org/3/library/functions.html#open for details.217        Returns:218            file: a file-like object.219        """220        self._check_kwargs(kwargs)221        return open(  # type: ignore222            path,223            mode,224            buffering=buffering,225            encoding=encoding,226            errors=errors,227            newline=newline,228            closefd=closefd,229            opener=opener,230        )231    def _copy(232            self,233            src_path: str,234            dst_path: str,235            overwrite: bool = False,236            **kwargs: Any,237    ) -> bool:238        """239        Copies a source path to a destination path.240        Args:241            src_path (str): A URI supported by this PathHandler242            dst_path (str): A URI supported by this PathHandler243            overwrite (bool): Bool flag for forcing overwrite of existing file244        Returns:245            status (bool): True on success246        """247        self._check_kwargs(kwargs)248        if os.path.exists(dst_path) and not overwrite:249            logger = logging.getLogger(__name__)250            logger.error("Destination file {} already exists.".format(dst_path))251            return False252        try:253            shutil.copyfile(src_path, dst_path)254            return True255        except Exception as e:256            logger = logging.getLogger(__name__)257            logger.error("Error in file copy - {}".format(str(e)))258            return False259    def _exists(self, path: str, **kwargs: Any) -> bool:260        self._check_kwargs(kwargs)261        return os.path.exists(path)262    def _isfile(self, path: str, **kwargs: Any) -> bool:263        self._check_kwargs(kwargs)264        return os.path.isfile(path)265    def _isdir(self, path: str, **kwargs: Any) -> bool:266        self._check_kwargs(kwargs)267        return os.path.isdir(path)268    def _ls(self, path: str, **kwargs: Any) -> List[str]:269        self._check_kwargs(kwargs)270        return os.listdir(path)271    def _mkdirs(self, path: str, **kwargs: Any) -> None:272        self._check_kwargs(kwargs)273        try:274            os.makedirs(path, exist_ok=True)275        except OSError as e:276            # EEXIST it can still happen if multiple processes are creating the dir277            if e.errno != errno.EEXIST:278                raise279    def _rm(self, path: str, **kwargs: Any) -> None:280        self._check_kwargs(kwargs)281        os.remove(path)282class PathManager:283    """284    A class for users to open generic paths or translate generic paths to file names.285    """286    _PATH_HANDLERS: MutableMapping[str, PathHandler] = OrderedDict()287    _NATIVE_PATH_HANDLER = NativePathHandler()288    @staticmethod289    def __get_path_handler(path: str) -> PathHandler:290        """291        Finds a PathHandler that supports the given path. Falls back to the native292        PathHandler if no other handler is found.293        Args:294            path (str): URI path to resource...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.