Add a mechanism for adding headers to StacIO requests

CHANGELOG.md

@@ -4,6 +4,8 @@
 
 ### Added
 
+- Adds custom `header` support to `DefaultStacIO` ([#889](https://github.com/stac-utils/pystac/pull/889))
+
 ### Removed
 
 ### Changed

docs/concepts.rst

@@ -259,9 +259,25 @@ created automatically by all of the object-specific I/O methods (e.g.
 :meth:`pystac.Catalog.from_file`), so most users will not need to instantiate this
 class themselves.
 
-If you require custom logic for I/O operations or would like to use a 3rd-party library
-for I/O operations (e.g. ``requests``), you can create a sub-class of
-:class:`pystac.StacIO` (or :class:`pystac.DefaultStacIO`) and customize the methods as
+If you are dealing with a STAC catalog with URIs that require authentication.
+It is possible provide auth headers (or any other customer headers) to the
+:class:`pystac.stac_io.DefaultStacIO`.
+
+.. code-block:: python
+
+  from pystac import Catalog
+  from pystac import StacIO
+
+  stac_io = StacIO.default()
+  stac_io.headers = {"Authorization": "<some-auth-header>"}
+
+  catalog = Catalog.from_file("<URI-requiring-auth>", stac_io=stac_io)
+
+
+If you require more custom logic for I/O operations or would like to use a
+3rd-party library for I/O operations (e.g. ``requests``),
+you can create a sub-class of :class:`pystac.StacIO`
+(or :class:`pystac.DefaultStacIO`) and customize the methods as
 you see fit. You can then pass instances of this custom sub-class into the ``stac_io``
 argument of most object-specific I/O methods. You can also use
 :meth:`pystac.StacIO.set_default` in your client's ``__init__.py`` file to make this

pystac/stac_io.py

@@ -12,6 +12,7 @@
 )
 
 from urllib.request import urlopen
+from urllib.request import Request
 from urllib.error import HTTPError
 
 import pystac
@@ -38,6 +39,9 @@
 class StacIO(ABC):
     _default_io: Optional[Callable[[], "StacIO"]] = None
 
+    def __init__(self, headers: Optional[Dict[str, str]] = None):
+        self.headers = headers or {}
+
     @abstractmethod
     def read_text(self, source: HREF, *args: Any, **kwargs: Any) -> str:
         """Read text from the given URI.
@@ -289,7 +293,8 @@ def read_text_from_href(self, href: str) -> str:
         href_contents: str
         if parsed.scheme != "":
             try:
-                with urlopen(href) as f:
+                req = Request(href, headers=self.headers)
+                with urlopen(req) as f:
                     href_contents = f.read().decode("utf-8")
             except HTTPError as e:
                 raise Exception("Could not read uri {}".format(href)) from e

tests/test_stac_io.py

@@ -1,3 +1,4 @@
+import json
 import os
 import unittest
 import tempfile
@@ -98,3 +99,18 @@ class ReportingStacIO(DefaultStacIO, DuplicateKeyReportingMixin):
                 str(excinfo.exception),
                 f'Found duplicate object name "key" in {src_href}',
             )
+
+    @unittest.mock.patch("pystac.stac_io.urlopen")
+    def test_headers_stac_io(self, urlopen_mock: unittest.mock.MagicMock) -> None:
+        stac_io = DefaultStacIO(headers={"Authorization": "api-key fake-api-key-value"})
+
+        catalog = pystac.Catalog("an-id", "a description").to_dict()
+        # required until https://github.com/stac-utils/pystac/pull/896 is merged
+        catalog["links"] = []
+        urlopen_mock.return_value.__enter__.return_value.read.return_value = json.dumps(
+            catalog
+        ).encode("utf-8")
+        pystac.Catalog.from_file("https://example.com/catalog.json", stac_io=stac_io)
+
+        request_obj = urlopen_mock.call_args[0][0]
+        self.assertEqual(request_obj.headers, stac_io.headers)