Improvement/use pathlib

earthaccess/auth.py

@@ -2,6 +2,7 @@
 import importlib.metadata
 import logging
 import os
+from pathlib import Path
 from netrc import NetrcParseError
 from pathlib import Path
 from typing import Any, Dict, List, Optional
@@ -258,7 +259,7 @@ def _netrc(self) -> bool:
             my_netrc = Netrc()
         except FileNotFoundError as err:
             raise FileNotFoundError(
-                f"No .netrc found in {os.path.expanduser('~')}"
+                f"No .netrc found in {Path('~').expanduser()}"
             ) from err
         except NetrcParseError as err:
             raise NetrcParseError("Unable to parse .netrc") from err
@@ -365,7 +366,7 @@ def _persist_user_credentials(self, username: str, password: str) -> bool:
         try:
             netrc_path = Path().home().joinpath(".netrc")
             netrc_path.touch(exist_ok=True)
-            os.chmod(netrc_path.absolute(), 0o600)
+            netrc_path.chmod(0o600)
         except Exception as e:
             print(e)
             return False

earthaccess/store.py

@@ -1,5 +1,5 @@
 import datetime
-import os
+from pathlib import Path
 import shutil
 import traceback
 from functools import lru_cache
@@ -466,7 +466,7 @@ def get(
             List of downloaded files
         """
         if local_path is None:
-            local_path = os.path.join(
+            local_path = Path(
                 ".",
                 "data",
                 f"{datetime.datetime.today().strftime('%Y-%m-%d')}-{uuid4().hex[:6]}",
@@ -526,7 +526,7 @@ def _get_urls(
             # TODO: make this parallel or concurrent
             for file in data_links:
                 s3_fs.get(file, local_path)
-                file_name = os.path.join(local_path, os.path.basename(file))
+                file_name = Path(local_path, Path(file).name)
                 print(f"Downloaded: {file_name}")
                 downloaded_files.append(file_name)
             return downloaded_files
@@ -572,7 +572,7 @@ def _get_granules(
             # TODO: make this async
             for file in data_links:
                 s3_fs.get(file, local_path)
-                file_name = os.path.join(local_path, os.path.basename(file))
+                file_name = Path(local_path, Path(file).name)
                 print(f"Downloaded: {file_name}")
                 downloaded_files.append(file_name)
             return downloaded_files
@@ -597,7 +597,7 @@ def _download_file(self, url: str, directory: str) -> str:
         local_filename = url.split("/")[-1]
         path = Path(directory) / Path(local_filename)
         local_path = str(path)
-        if not os.path.exists(local_path):
+        if not Path(local_path).exists():
             try:
                 session = self.auth.get_session()
                 with session.get(
@@ -638,8 +638,8 @@ def _download_onprem_granules(
             raise ValueError(
                 "We need to be logged into NASA EDL in order to download data granules"
             )
-        if not os.path.exists(directory):
-            os.makedirs(directory)
+        if not Path(directory).exists():
+            Path(directory).mkdir(parents=True)
 
         arguments = [(url, directory) for url in urls]
         results = pqdm(

tests/integration/test_api.py

@@ -1,6 +1,7 @@
 # package imports
 import logging
 import os
+from pathlib import Path
 import unittest
 
 import earthaccess
@@ -84,7 +85,7 @@ def test_download(tmp_path, selection, use_url):
     result = results[selection]
     files = earthaccess.download(result, str(tmp_path))
     assertions.assertIsInstance(files, list)
-    assert all(os.path.exists(f) for f in files)
+    assert all(Path(f).exists() for f in files)
 
 
 def test_auth_environ():

tests/integration/test_cloud_download.py

@@ -1,6 +1,7 @@
 # package imports
 import logging
 import os
+from pathlib import Path
 import random
 import shutil
 import unittest
@@ -166,4 +167,4 @@ def test_multi_file_granule(tmp_path):
     urls = granules[0].data_links()
     assert len(urls) > 1
     files = earthaccess.download(granules, str(tmp_path))
-    assert set(map(os.path.basename, urls)) == set(map(os.path.basename, files))
+    assert set([Path(f).name for f in urls]) == set([Path(f).name for f in files])

tests/integration/test_kerchunk.py

@@ -1,5 +1,6 @@
 import logging
 import os
+from pathlib import Path
 import unittest
 
 import earthaccess
@@ -32,14 +33,14 @@ def granules():
 @pytest.mark.parametrize("protocol", ["", "file://"])
 def test_consolidate_metadata_outfile(tmp_path, granules, protocol):
     outfile = f"{protocol}{tmp_path / 'metadata.json'}"
-    assert not os.path.exists(outfile)
+    assert not Path(outfile).exists()
     result = earthaccess.consolidate_metadata(
         granules,
         outfile=outfile,
         access="indirect",
         kerchunk_options={"concat_dims": "Time"},
     )
-    assert os.path.exists(strip_protocol(outfile))
+    assert Path(strip_protocol(outfile)).exists()
     assert result == outfile