WIP

patent_client/epo/ops/published/manager_test.py

@@ -5,7 +5,6 @@
 from ..session import OpsAsyncSession
 from ..session import OpsAuthenticationError
 from .model.search import Inpadoc
-from patent_client.util.hishel_util import cache_disabled
 
 
 class TestPublished:
@@ -54,7 +53,7 @@ def test_issue_41(self):
     def test_issue_76_no_credential(self):
         good_session = api.asession
         api.asession = OpsAsyncSession()
-        with cache_disabled(asession):
+        with asession.cache_disabled():
             with pytest.raises(OpsAuthenticationError):
                 pub = Inpadoc.objects.get("EP3082535A1")
         api.asession = good_session

patent_client/epo/ops/session.py

@@ -4,7 +4,7 @@
 from patent_client import SETTINGS
 from patent_client.session import PatentClientSession
 from patent_client.util.asyncio_util import SyncProxy
-from patent_client.util.hishel_util import cache_disabled
+
 
 NS = {
     "http://ops.epo.org": None,
@@ -58,7 +58,7 @@ async def request(self, *args, **kwargs):
 
     async def get_token(self):
         auth_url = "https://ops.epo.org/3.2/auth/accesstoken"
-        with cache_disabled(self):
+        with self.cache_disabled():
             response = await super(OpsAsyncSession, self).request(
                 "post",
                 auth_url,

patent_client/session.py

@@ -1,5 +1,6 @@
 import re
 import warnings
+from contextlib import contextmanager
 from pathlib import Path
 from typing import Optional
 
@@ -57,3 +58,13 @@ async def adownload(self, url, method: str = "GET", path: Optional[str | Path] =
                 async for chunk in response.aiter_bytes():
                     f.write(chunk)
         return path
+
+    @contextmanager
+    def cache_disabled(session):
+        if hasattr(session._transport, "_transport"):
+            cached_transport = session._transport._transport
+            session._transport = session._transport._transport
+            yield
+            session._transport = cached_transport
+        else:
+            yield

patent_client/uspto/peds/api.py

@@ -6,7 +6,7 @@
 from .model import Document
 from .model import PedsPage
 from .session import session
-from patent_client.util.hishel_util import cache_disabled
+
 
 type_map = {
     "string": str,
@@ -30,7 +30,7 @@ def __init__(self):
 
     @classmethod
     async def is_online(cls) -> bool:
-        with cache_disabled(session):
+        with session.cache_disabled():
             response = await session.get("https://ped.uspto.gov/api/search-fields")
             if response.status_code == 200:
                 return True

patent_client/uspto/public_search/api.py

@@ -6,7 +6,6 @@
 from .model import PublicSearchBiblioPage
 from .model import PublicSearchDocument
 from .session import session
-from patent_client.util.hishel_util import cache_disabled
 
 
 class UsptoException(Exception):
@@ -100,7 +99,7 @@ async def get_document(self, bib) -> "PublicSearchDocument":
 
     async def get_session(self):
         url = "https://ppubs.uspto.gov/dirsearch-public/users/me/session"
-        with cache_disabled(session):
+        with session.cache_disabled():
             response = await session.post(url, json=-1)  # json=str(random.randint(10000, 99999)))
             self.session = response.json()
             self.case_id = self.session["userCase"]["caseId"]