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...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!