Switch to Ruff

.github/workflows/static-analysis.yml

@@ -0,0 +1,27 @@
+name: Lint and Format with Ruff
+
+on: push
+
+jobs:
+  check-with-ruff:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Install Python
+        uses: actions/setup-python@v4
+        with:
+          python-version: 3.x
+
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install ruff
+
+      - name: Ruff linting check
+        run: |
+          ruff check --output-format=github earthaccess tests
+
+      - name: Ruff format check
+        run: |
+          ruff format --diff earthaccess tests

.gitignore

@@ -18,6 +18,7 @@ build/
 *.egg-info/
 docs/tutorials/data
 tests/integration/data
+.ruff_cache
 
 # OS X
 .DS_Store

.pre-commit-config.yaml

@@ -11,7 +11,9 @@ repos:
     - id: trailing-whitespace
     - id: check-toml
     - id: check-json
-- repo: https://github.com/psf/black
-  rev: 23.10.1
+- repo: https://github.com/astral-sh/ruff-pre-commit
+  rev: v0.1.6
   hooks:
-    - id: black
+    - id: ruff
+      args: [ --fix ]
+    - id: ruff-format

earthaccess/__init__.py

@@ -1,7 +1,6 @@
 import logging
 import threading
 from importlib.metadata import version
-from typing import Any
 
 from .api import (
     auth_environ,
@@ -65,18 +64,14 @@ def __getattr__(name):  # type: ignore
                     try:
                         _auth.login(strategy=strategy)
                     except Exception as e:
-                        logger.debug(
-                            f"An error occurred during automatic authentication with {strategy=}: {str(e)}"
-                        )
+                        logger.debug(f"An error occurred during automatic authentication with {strategy=}: {str(e)}")
                         continue
                     else:
                         if not _auth.authenticated:
                             continue
                         else:
                             _store = Store(_auth)
-                            logger.debug(
-                                f"Automatic authentication with {strategy=} was successful"
-                            )
+                            logger.debug(f"Automatic authentication with {strategy=} was successful")
                             break
             return _auth if name == "__auth__" else _store
     else:

earthaccess/api.py

@@ -1,10 +1,11 @@
 from typing import Any, Dict, List, Optional, Type, Union
 
-import earthaccess
 import requests
 import s3fs
 from fsspec import AbstractFileSystem
 
+import earthaccess
+
 from .auth import Auth
 from .results import DataGranule
 from .search import CollectionQuery, DataCollections, DataGranules, GranuleQuery
@@ -27,9 +28,7 @@ def _normalize_location(location: Union[str, None]) -> Union[str, None]:
     return location
 
 
-def search_datasets(
-    count: int = -1, **kwargs: Any
-) -> List[earthaccess.results.DataCollection]:
+def search_datasets(count: int = -1, **kwargs: Any) -> List[earthaccess.results.DataCollection]:
     """Search datasets using NASA's CMR
 
     [https://cmr.earthdata.nasa.gov/search/site/docs/search/api.html](https://cmr.earthdata.nasa.gov/search/site/docs/search/api.html)
@@ -65,9 +64,7 @@ def search_datasets(
         ```
     """
     if not validate.valid_dataset_parameters(**kwargs):
-        print(
-            "Warning: a valid set of parameters is needed to search for datasets on CMR"
-        )
+        print("Warning: a valid set of parameters is needed to search for datasets on CMR")
         return []
     if earthaccess.__auth__.authenticated:
         query = DataCollections(auth=earthaccess.__auth__).parameters(**kwargs)
@@ -80,9 +77,7 @@ def search_datasets(
     return query.get_all()
 
 
-def search_data(
-    count: int = -1, **kwargs: Any
-) -> List[earthaccess.results.DataGranule]:
+def search_data(count: int = -1, **kwargs: Any) -> List[earthaccess.results.DataGranule]:
     """Search dataset granules using NASA's CMR.
 
     [https://cmr.earthdata.nasa.gov/search/site/docs/search/api.html](https://cmr.earthdata.nasa.gov/search/site/docs/search/api.html)
@@ -357,7 +352,5 @@ def get_edl_token() -> str:
 def auth_environ() -> Dict[str, str]:
     auth = earthaccess.__auth__
     if not auth.authenticated:
-        raise RuntimeError(
-            "`auth_environ()` requires you to first authenticate with `earthaccess.login()`"
-        )
+        raise RuntimeError("`auth_environ()` requires you to first authenticate with `earthaccess.login()`")
     return {"EARTHDATA_USERNAME": auth.username, "EARTHDATA_PASSWORD": auth.password}

earthaccess/auth.py

@@ -23,9 +23,7 @@ class SessionWithHeaderRedirection(requests.Session):
 
     AUTH_HOST = "urs.earthdata.nasa.gov"
 
-    def __init__(
-        self, username: Optional[str] = None, password: Optional[str] = None
-    ) -> None:
+    def __init__(self, username: Optional[str] = None, password: Optional[str] = None) -> None:
         super().__init__()
         if username and password:
             self.auth = (username, password)
@@ -95,9 +93,7 @@ def refresh_tokens(self) -> bool:
         This method renews the tokens to make sure we can query the collections allowed to our EDL user.
         """
         if len(self.tokens) == 0:
-            resp_tokens = self._generate_user_token(
-                username=self.username, password=self.password
-            )
+            resp_tokens = self._generate_user_token(username=self.username, password=self.password)
             if resp_tokens.ok:
                 self.token = resp_tokens.json()
                 self.tokens = [self.token]
@@ -108,9 +104,7 @@ def refresh_tokens(self) -> bool:
             else:
                 return False
         if len(self.tokens) == 1:
-            resp_tokens = self._generate_user_token(
-                username=self.username, password=self.password
-            )
+            resp_tokens = self._generate_user_token(username=self.username, password=self.password)
             if resp_tokens.ok:
                 self.token = resp_tokens.json()
                 self.tokens.extend(self.token)
@@ -124,9 +118,7 @@ def refresh_tokens(self) -> bool:
         if len(self.tokens) == 2:
             resp_revoked = self._revoke_user_token(self.token["access_token"])
             if resp_revoked:
-                resp_tokens = self._generate_user_token(
-                    username=self.username, password=self.password
-                )
+                resp_tokens = self._generate_user_token(username=self.username, password=self.password)
                 if resp_tokens.ok:
                     self.token = resp_tokens.json()
                     self.tokens[0] = self.token
@@ -161,34 +153,22 @@ def get_s3_credentials(
         if self.authenticated:
             session = SessionWithHeaderRedirection(self.username, self.password)
             if endpoint is None:
-                auth_url = self._get_cloud_auth_url(
-                    daac_shortname=daac, provider=provider
-                )
+                auth_url = self._get_cloud_auth_url(daac_shortname=daac, provider=provider)
             else:
                 auth_url = endpoint
             if auth_url.startswith("https://"):
                 cumulus_resp = session.get(auth_url, timeout=15, allow_redirects=True)
-                auth_resp = session.get(
-                    cumulus_resp.url, allow_redirects=True, timeout=15
-                )
+                auth_resp = session.get(cumulus_resp.url, allow_redirects=True, timeout=15)
                 if not (auth_resp.ok):  # type: ignore
                     # Let's try to authenticate with Bearer tokens
                     _session = self.get_session()
-                    cumulus_resp = _session.get(
-                        auth_url, timeout=15, allow_redirects=True
-                    )
-                    auth_resp = _session.get(
-                        cumulus_resp.url, allow_redirects=True, timeout=15
-                    )
+                    cumulus_resp = _session.get(auth_url, timeout=15, allow_redirects=True)
+                    auth_resp = _session.get(cumulus_resp.url, allow_redirects=True, timeout=15)
                     if not (auth_resp.ok):
-                        print(
-                            f"Authentication with Earthdata Login failed with:\n{auth_resp.text[0:1000]}"
-                        )
+                        print(f"Authentication with Earthdata Login failed with:\n{auth_resp.text[0:1000]}")
                         eula_url = "https://urs.earthdata.nasa.gov/users/earthaccess/unaccepted_eulas"
                         apps_url = "https://urs.earthdata.nasa.gov/application_search"
-                        print(
-                            f"Consider accepting the EULAs available at {eula_url} and applications at {apps_url}"
-                        )
+                        print(f"Consider accepting the EULAs available at {eula_url} and applications at {apps_url}")
                         return {}
 
                     return auth_resp.json()
@@ -214,9 +194,7 @@ class Session instance with Auth and bearer token headers
         if bearer_token and self.authenticated:
             # This will avoid the use of the netrc after we are logged in
             session.trust_env = False
-            session.headers.update(
-                {"Authorization": f'Bearer {self.token["access_token"]}'}
-            )
+            session.headers.update({"Authorization": f'Bearer {self.token["access_token"]}'})
         return session
 
     def get_user_profile(self) -> Dict[str, Any]:
@@ -243,9 +221,7 @@ def _netrc(self) -> bool:
         try:
             my_netrc = Netrc()
         except FileNotFoundError as err:
-            raise FileNotFoundError(
-                f"No .netrc found in {os.path.expanduser('~')}"
-            ) from err
+            raise FileNotFoundError(f"No .netrc found in {os.path.expanduser('~')}") from err
         except NetrcParseError as err:
             raise NetrcParseError("Unable to parse .netrc") from err
         if my_netrc["urs.earthdata.nasa.gov"] is not None:
@@ -271,16 +247,12 @@ def _environment(self) -> bool:
             )
         return authenticated
 
-    def _get_credentials(
-        self, username: Optional[str], password: Optional[str]
-    ) -> bool:
+    def _get_credentials(self, username: Optional[str], password: Optional[str]) -> bool:
         if username is not None and password is not None:
             token_resp = self._get_user_tokens(username, password)
 
             if not (token_resp.ok):  # type: ignore
-                print(
-                    f"Authentication with Earthdata Login failed with:\n{token_resp.text}"
-                )
+                print(f"Authentication with Earthdata Login failed with:\n{token_resp.text}")
                 return False
             logger.debug("You're now authenticated with NASA Earthdata Login")
             self.username = username
@@ -297,9 +269,7 @@ def _get_credentials(
                 self.token = self.tokens[0]
             elif len(self.tokens) > 0:
                 self.token = self.tokens[0]
-                logger.debug(
-                    f"Using token with expiration date: {self.token['expiration_date']}"
-                )
+                logger.debug(f"Using token with expiration date: {self.token['expiration_date']}")
             profile = self.get_user_profile()
             if "email_address" in profile:
                 self.user_profile = profile
@@ -360,9 +330,7 @@ def _persist_user_credentials(self, username: str, password: str) -> bool:
         my_netrc.save()
         return True
 
-    def _get_cloud_auth_url(
-        self, daac_shortname: Optional[str] = "", provider: Optional[str] = ""
-    ) -> str:
+    def _get_cloud_auth_url(self, daac_shortname: Optional[str] = "", provider: Optional[str] = "") -> str:
         for daac in DAACS:
             if (
                 daac_shortname == daac["short-name"]

earthaccess/daac.py

@@ -110,30 +110,16 @@
 
 # Some testing urls behind EDL
 DAAC_TEST_URLS = [
-    (
-        "https://archive.podaac.earthdata.nasa.gov/podaac-ops-cumulus-protected/"
-        "JASON_CS_S6A_L2_ALT_LR_STD_OST_NRT_F/"
-    ),
-    (
-        "https://data.nsidc.earthdatacloud.nasa.gov/nsidc-cumulus-prod-protected/ATLAS/"
-        "ATL03/005/2018/10/14/dummy.nc"
-    ),
-    (
-        "https://n5eil01u.ecs.nsidc.org/DP7/ATLAS/ATL06.005/2018.10.14/"
-        "ATL06_20181014045341_02380102_005_01.iso.xml"
-    ),
-    ("https://hydro1.gesdisc.eosdis.nasa.gov/data/GLDAS/GLDAS_NOAH10_M.2.0/1948/"),
-    (
-        "https://e4ftl01.cr.usgs.gov//DP114/MOTA/MCD43A3.006/2000.02.24/"
-        "MCD43A3.A2000055.h15v07.006.2016101151720.hdf.xml"
-    ),
+    "https://archive.podaac.earthdata.nasa.gov/podaac-ops-cumulus-protected/JASON_CS_S6A_L2_ALT_LR_STD_OST_NRT_F/",
+    "https://data.nsidc.earthdatacloud.nasa.gov/nsidc-cumulus-prod-protected/ATLAS/ATL03/005/2018/10/14/dummy.nc",
+    "https://n5eil01u.ecs.nsidc.org/DP7/ATLAS/ATL06.005/2018.10.14/ATL06_20181014045341_02380102_005_01.iso.xml",
+    "https://hydro1.gesdisc.eosdis.nasa.gov/data/GLDAS/GLDAS_NOAH10_M.2.0/1948/",
+    "https://e4ftl01.cr.usgs.gov//DP114/MOTA/MCD43A3.006/2000.02.24/MCD43A3.A2000055.h15v07.006.2016101151720.hdf.xml",
     "https://daac.ornl.gov/daacdata/npp/grassland/NPP_BCN/data/bcn_cli.txt",
 ]
 
 
-def find_provider(
-    daac_short_name: Optional[str] = None, cloud_hosted: Optional[bool] = None
-) -> Union[str, None]:
+def find_provider(daac_short_name: Optional[str] = None, cloud_hosted: Optional[bool] = None) -> Union[str, None]:
     for daac in DAACS:
         if daac_short_name == daac["short-name"]:
             if cloud_hosted:
@@ -150,9 +136,7 @@ def find_provider(
 
 def find_provider_by_shortname(short_name: str, cloud_hosted: bool) -> Union[str, None]:
     base_url = "https://cmr.earthdata.nasa.gov/search/collections.umm_json?"
-    providers = requests.get(
-        f"{base_url}&cloud_hosted={cloud_hosted}&short_name={short_name}"
-    ).json()
+    providers = requests.get(f"{base_url}&cloud_hosted={cloud_hosted}&short_name={short_name}").json()
     if int(providers["hits"]) > 0:
         return providers["items"][0]["meta"]["provider-id"]
     else:

earthaccess/formatters.py

@@ -8,10 +8,7 @@
 
 def _load_static_files() -> List[str]:
     """Load styles"""
-    return [
-        pkg_resources.resource_string("earthaccess", fname).decode("utf8")
-        for fname in STATIC_FILES
-    ]
+    return [pkg_resources.resource_string("earthaccess", fname).decode("utf8") for fname in STATIC_FILES]
 
 
 def _repr_collection_html() -> str: