Close (almost) all file handles

What's remaining are tests and file serving in WSGI app.
The WGSI app use is probably fine but also not a problem for us
currently.

Author: mvdbeek

pulsar/client/action_mapper.py

@@ -446,7 +446,8 @@ def from_dict(cls, action_dict):
         return RemoteCopyAction(source=action_dict["source"])
 
     def write_to_path(self, path):
-        copy_to_path(open(self.path, "rb"), path)
+        with open(self.path, "rb") as f:
+            copy_to_path(f, path)
 
     def write_from_path(self, pulsar_path):
         destination = self.path
@@ -534,7 +535,8 @@ def write_to_path(self, path):
             object_store_id=object_store_ref["object_store_id"],
         )
         filename = self.object_store.get_filename(dataset_object)
-        copy_to_path(open(filename, 'rb'), path)
+        with open(filename, "rb") as f:
+            copy_to_path(f, path)
 
     def write_from_path(self, pulsar_path):
         raise NotImplementedError("Writing raw files to object store not supported at this time.")
@@ -660,7 +662,8 @@ def from_dict(cls, action_dict):
         return MessageAction(contents=action_dict["contents"])
 
     def write_to_path(self, path):
-        open(path, "w").write(self.contents)
+        with open(path, "w") as f:
+            f.write(self.contents)
 
 
 DICTIFIABLE_ACTION_CLASSES = [

pulsar/client/staging/up.py

@@ -579,11 +579,8 @@ def _read(path):
     Utility method to quickly read small files (config files and tool
     wrappers) into memory as bytes.
     """
-    input = open(path, encoding="utf-8")
-    try:
+    with open(path, encoding="utf-8") as input:
         return input.read()
-    finally:
-        input.close()
 
 
 __all__ = ['submit_job']

pulsar/client/transport/curl.py

@@ -33,14 +33,16 @@ def __init__(self, timeout=None, **kwrgs):
 
     def execute(self, url, method=None, data=None, input_path=None, output_path=None):
         buf = _open_output(output_path)
+        input_fh = None
         try:
             c = _new_curl_object_for_url(url)
             c.setopt(c.WRITEFUNCTION, buf.write)
             if method:
                 c.setopt(c.CUSTOMREQUEST, method)
             if input_path:
+                input_fh = open(input_path, "rb")
                 c.setopt(c.UPLOAD, 1)
-                c.setopt(c.READFUNCTION, open(input_path, 'rb').read)
+                c.setopt(c.READFUNCTION, input_fh.read)
                 filesize = os.path.getsize(input_path)
                 c.setopt(c.INFILESIZE, filesize)
             if data:
@@ -61,6 +63,8 @@ def execute(self, url, method=None, data=None, input_path=None, output_path=None
                 return buf.getvalue()
         finally:
             buf.close()
+            if input_fh:
+                input_fh.close()
 
 
 def post_file(url, path):

pulsar/client/transport/poster.py

@@ -26,9 +26,10 @@
 def post_file(url, path):
     __ensure_poster()
     try:
-        datagen, headers = poster.encode.multipart_encode({"file": open(path, "rb")})
-        request = Request(url, datagen, headers)
-        return urlopen(request).read()
+        with open(path, "rb") as fh:
+            datagen, headers = poster.encode.multipart_encode({"file": fh})
+            request = Request(url, datagen, headers)
+            return urlopen(request).read()
     except Exception:
         log.exception("Problem with poster post of [%s]" % path)
         raise

pulsar/client/transport/requests.py

@@ -21,10 +21,11 @@ def post_file(url, path):
         raise ImportError(REQUESTS_TOOLBELT_UNAVAILABLE_MESSAGE)
 
     __ensure_requests()
-    m = requests_toolbelt.MultipartEncoder(
-        fields={'file': ('filename', open(path, 'rb'))}
-    )
-    requests.post(url, data=m, headers={'Content-Type': m.content_type})
+    with open(path, "rb") as f:
+        m = requests_toolbelt.MultipartEncoder(
+            fields={'file': ('filename', f)}
+        )
+        requests.post(url, data=m, headers={'Content-Type': m.content_type})
 
 
 def get_file(url, path):

pulsar/client/util.py

@@ -40,19 +40,16 @@ def copy_to_path(object, path):
     """
     Copy file-like object to path.
     """
-    output = open(path, 'wb')
-    _copy_and_close(object, output)
+    with open(path, 'wb') as output:
+        _copy(object, output)
 
 
-def _copy_and_close(object, output):
-    try:
-        while True:
-            buffer = object.read(BUFFER_SIZE)
-            if not buffer:
-                break
-            output.write(buffer)
-    finally:
-        output.close()
+def _copy(object, output):
+    while True:
+        buffer = object.read(BUFFER_SIZE)
+        if not buffer:
+            break
+        output.write(buffer)
 
 
 # Variant of base64 compat layer inspired by BSD code from Bcfg2
@@ -283,10 +280,12 @@ def __contains__(self, item):
         return exists(self.__path(item))
 
     def __setitem__(self, key, value):
-        open(self.__path(key), 'w').write(json.dumps(value))
+        with open(self.__path(key), "w") as f:
+            f.write(json.dumps(value))
 
     def __getitem__(self, key):
-        return json.loads(open(self.__path(key)).read())
+        with open(self.__path(key)) as f:
+            return json.loads(f).read()
 
     def __delitem__(self, key):
         try:

pulsar/manager_factory.py

@@ -52,7 +52,8 @@ def build_managers(app, conf):
 
 def _populate_manager_descriptions_from_ini(manager_descriptions, job_managers_config):
     config = configparser.ConfigParser()
-    config.readfp(open(job_managers_config))
+    with open(job_managers_config) as config_fh:
+        config.read_file(config_fh)
     for section in config.sections():
         if not section.startswith(MANAGER_PREFIX):
             continue

pulsar/managers/base/__init__.py

@@ -186,7 +186,8 @@ def _check_execution(self, job_id, tool_id, command_line):
         for file in self._list_dir(tool_files_dir):
             if os.path.isdir(join(tool_files_dir, file)):
                 continue
-            contents = open(join(tool_files_dir, file)).read()
+            with open(join(tool_files_dir, file)) as fh:
+                contents = fh.read()
             log.debug("job_id: {} - checking tool file {}".format(job_id, file))
             authorization.authorize_tool_file(basename(file), contents)
         config_files_dir = job_directory.configs_directory()
@@ -255,28 +256,21 @@ def calculate_path(self, remote_path, input_type):
 
     def read_file(self, name, size=-1, default=None):
         path = self._job_file(name)
-        job_file = None
         try:
-            job_file = open(path, 'rb')
-            return job_file.read(size)
+            with open(path, 'rb') as job_file:
+                return job_file.read(size)
         except Exception:
             if default is not None:
                 return default
             else:
                 raise
-        finally:
-            if job_file:
-                job_file.close()
 
     def write_file(self, name, contents):
         path = self._job_file(name)
-        job_file = open(path, 'wb')
-        try:
-            if isinstance(contents, str):
-                contents = contents.encode("UTF-8")
+        if isinstance(contents, str):
+            contents = contents.encode("UTF-8")
+        with open(path, "wb") as job_file:
             job_file.write(contents)
-        finally:
-            job_file.close()
         return path
 
     def remove_file(self, name):

pulsar/managers/queued_condor.py

@@ -41,7 +41,9 @@ def launch(self, job_id, command_line, submit_params={}, dependencies_descriptio
             setup_params=setup_params
         )
         log_path = self.__condor_user_log(job_id)
-        open(log_path, 'w')  # Touch log file
+        with open(log_path, "w"):
+            # Touch log file
+            pass
 
         submit_params.update(self.submission_params)
         build_submit_params = dict(

pulsar/managers/queued_external_drmaa.py

@@ -44,7 +44,8 @@ def launch(self, job_id, command_line, submit_params={}, dependencies_descriptio
             submit_params=submit_params,
             setup_params=setup_params,
         )
-        print(open(attributes['remoteCommand']).read())
+        with open(attributes["remoteCommand"]) as fh:
+            print(fh.read())
         job_attributes_file = self._write_job_file(job_id, 'jt.json', dumps(attributes))
         user = submit_params.get('user', None)
         log.info("Submit as user %s" % user)