Cookie jar delete specific cookie

CHANGES/4942.feature

@@ -0,0 +1,2 @@
+Add predicate to ``AbstractCookieJar.clear``.
+Add ``AbstractCookieJar.clear_domain`` to clean all domain and subdomains cookies only.

aiohttp/abc.py

@@ -133,12 +133,19 @@ async def close(self) -> None:
     IterableBase = Iterable
 
 
+ClearCookiePredicate = Callable[["Morsel[str]"], bool]
+
+
 class AbstractCookieJar(Sized, IterableBase):
     """Abstract Cookie Jar."""
 
     @abstractmethod
-    def clear(self) -> None:
-        """Clear all cookies."""
+    def clear(self, predicate: Optional[ClearCookiePredicate] = None) -> None:
+        """Clear all cookies if no predicate is passed."""
+
+    @abstractmethod
+    def clear_domain(self, domain: str) -> None:
+        """Clear all cookies for domain and all subdomains."""
 
     @abstractmethod
     def update_cookies(self, cookies: LooseCookies, response_url: URL = URL()) -> None:

aiohttp/cookiejar.py

@@ -21,7 +21,7 @@
 
 from yarl import URL
 
-from .abc import AbstractCookieJar
+from .abc import AbstractCookieJar, ClearCookiePredicate
 from .helpers import get_running_loop, is_ip_address, next_whole_second
 from .typedefs import LooseCookies, PathLike
 
@@ -81,11 +81,41 @@ def load(self, file_path: PathLike) -> None:
         with file_path.open(mode="rb") as f:
             self._cookies = pickle.load(f)
 
-    def clear(self) -> None:
-        self._cookies.clear()
-        self._host_only_cookies.clear()
-        self._next_expiration = next_whole_second()
-        self._expirations.clear()
+    def clear(self, predicate: Optional[ClearCookiePredicate] = None) -> None:
+        if predicate is None:
+            self._next_expiration = next_whole_second()
+            self._cookies.clear()
+            self._host_only_cookies.clear()
+            self._expirations.clear()
+            return
+
+        to_del = []
+        now = datetime.datetime.now(datetime.timezone.utc)
+        for domain, cookie in self._cookies.items():
+            for name, morsel in cookie.items():
+                key = (domain, name)
+                if (
+                    key in self._expirations and self._expirations[key] <= now
+                ) or predicate(morsel):
+                    to_del.append(key)
+
+        for domain, name in to_del:
+            key = (domain, name)
+            self._host_only_cookies.discard(key)
+            if key in self._expirations:
+                del self._expirations[(domain, name)]
+            self._cookies[domain].pop(name, None)
+
+        next_expiration = min(self._expirations.values(), default=self._max_time)
+        try:
+            self._next_expiration = next_expiration.replace(
+                microsecond=0
+            ) + datetime.timedelta(seconds=1)
+        except OverflowError:
+            self._next_expiration = self._max_time
+
+    def clear_domain(self, domain: str) -> None:
+        self.clear(lambda x: self._is_domain_match(domain, x["domain"]))
 
     def __iter__(self) -> "Iterator[Morsel[str]]":
         self._do_expiration()
@@ -96,31 +126,7 @@ def __len__(self) -> int:
         return sum(1 for i in self)
 
     def _do_expiration(self) -> None:
-        now = datetime.datetime.now(datetime.timezone.utc)
-        if self._next_expiration > now:
-            return
-        if not self._expirations:
-            return
-        next_expiration = self._max_time
-        to_del = []
-        cookies = self._cookies
-        expirations = self._expirations
-        for (domain, name), when in expirations.items():
-            if when <= now:
-                cookies[domain].pop(name, None)
-                to_del.append((domain, name))
-                self._host_only_cookies.discard((domain, name))
-            else:
-                next_expiration = min(next_expiration, when)
-        for key in to_del:
-            del expirations[key]
-
-        try:
-            self._next_expiration = next_expiration.replace(
-                microsecond=0
-            ) + datetime.timedelta(seconds=1)
-        except OverflowError:
-            self._next_expiration = self._max_time
+        self.clear(lambda x: False)
 
     def _expire_cookie(self, when: datetime.datetime, domain: str, name: str) -> None:
         self._next_expiration = min(self._next_expiration, when)
@@ -370,7 +376,10 @@ def __iter__(self) -> "Iterator[Morsel[str]]":
     def __len__(self) -> int:
         return 0
 
-    def clear(self) -> None:
+    def clear(self, predicate: Optional[ClearCookiePredicate] = None) -> None:
+        pass
+
+    def clear_domain(self, domain: str) -> None:
         pass
 
     def update_cookies(self, cookies: LooseCookies, response_url: URL = URL()) -> None:

docs/abc.rst

@@ -145,6 +145,22 @@ Abstract Cookie Jar
       :return: :class:`http.cookies.SimpleCookie` with filtered
          cookies for given URL.
 
+   .. method:: clear(predicate=None)
+
+      Removes all cookies from the jar if the predicate is ``None``. Otherwise remove only those :class:`~http.cookies.Morsel` that ``predicate(morsel)`` returns ``True``.
+
+      :param predicate: callable that gets :class:`~http.cookies.Morsel` as a parameter and returns ``True`` if this :class:`~http.cookies.Morsel` must be deleted from the jar.
+
+          .. versionadded:: 3.8
+
+   .. method:: clear_domain(domain)
+
+      Remove all cookies from the jar that belongs to the specified domain or its subdomains.
+
+      :param str domain: domain for which cookies must be deleted from the jar.
+
+      .. versionadded:: 3.8
+
 Abstract Abstract Access Logger
 -------------------------------
 

docs/client_reference.rst

@@ -1842,6 +1842,22 @@ CookieJar
       :param file_path: Path to file from where cookies will be
            imported, :class:`str` or :class:`pathlib.Path` instance.
 
+   .. method:: clear(predicate=None)
+
+      Removes all cookies from the jar if the predicate is ``None``. Otherwise remove only those :class:`~http.cookies.Morsel` that ``predicate(morsel)`` returns ``True``.
+
+      :param predicate: callable that gets :class:`~http.cookies.Morsel` as a parameter and returns ``True`` if this :class:`~http.cookies.Morsel` must be deleted from the jar.
+
+          .. versionadded:: 4.0
+
+   .. method:: clear_domain(domain)
+
+      Remove all cookies from the jar that belongs to the specified domain or its subdomains.
+
+      :param str domain: domain for which cookies must be deleted from the jar.
+
+      .. versionadded:: 4.0
+
 
 .. class:: DummyCookieJar(*, loop=None)
 

tests/test_cookiejar.py

@@ -698,3 +698,48 @@ async def test_loose_cookies_types() -> None:
 
     for loose_cookies_type in accepted_types:
         jar.update_cookies(cookies=loose_cookies_type)
+
+
+async def test_cookie_jar_clear_all():
+    sut = CookieJar()
+    cookie = SimpleCookie()
+    cookie["foo"] = "bar"
+    sut.update_cookies(cookie)
+
+    sut.clear()
+    assert len(sut) == 0
+
+
+async def test_cookie_jar_clear_expired():
+    sut = CookieJar()
+
+    cookie = SimpleCookie()
+
+    cookie["foo"] = "bar"
+    cookie["foo"]["expires"] = "Tue, 1 Jan 1990 12:00:00 GMT"
+
+    with freeze_time("1980-01-01"):
+        sut.update_cookies(cookie)
+
+    sut.clear(lambda x: False)
+    with freeze_time("1980-01-01"):
+        assert len(sut) == 0
+
+
+async def test_cookie_jar_clear_domain():
+    sut = CookieJar()
+    cookie = SimpleCookie()
+    cookie["foo"] = "bar"
+    cookie["domain_cookie"] = "value"
+    cookie["domain_cookie"]["domain"] = "example.com"
+    cookie["subdomain_cookie"] = "value"
+    cookie["subdomain_cookie"]["domain"] = "test.example.com"
+    sut.update_cookies(cookie)
+
+    sut.clear_domain("example.com")
+    iterator = iter(sut)
+    morsel = next(iterator)
+    assert morsel.key == "foo"
+    assert morsel.value == "bar"
+    with pytest.raises(StopIteration):
+        next(iterator)