Use pathlibfs for scheme-agnostic file access

Author: amotl

CHANGES.md

@@ -2,6 +2,7 @@
 
 
 ## Unreleased
+- Use `pathlibfs` for scheme-agnostic file access
 
 ## 2023-10-07 0.1.0
 - Add example data files in different formats

hubspot_tech_writing/core.py

@@ -2,7 +2,7 @@
 import logging
 import typing as t
 import warnings
-from pathlib import Path
+from pathlibfs import Path
 from tempfile import NamedTemporaryFile
 
 import hubspot
@@ -14,7 +14,7 @@
 from hubspot_tech_writing.hubspot_api import HubSpotAdapter, HubSpotBlogPost, HubSpotFile
 from hubspot_tech_writing.util.common import ContentTypeResolver
 from hubspot_tech_writing.util.html import HTMLImageTranslator
-from hubspot_tech_writing.util.io import to_io
+from hubspot_tech_writing.util.io import to_io, path_from_url
 
 logger = logging.getLogger(__name__)
 
@@ -75,9 +75,13 @@ def upload(
     folder_id: t.Optional[str] = None,
     folder_path: t.Optional[str] = None,
 ):
-    source_path = Path(source)
+    source_path = source
+    if isinstance(source_path, str):
+        source_path = path_from_url(source)
+    logger.info(f"Source: {source}")
+    logger.info(f"Source path: {source_path} {type(source_path)}")
 
-    ctr = ContentTypeResolver(name=source_path)
+    ctr = ContentTypeResolver(filepath=source_path)
 
     logger.info(f"Uploading file: {source}")
     hsa = HubSpotAdapter(access_token=access_token)
@@ -100,6 +104,7 @@ def upload(
                 upload, access_token=access_token, folder_id=folder_id, folder_path=folder_path
             )
             hit = HTMLImageTranslator(html=html, source_path=source_path, uploader=uploader)
+            logger.info(hit)
             hit.discover().process()
             html = hit.html_out
 

hubspot_tech_writing/hubspot_api.py

@@ -4,6 +4,7 @@
 import typing as t
 from copy import deepcopy
 from pathlib import Path
+from tempfile import NamedTemporaryFile
 
 import hubspot
 from click import confirm
@@ -134,8 +135,10 @@ def save_file(self, file_id: str, source: str):
         """
         Save / overwrite existing file.
         """
+        tmpfile = NamedTemporaryFile()
+        tmpfile.write(source.read_bytes())
         return self.hs.files.files.files_api.replace(
-            file_id=file_id, file=source, options=json.dumps(self.FILE_OPTIONS)
+            file_id=file_id, file=tmpfile.name, options=json.dumps(self.FILE_OPTIONS)
         )
 
     def delete_file_by_id(self, identifier: str) -> t.Optional[File]:
