| # |
| # Licensed to the Apache Software Foundation (ASF) under one or more |
| # contributor license agreements. See the NOTICE file distributed with |
| # this work for additional information regarding copyright ownership. |
| # The ASF licenses this file to You under the Apache License, Version 2.0 |
| # (the "License"); you may not use this file except in compliance with |
| # the License. You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| # |
| |
| """Local File system implementation for accessing files on disk.""" |
| |
| # pytype: skip-file |
| |
| import io |
| import os |
| import shutil |
| from typing import BinaryIO # pylint: disable=unused-import |
| |
| from apache_beam.io.filesystem import BeamIOError |
| from apache_beam.io.filesystem import CompressedFile |
| from apache_beam.io.filesystem import CompressionTypes |
| from apache_beam.io.filesystem import FileMetadata |
| from apache_beam.io.filesystem import FileSystem |
| |
| __all__ = ['LocalFileSystem'] |
| |
| |
| class LocalFileSystem(FileSystem): |
| """A Local ``FileSystem`` implementation for accessing files on disk. |
| """ |
| @classmethod |
| def scheme(cls): |
| """URI scheme for the FileSystem |
| """ |
| return None |
| |
| def join(self, basepath, *paths): |
| """Join two or more pathname components for the filesystem |
| |
| Args: |
| basepath: string path of the first component of the path |
| paths: path components to be added |
| |
| Returns: full path after combining all the passed components |
| """ |
| return os.path.join(basepath, *paths) |
| |
| def split(self, path): |
| """Splits the given path into two parts. |
| |
| Splits the path into a pair (head, tail) such that tail contains the last |
| component of the path and head contains everything up to that. |
| |
| Args: |
| path: path as a string |
| Returns: |
| a pair of path components as strings. |
| """ |
| return os.path.split(os.path.abspath(path)) |
| |
| def mkdirs(self, path): |
| """Recursively create directories for the provided path. |
| |
| Args: |
| path: string path of the directory structure that should be created |
| |
| Raises: |
| IOError: if leaf directory already exists. |
| """ |
| try: |
| os.makedirs(path) |
| except OSError as err: |
| raise IOError(err) |
| |
| def has_dirs(self): |
| """Whether this FileSystem supports directories.""" |
| return True |
| |
| def _url_dirname(self, url_or_path): |
| """Pass through to os.path.dirname. |
| |
| This version uses os.path instead of posixpath to be compatible with the |
| host OS. |
| |
| Args: |
| url_or_path: A string in the form of /some/path. |
| """ |
| return os.path.dirname(url_or_path) |
| |
| def _list(self, dir_or_prefix): |
| """List files in a location. |
| |
| Listing is non-recursive, for filesystems that support directories. |
| |
| Args: |
| dir_or_prefix: (string) A directory or location prefix (for filesystems |
| that don't have directories). |
| |
| Returns: |
| Generator of ``FileMetadata`` objects. |
| |
| Raises: |
| ``BeamIOError``: if listing fails, but not if no files were found. |
| """ |
| if not self.exists(dir_or_prefix): |
| return |
| |
| def list_files(root): |
| for dirpath, _, files in os.walk(root): |
| for filename in files: |
| yield self.join(dirpath, filename) |
| |
| try: |
| for f in list_files(dir_or_prefix): |
| try: |
| yield FileMetadata(f, os.path.getsize(f)) |
| except OSError: |
| # Files may disappear, such as when listing /tmp. |
| pass |
| except Exception as e: # pylint: disable=broad-except |
| raise BeamIOError("List operation failed", {dir_or_prefix: e}) |
| |
| def _path_open( |
| self, |
| path, |
| mode, |
| mime_type='application/octet-stream', |
| compression_type=CompressionTypes.AUTO): |
| """Helper functions to open a file in the provided mode. |
| """ |
| compression_type = FileSystem._get_compression_type(path, compression_type) |
| raw_file = io.open(path, mode) |
| if compression_type == CompressionTypes.UNCOMPRESSED: |
| return raw_file |
| else: |
| return CompressedFile(raw_file, compression_type=compression_type) |
| |
| def create( |
| self, |
| path, |
| mime_type='application/octet-stream', |
| compression_type=CompressionTypes.AUTO): |
| # type: (...) -> BinaryIO |
| |
| """Returns a write channel for the given file path. |
| |
| Args: |
| path: string path of the file object to be written to the system |
| mime_type: MIME type to specify the type of content in the file object |
| compression_type: Type of compression to be used for this object |
| |
| Returns: file handle with a close function for the user to use |
| """ |
| os.makedirs(os.path.dirname(path), exist_ok=True) |
| return self._path_open(path, 'wb', mime_type, compression_type) |
| |
| def open( |
| self, |
| path, |
| mime_type='application/octet-stream', |
| compression_type=CompressionTypes.AUTO): |
| # type: (...) -> BinaryIO |
| |
| """Returns a read channel for the given file path. |
| |
| Args: |
| path: string path of the file object to be written to the system |
| mime_type: MIME type to specify the type of content in the file object |
| compression_type: Type of compression to be used for this object |
| |
| Returns: file handle with a close function for the user to use |
| """ |
| return self._path_open(path, 'rb', mime_type, compression_type) |
| |
| def copy(self, source_file_names, destination_file_names): |
| """Recursively copy the file tree from the source to the destination |
| |
| Args: |
| source_file_names: list of source file objects that needs to be copied |
| destination_file_names: list of destination of the new object |
| |
| Raises: |
| ``BeamIOError``: if any of the copy operations fail |
| """ |
| err_msg = ( |
| "source_file_names and destination_file_names should " |
| "be equal in length") |
| assert len(source_file_names) == len(destination_file_names), err_msg |
| |
| def _copy_path(source, destination): |
| """Recursively copy the file tree from the source to the destination |
| """ |
| try: |
| if os.path.exists(destination): |
| if os.path.isdir(destination): |
| shutil.rmtree(destination) |
| else: |
| os.remove(destination) |
| if os.path.isdir(source): |
| shutil.copytree(source, destination) |
| else: |
| shutil.copy2(source, destination) |
| except OSError as err: |
| raise IOError(err) |
| |
| exceptions = {} |
| for source, destination in zip(source_file_names, destination_file_names): |
| try: |
| _copy_path(source, destination) |
| except Exception as e: # pylint: disable=broad-except |
| exceptions[(source, destination)] = e |
| |
| if exceptions: |
| raise BeamIOError("Copy operation failed", exceptions) |
| |
| def rename(self, source_file_names, destination_file_names): |
| """Rename the files at the source list to the destination list. |
| Source and destination lists should be of the same size. |
| |
| Args: |
| source_file_names: List of file paths that need to be moved |
| destination_file_names: List of destination_file_names for the files |
| |
| Raises: |
| ``BeamIOError``: if any of the rename operations fail |
| """ |
| err_msg = ( |
| "source_file_names and destination_file_names should " |
| "be equal in length") |
| assert len(source_file_names) == len(destination_file_names), err_msg |
| |
| def _rename_file(source, destination): |
| """Rename a single file object""" |
| try: |
| os.rename(source, destination) |
| except OSError as err: |
| raise IOError(err) |
| |
| exceptions = {} |
| for source, destination in zip(source_file_names, destination_file_names): |
| try: |
| _rename_file(source, destination) |
| except Exception as e: # pylint: disable=broad-except |
| exceptions[(source, destination)] = e |
| |
| if exceptions: |
| raise BeamIOError("Rename operation failed", exceptions) |
| |
| def exists(self, path): |
| """Check if the provided path exists on the FileSystem. |
| |
| Args: |
| path: string path that needs to be checked. |
| |
| Returns: boolean flag indicating if path exists |
| """ |
| return os.path.exists(path) |
| |
| def size(self, path): |
| """Get size of path on the FileSystem. |
| |
| Args: |
| path: string path in question. |
| |
| Returns: int size of path according to the FileSystem. |
| |
| Raises: |
| ``BeamIOError``: if path doesn't exist. |
| """ |
| try: |
| return os.path.getsize(path) |
| except Exception as e: # pylint: disable=broad-except |
| raise BeamIOError("Size operation failed", {path: e}) |
| |
| def last_updated(self, path): |
| """Get UNIX Epoch time in seconds on the FileSystem. |
| |
| Args: |
| path: string path of file. |
| |
| Returns: float UNIX Epoch time |
| |
| Raises: |
| ``BeamIOError``: if path doesn't exist. |
| """ |
| if not self.exists(path): |
| raise BeamIOError('Path does not exist: %s' % path) |
| return os.path.getmtime(path) |
| |
| def checksum(self, path): |
| """Fetch checksum metadata of a file on the |
| :class:`~apache_beam.io.filesystem.FileSystem`. |
| |
| Args: |
| path: string path of a file. |
| |
| Returns: string containing file size. |
| |
| Raises: |
| ``BeamIOError``: if path isn't a file or doesn't exist. |
| """ |
| if not self.exists(path): |
| raise BeamIOError('Path does not exist: %s' % path) |
| return str(os.path.getsize(path)) |
| |
| def delete(self, paths): |
| """Deletes files or directories at the provided paths. |
| Directories will be deleted recursively. |
| |
| Args: |
| paths: list of paths that give the file objects to be deleted |
| |
| Raises: |
| ``BeamIOError``: if any of the delete operations fail |
| """ |
| def _delete_path(path): |
| """Recursively delete the file or directory at the provided path. |
| """ |
| try: |
| if os.path.isdir(path): |
| shutil.rmtree(path) |
| else: |
| os.remove(path) |
| except OSError as err: |
| raise IOError(err) |
| |
| exceptions = {} |
| |
| def try_delete(path): |
| try: |
| _delete_path(path) |
| except Exception as e: # pylint: disable=broad-except |
| exceptions[path] = e |
| |
| for match_result in self.match(paths): |
| metadata_list = match_result.metadata_list |
| |
| if not metadata_list: |
| exceptions[match_result.pattern] = \ |
| IOError('No files found to delete under: %s' % match_result.pattern) |
| |
| for metadata in match_result.metadata_list: |
| try_delete(metadata.path) |
| |
| if exceptions: |
| raise BeamIOError("Delete operation failed", exceptions) |