feat: Define client workspace to management caches.

- Check cached time of providers.json when it initializes.

Refs: #3

oembedpy/application.py

@@ -1,10 +1,14 @@
 """Core endpoint."""
 
+import json
 import logging
+import time
 from typing import Optional
 
 import httpx
 
+from platformdirs import PlatformDirs
+
 from oembedpy import consumer, discovery
 from oembedpy.provider import ProviderRegistry
 from oembedpy.types import Content
@@ -15,11 +19,14 @@
 class Oembed:
     """Application of oEmbed."""
 
-    registry: ProviderRegistry
+    _registry: ProviderRegistry
 
     def __init__(self):  # noqa: D107
+        pass
+
+    def init(self):
         resp = httpx.get("https://oembed.com/providers.json")
-        self.registry = ProviderRegistry.from_dict(resp.json())
+        self._registry = ProviderRegistry.from_dict(resp.json())
 
     def fetch(
         self,
@@ -41,3 +48,34 @@ def fetch(
             params.max_height = max_height
         content = consumer.fetch_content(api_url, params)
         return content
+
+
+class Workspace(Oembed):
+    """oEmbed client with workspace."""
+
+    def __init__(self):
+        self._dirs = PlatformDirs("oembedpy")
+
+    @property
+    def cache_dir(self):
+        return self._dirs.user_data_path
+
+    def init(self):
+        providers_json = self.cache_dir / "providers.json"
+        use_cache = providers_json.exists()
+        if use_cache:
+            now_ts = time.mktime(time.localtime())
+            file_ts = providers_json.stat().st_mtime
+            # TODO: expired time is temporary value, refer settings or default after.
+            use_cache = file_ts + (3600 * 24) > now_ts
+
+        providers_data: dict
+        if use_cache:
+            providers_data = json.loads(providers_json.read_text())
+        else:
+            providers_json.parent.mkdir(parents=True, exist_ok=True)
+            resp = httpx.get("https://oembed.com/providers.json")
+            providers_data = resp.json()
+            providers_json.write_text(resp.text)
+
+        self._registry = ProviderRegistry.from_dict(providers_data)

oembedpy/cli.py

@@ -22,6 +22,9 @@
 @click.option(
     "--version", is_flag=True, default=False, help="Show version information and exit."
 )
+@click.option(
+    "-w", "--workspace", is_flag=True, default=False, help="Use caching workspace."
+)
 @click.option(
     "--format",
     type=click.Choice(["text", "json"]),
@@ -35,6 +38,7 @@
 def cli(
     ctx: click.Context,
     version: bool,
+    workspace: bool,
     url: str,
     format: OUTPUT_FORMAT,
     max_width: Optional[int] = None,
@@ -47,7 +51,8 @@ def cli(
 
     # Fetch content to find meta tags.
     logger.debug(f"Target Content URL is {url}")
-    oembed = application.Oembed()
+    oembed = application.Workspace() if workspace else application.Oembed()
+    oembed.init()
     try:
         content = oembed.fetch(url, max_width, max_height)
     except Exception as err:

oembedpy/ext/sphinx.py

@@ -63,6 +63,7 @@ def setup(app: Sphinx):  # noqa: D103
         html=(visit_oembed_node, depart_oembed_node),
     )
     _oembed = Oembed()
+    _oembed.init()
     return {
         "version": __version__,
         "parallel_read_safe": True,

pyproject.toml

@@ -46,6 +46,7 @@ test = [
   "pytest==7.*",
   "pytest-cov==4.*",
   "pytest-httpx==0.21.*",
+  "pytest-mock==3.14.*",
 ]
 
 [project.scripts]

tests/test_application.py

@@ -0,0 +1,72 @@
+# flake8: noqa
+import subprocess
+from pathlib import Path
+
+import httpx
+import pytest
+from platformdirs.api import PlatformDirsABC
+from pytest_mock import MockerFixture
+
+
+from oembedpy import application
+
+
+@pytest.fixture
+def mocked_workspace(monkeypatch, tmp_path):
+    def _append_app_name_and_version(*arg, **kwargs):
+        return tmp_path
+
+    monkeypatch.setattr(
+        PlatformDirsABC,
+        "_append_app_name_and_version",
+        _append_app_name_and_version,
+    )
+    ws = application.Workspace()
+    return ws
+
+
+@pytest.mark.skipif("sys.platform != 'linux'")
+def test_workspace():
+    ws = application.Workspace()
+    assert ws.cache_dir == Path.home() / ".local" / "share/oembedpy"
+
+
+@pytest.mark.skipif("sys.platform != 'linux'")
+class TestForWorkspace:
+    def test_properties(self, mocked_workspace: application.Workspace, tmp_path):
+        assert mocked_workspace.cache_dir == tmp_path
+
+    def test_cache_providers_json(
+        self, mocked_workspace: application.Workspace, tmp_path
+    ):
+        mocked_workspace.init()
+        assert (tmp_path / "providers.json").exists()
+        assert mocked_workspace._registry is not None
+
+    def test_fetch_using_cached_providers_json(
+        self, mocked_workspace: application.Workspace, mocker: MockerFixture, tmp_path
+    ):
+        spy = mocker.spy(httpx, "get")
+        mocked_workspace.init()
+        assert spy.call_count == 1
+        other_workspace = application.Workspace()
+        other_workspace.init()
+        assert spy.call_count == 1
+
+    def test_purge_cached_providers_json(
+        self, mocked_workspace: application.Workspace, mocker: MockerFixture, tmp_path
+    ):
+        spy = mocker.spy(httpx, "get")
+        mocked_workspace.init()
+        assert spy.call_count == 1
+        subprocess.run(
+            [
+                "touch",
+                "-t",
+                "200001010000",
+                (mocked_workspace.cache_dir / "providers.json"),
+            ]
+        )
+        other_workspace = application.Workspace()
+        other_workspace.init()
+        assert spy.call_count == 2