@@ -286,7 +289,7 @@ def __init__(
     def __str__(self):
         return (
             f"{self.__class__.__name__} identifier={self.identifier}, "
-            f"name={self.name}, folder={self.folder_id or self.folder_path}"
+            f"name={self.name}, folder={self.folder_id or self.folder_path}, source={self.source}"
         )
 
     def load(self):
@@ -310,7 +313,7 @@ def save(self):
         if not self.source:
             raise ValueError(f"Unable to save file without source: {self}")
         logger.info(f"Saving file: {self}")
-        return self.hsa.save_file(file_id=self.identifier, source=str(self.source))
+        return self.hsa.save_file(file_id=self.identifier, source=self.source)
 
     def delete(self):
         """

hubspot_tech_writing/util/common.py

@@ -1,9 +1,10 @@
 import logging
 import typing as t
-from pathlib import Path
+from pathlibfs import Path
 
 import colorlog
 from colorlog.escape_codes import escape_codes
+from yarl import URL
 
 
 def setup_logging(level=logging.INFO, verbose: bool = False):
@@ -23,9 +24,10 @@ class ContentTypeResolver:
     HTML_SUFFIXES = [".html", ".html5", ".htm"]
     TEXT_SUFFIXES = MARKUP_SUFFIXES + HTML_SUFFIXES + [".txt"]
 
-    def __init__(self, name: t.Union[str, Path]):
-        self.name = name
-        self.suffix = Path(name).suffix
+    def __init__(self, filepath: t.Union[str, Path]):
+        self.url = URL(str(filepath)).with_scheme("")
+        self.path = Path(str(self.url))
+        self.suffix = self.path.suffix
 
     def is_markup(self):
         return self.suffix in self.MARKUP_SUFFIXES
@@ -38,3 +40,8 @@ def is_text(self):
 
     def is_file(self):
         return not self.is_text()
+
+
+def url_to_path(filepath: str):
+    url = URL(str(filepath)).with_scheme("")
+    return Path(str(url))

hubspot_tech_writing/util/html.py

@@ -2,10 +2,12 @@
 import logging
 import typing as t
 from copy import deepcopy
-from pathlib import Path
+from pathlibfs import Path
 
 from bs4 import BeautifulSoup
 
+from hubspot_tech_writing.util.common import url_to_path
+
 logger = logging.getLogger(__name__)
 
 
@@ -32,7 +34,7 @@ def __init__(self, html: str, source_path: t.Union[str, Path], uploader: t.Optio
 
     def __str__(self):
         return (
-            f"HTMLImageTranslator:\nin:     {self.images_in}\nlocal:  {self.images_local}\nremote: {self.images_remote}"
+            f"HTMLImageTranslator:\nsource path: {self.source_path}\nin:     {self.images_in}\nlocal:  {self.images_local}\nremote: {self.images_remote}"
         )
 
     def discover(self):
@@ -60,8 +62,9 @@ def resolve(self) -> "HTMLImageTranslator":
         Process discovered image elements, computing effective paths.
         """
         if self.source_path is None:
+            logger.warning("No resolving without source path")
             return self
-        parent_path = Path(self.source_path)
+        parent_path = self.source_path
         if parent_path.is_file():
             parent_path = parent_path.parent
         self.images_local = []
@@ -74,7 +77,7 @@ def resolve(self) -> "HTMLImageTranslator":
 
                 # Relative paths are relative to the original document.
                 else:
-                    image_new.src = str(Path(parent_path) / image.src)
+                    image_new.src = parent_path / image.src
             self.images_local.append(image_new)
         return self
 
@@ -86,7 +89,7 @@ def upload(self) -> "HTMLImageTranslator":
             logger.warning("No upload without uploader")
             return self
         for image_local in self.images_local:
-            hs_file = self.uploader(source=image_local.src, name=Path(image_local.src).name)
+            hs_file = self.uploader(source=image_local.src, name=image_local.src.name)
             image_url = hs_file.url
             image_remote: HTMLImage = deepcopy(image_local)
             image_remote.src = image_url

hubspot_tech_writing/util/io.py

@@ -1,22 +1,64 @@
 import contextlib
 import io
 import typing as t
-from pathlib import Path
+from pathlibfs import Path
 
 import requests
+from yarl import URL
 
 
 @contextlib.contextmanager
 def to_io(source: t.Union[str, Path, t.IO]) -> t.Generator[t.IO, None, None]:
     if isinstance(source, (str, Path)):
         source = str(source)
         fp: t.IO
+        path = path_from_url(source)
+        fp = path.open(mode="rt")
+        """
         if source.startswith("http://") or source.startswith("https://"):
             response = requests.get(source, timeout=10.0)
             fp = io.StringIO(response.text)
         else:
             fp = open(source, "r")
+        """
     else:
         fp = source
     yield fp
     fp.close()
+
+
+def path_from_url(url: str) -> Path:
+    """
+    Convert GitHub HTTP URL to pathlibfs / fsspec URL.
+
+    Input URLs
+    ----------
+    github+https://foobar:[email protected]/acme/sweet-camino/path/to/document.md
+    github+https://foobar:[email protected]/acme/sweet-camino/blob/main/path/to/document.md
+
+    Output Path
+    -----------
+    fs = Path("github://path/to/document.md", username="foobar", token="ghp_lalala", org="acme", repo="sweet-camino")
+    """
+    uri = URL(url)
+
+    if uri.scheme.startswith("github+https"):
+
+        path_fragments = uri.path.split("/")[1:]
+        path_kwargs = {
+            "username": uri.user,
+            "token": uri.password,
+            "org": path_fragments[0],
+            "repo": path_fragments[1],
+        }
+
+        real_path_fragments = path_fragments[2:]
+        if path_fragments[2] == "blob":
+            real_path_fragments = path_fragments[4:]
+
+        downstream_url = "github://" + "/".join(real_path_fragments)
+        path = Path(downstream_url, **path_kwargs)
+
+    else:
+        path = Path(url)
+    return path

pyproject.toml

@@ -70,7 +70,9 @@ dependencies = [
   "hubspot-api-client<9",
   "markdown<4",
   "mkdocs-linkcheck<2",
+  "pathlibfs<0.6",
   "requests<3",
+  "yarl<2",
 ]
 
 [project.optional-dependencies]