Use an API key when requesting OpenTopography data (#25)

* add api_key keyword to Topography

* ignore .open_topography.txt file

* return empty string rather than None for file contents

* add OpenTopography API key to the test action

* add unit tests for finding the api key

* Add contributors

* add a teeny bit of documentation about api keys

* Use opentopography in place of open_topography

Squished together is closer to the org name.

Co-authored-by: Mark Piper <[email protected]>

.github/workflows/build-test-ci.yml

@@ -42,6 +42,8 @@ jobs:
         run: make install
 
       - name: Test
+        env:
+          OPENTOPOGRAPHY_API_KEY: ${{ secrets.OPENTOPOGRAPHY_API_KEY }}
         run: |
           pytest --cov=bmi_topography --cov-report=xml:./coverage.xml --cov-config=./setup.cfg -vvv
           bmi-test bmi_topography:BmiTopography --config-file=./examples/config.yaml --root-dir=./examples -vvv

.gitignore

@@ -127,3 +127,6 @@ dmypy.json
 
 # Pyre type checker
 .pyre/
+
+# OpenTopography API key file
+.opentopography.txt

CREDITS.md

@@ -6,6 +6,12 @@ Project lead
 
 * Mark Piper
 
+Contributors
+------------
+
+* Eric Hutton
+* Mark Piper
+
 Acknowledgments
 ---------------
 

bmi_topography/cli.py

@@ -51,7 +51,20 @@
 )
 @click.option("--no_fetch", is_flag=True, help="Do not fetch data from server.")
 def main(quiet, dem_type, south, north, west, east, output_format, no_fetch):
-    """Fetch and cache NASA SRTM and JAXA ALOS land elevation data"""
+    """Fetch and cache NASA SRTM and JAXA ALOS land elevation data
+
+    To fetch some datasets you will need an OpenTopography API key.
+    You can find instructions on how to obtain one from the OpenTopography
+    website:
+
+        https://opentopography.org/blog/introducing-api-keys-access-opentopography-global-datasets
+
+    Once you have received your key, you can pass it to the *bmi-topography*
+    command in one of two ways:
+    1. As the environment variable, OPENTOPOGRAPHY_API_KEY.
+    2. As the contents of an *.opentopography.txt* file located either in
+       your current directory or you home directory.
+    """
     topo = Topography(dem_type, south, north, west, east, output_format)
     if not no_fetch:
         if not quiet:

bmi_topography/topography.py

@@ -1,4 +1,5 @@
 """Base class to access SRTM elevation data"""
+import os
 import urllib
 from pathlib import Path
 
@@ -8,6 +9,31 @@
 from .bbox import BoundingBox
 
 
+def find_api_key():
+    """Search for an API key."""
+    if "OPENTOPOGRAPHY_API_KEY" in os.environ:
+        api_key = os.environ["OPENTOPOGRAPHY_API_KEY"]
+    else:
+        api_key = read_first_of(
+            [".opentopography.txt", "~/.opentopography.txt"]
+        ).strip()
+
+    return api_key
+
+
+def read_first_of(files):
+    """Read the contents of the first file encountered."""
+    contents = ""
+    for path in files:
+        try:
+            contents = open(path, "r").read()
+        except OSError:
+            pass
+        else:
+            break
+    return contents
+
+
 class Topography:
 
     """Fetch and cache NASA SRTM land elevation data."""
@@ -38,7 +64,12 @@ def __init__(
         east=None,
         output_format=None,
         cache_dir=None,
+        api_key=None,
     ):
+        if api_key is None:
+            self._api_key = find_api_key()
+        else:
+            self._api_key = api_key
 
         if dem_type in Topography.VALID_DEM_TYPES:
             self._dem_type = dem_type
@@ -118,6 +149,8 @@ def fetch(self):
                 "east": self.bbox.east,
                 "outputFormat": self.output_format,
             }
+            if self._api_key:
+                params["API_Key"] = self._api_key
 
             response = requests.get(Topography.data_url(), params=params, stream=True)
             response.raise_for_status()

tests/test_api_key.py

@@ -0,0 +1,58 @@
+import os
+from unittest import mock
+
+from bmi_topography.topography import find_api_key, read_first_of
+
+
+def copy_environ(exclude=None):
+    if exclude is None:
+        exclude = {}
+    elif isinstance(exclude, str):
+        exclude = {exclude}
+
+    return {key: value for key, value in os.environ.items() if key not in exclude}
+
+
+def test_find_api_key_not_found():
+    """The API key is not given anywhere"""
+    env = copy_environ(exclude="OPENTOPOGRAPHY_API_KEY")
+    with mock.patch.dict(os.environ, env, clear=True):
+        assert find_api_key() == ""
+
+
+@mock.patch.dict(os.environ, {"OPENTOPOGRAPHY_API_KEY": "foo"})
+def test_find_api_key_env(tmpdir):
+    """The API key is an environment variable"""
+    with tmpdir.as_cwd():
+        with open(".opentopography.txt", "w") as fp:
+            fp.write("bar")
+    assert find_api_key() == "foo"
+
+
+@mock.patch.dict(os.environ, {"OPENTOPOGRAPHY_API_KEY": "foo"})
+def test_find_api_key_from_file(tmpdir):
+    """The API key is in a file"""
+    env = copy_environ(exclude="OPENTOPOGRAPHY_API_KEY")
+    with tmpdir.as_cwd():
+        with open(".opentopography.txt", "w") as fp:
+            fp.write("bar")
+
+        with mock.patch.dict(os.environ, env, clear=True):
+            assert find_api_key() == "bar"
+
+
+def test_read_first_missing(tmpdir):
+    with tmpdir.as_cwd():
+        assert read_first_of(["foo.txt"]) == ""
+        assert read_first_of([]) == ""
+
+
+def test_read_first_file(tmpdir):
+    with tmpdir.as_cwd():
+        with open("foo.txt", "w") as fp:
+            fp.write("foo")
+        with open("bar.txt", "w") as fp:
+            fp.write("bar")
+
+        assert read_first_of(["foo.txt", "bar.txt"]) == "foo"
+        assert read_first_of(["bar.txt", "foo.txt"]) == "bar"