Use CurlCffiWarning as a base and disable warnings be default

curl_cffi/__init__.py

@@ -10,6 +10,7 @@
     "CurlHttpVersion",
     "CurlSslVersion",
     "CurlWsFlag",
+    "config_warnings",
     "ffi",
     "lib",
 ]
@@ -31,3 +32,6 @@
     CurlWsFlag,
 )
 from .curl import Curl, CurlError, CurlMime
+from .utils import config_warnings
+
+config_warnings(on=False)

curl_cffi/aio.py

@@ -6,9 +6,9 @@
 from weakref import WeakKeyDictionary, WeakSet
 
 from ._wrapper import ffi, lib
-from .utils import _SHOW_WARNINGS
 from .const import CurlMOpt
 from .curl import DEFAULT_CACERT, Curl
+from .utils import CurlCffiWarning
 
 __all__ = ["AsyncCurl"]
 
@@ -37,8 +37,7 @@ def _get_selector(asyncio_loop) -> asyncio.AbstractEventLoop:
         if not isinstance(asyncio_loop, getattr(asyncio, "ProactorEventLoop", type(None))):
             return asyncio_loop
 
-        if _SHOW_WARNINGS:
-            warnings.warn(PROACTOR_WARNING, RuntimeWarning, stacklevel=2)
+        warnings.warn(PROACTOR_WARNING, CurlCffiWarning, stacklevel=2)
 
         from ._asyncio_selector import AddThreadSelectorEventLoop
 
@@ -204,8 +203,10 @@ def socket_action(self, sockfd: int, ev_bitmask: int) -> int:
 
     def process_data(self, sockfd: int, ev_bitmask: int):
         """Call curl_multi_info_read to read data for given socket."""
-        if _SHOW_WARNINGS and not self._curlm:
-            warnings.warn("Curlm alread closed! quitting from process_data", stacklevel=2)
+        if not self._curlm:
+            warnings.warn(
+                "Curlm alread closed! quitting from process_data", CurlCffiWarning, stacklevel=2
+            )
             return
 
         self.socket_action(sockfd, ev_bitmask)

curl_cffi/curl.py

@@ -11,7 +11,7 @@
 
 from ._wrapper import ffi, lib
 from .const import CurlECode, CurlHttpVersion, CurlInfo, CurlOpt, CurlWsFlag
-from .utils import _SHOW_WARNINGS
+from .utils import CurlCffiWarning
 
 DEFAULT_CACERT = certifi.where()
 REASON_PHRASE_RE = re.compile(rb"HTTP/\d\.\d [0-9]{3} (.*)")
@@ -85,8 +85,8 @@ def write_callback(ptr, size, nmemb, userdata):
     if wrote == CURL_WRITEFUNC_PAUSE or wrote == CURL_WRITEFUNC_ERROR:  # noqa: SIM109
         return wrote
     # should make this an exception in future versions
-    if _SHOW_WARNINGS and wrote != nmemb * size:
-        warnings.warn("Wrote bytes != received bytes.", RuntimeWarning, stacklevel=2)
+    if wrote != nmemb * size:
+        warnings.warn("Wrote bytes != received bytes.", CurlCffiWarning, stacklevel=2)
     return nmemb * size
 
 
@@ -130,8 +130,8 @@ def __init__(self, cacert: str = "", debug: bool = False, handle=None) -> None:
 
     def _set_error_buffer(self) -> None:
         ret = lib._curl_easy_setopt(self._curl, CurlOpt.ERRORBUFFER, self._error_buffer)
-        if _SHOW_WARNINGS and ret != 0:
-            warnings.warn("Failed to set error buffer", stacklevel=2)
+        if ret != 0:
+            warnings.warn("Failed to set error buffer", CurlCffiWarning, stacklevel=2)
         if self._debug:
             self.setopt(CurlOpt.VERBOSE, 1)
             lib._curl_easy_setopt(self._curl, CurlOpt.DEBUGFUNCTION, lib.debug_function)

curl_cffi/requests/cookies.py

@@ -13,7 +13,7 @@
 from typing import Dict, Iterator, List, MutableMapping, Optional, Tuple, Union
 from urllib.parse import urlparse
 
-from ..utils import _SHOW_WARNINGS
+from ..utils import CurlCffiWarning
 from .errors import CookieConflict, RequestsError
 
 CookieTypes = Union["Cookies", CookieJar, Dict[str, str], List[Tuple[str, str]]]
@@ -192,19 +192,19 @@ def set(self, name: str, value: str, domain: str = "", path: str = "/", secure=F
         """
         # https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Set-Cookie
         if name.startswith("__Secure-") and secure is False:
-            if _SHOW_WARNINGS:
-                warnings.warn(
-                    "`secure` changed to True for `__Secure-` prefixed cookies",
-                    stacklevel=2,
-                )
+            warnings.warn(
+                "`secure` changed to True for `__Secure-` prefixed cookies",
+                CurlCffiWarning,
+                stacklevel=2,
+            )
             secure = True
         elif name.startswith("__Host-") and (secure is False or domain or path != "/"):
-            if _SHOW_WARNINGS:
-                warnings.warn(
-                    "`host` changed to True, `domain` removed, `path` changed to `/` "
-                    "for `__Host-` prefixed cookies",
-                    stacklevel=2,
-                )
+            warnings.warn(
+                "`host` changed to True, `domain` removed, `path` changed to `/` "
+                "for `__Host-` prefixed cookies",
+                CurlCffiWarning,
+                stacklevel=2,
+            )
             secure = True
             domain = ""
             path = "/"

curl_cffi/requests/impersonate.py

@@ -4,7 +4,7 @@
 from typing import List, Literal, Optional, TypedDict
 
 from ..const import CurlOpt, CurlSslVersion
-from ..utils import _SHOW_WARNINGS
+from ..utils import CurlCffiWarning
 
 BrowserTypeLiteral = Literal[
     # Edge
@@ -304,10 +304,11 @@ def toggle_extension(curl, extension_id: int, enable: bool):
             curl.setopt(CurlOpt.ECH, "")
     # compress certificate
     elif extension_id == 27:
-        if _SHOW_WARNINGS and enable:
+        if enable:
             warnings.warn(
                 "Cert compression setting to brotli, "
                 "you had better specify which to use: zlib/brotli",
+                CurlCffiWarning,
                 stacklevel=1,
             )
             curl.setopt(CurlOpt.SSL_CERT_COMPRESSION, "brotli")

curl_cffi/requests/models.py

@@ -5,7 +5,7 @@
 from typing import Any, Awaitable, Callable, Dict, List, Optional, Union
 
 from ..curl import Curl
-from ..utils import _SHOW_WARNINGS
+from ..utils import CurlCffiWarning
 from .cookies import Cookies
 from .exceptions import HTTPError, RequestException
 from .headers import Headers
@@ -174,8 +174,12 @@ def iter_content(self, chunk_size=None, decode_unicode=False):
         """
         iterate streaming content chunk by chunk in bytes.
         """
-        if _SHOW_WARNINGS and chunk_size:
-            warnings.warn("chunk_size is ignored, there is no way to tell curl that.", stacklevel=2)
+        if chunk_size:
+            warnings.warn(
+                "chunk_size is ignored, there is no way to tell curl that.",
+                CurlCffiWarning,
+                stacklevel=2,
+            )
         if decode_unicode:
             raise NotImplementedError()
 
@@ -237,8 +241,12 @@ async def aiter_content(self, chunk_size=None, decode_unicode=False):
         """
         iterate streaming content chunk by chunk in bytes.
         """
-        if _SHOW_WARNINGS and chunk_size:
-            warnings.warn("chunk_size is ignored, there is no way to tell curl that.", stacklevel=2)
+        if chunk_size:
+            warnings.warn(
+                "chunk_size is ignored, there is no way to tell curl that.",
+                CurlCffiWarning,
+                stacklevel=2,
+            )
         if decode_unicode:
             raise NotImplementedError()
 

curl_cffi/requests/session.py

@@ -26,7 +26,7 @@
 from ..aio import AsyncCurl
 from ..const import CurlHttpVersion, CurlInfo, CurlOpt
 from ..curl import Curl, CurlError, CurlMime
-from ..utils import _SHOW_WARNINGS
+from ..utils import CurlCffiWarning
 from .cookies import Cookies, CookieTypes, CurlMorsel
 from .exceptions import RequestException, SessionClosed, code2error
 from .headers import Headers, HeaderTypes
@@ -320,8 +320,10 @@ def __init__(
     @property
     def curl(self):
         if self._use_thread_local_curl:
-            if _SHOW_WARNINGS and self._is_customized_curl:
-                warnings.warn("Creating fresh curl handle in different thread.", stacklevel=2)
+            if self._is_customized_curl:
+                warnings.warn(
+                    "Creating fresh curl handle in different thread.", CurlCffiWarning, stacklevel=2
+                )
             if not getattr(self._local, "curl", None):
                 self._local.curl = Curl(debug=self.debug)
             return self._local.curl

curl_cffi/requests/utils.py

@@ -27,7 +27,7 @@
 
 from ..const import CurlHttpVersion, CurlOpt, CurlSslVersion
 from ..curl import CURL_WRITEFUNC_ERROR, CurlMime
-from ..utils import _SHOW_WARNINGS
+from ..utils import CurlCffiWarning
 from .cookies import Cookies
 from .exceptions import ImpersonateError, InvalidURL
 from .headers import Headers
@@ -247,13 +247,13 @@ def set_ja3_options(curl: Curl, ja3: str, permute: bool = False):
 
     if extensions.endswith("-21"):
         extensions = extensions[:-3]
-        if _SHOW_WARNINGS:
-            warnings.warn(
-                "Padding(21) extension found in ja3 string, whether to add it should "
-                "be managed by the SSL engine. The TLS client hello packet may contain "
-                "or not contain this extension, any of which should be correct.",
-                stacklevel=1,
-            )
+        warnings.warn(
+            "Padding(21) extension found in ja3 string, whether to add it should "
+            "be managed by the SSL engine. The TLS client hello packet may contain "
+            "or not contain this extension, any of which should be correct.",
+            CurlCffiWarning,
+            stacklevel=1,
+        )
     extension_ids = set(int(e) for e in extensions.split("-"))
     toggle_extensions_by_ids(curl, extension_ids)
 
@@ -523,12 +523,12 @@ def set_curl_options(
             c.setopt(CurlOpt.PROXY, proxy)
 
             if parts.scheme == "https":
-                if _SHOW_WARNINGS and proxy.startswith("https://"):
+                if proxy.startswith("https://"):
                     warnings.warn(
                         "Make sure you are using https over https proxy, otherwise, "
                         "the proxy prefix should be 'http://' not 'https://', "
                         "see: https://github.com/lexiforest/curl_cffi/issues/6",
-                        RuntimeWarning,
+                        CurlCffiWarning,
                         stacklevel=2,
                     )
                 # For https site with http tunnel proxy, tell curl to enable tunneling
@@ -581,8 +581,10 @@ def set_curl_options(
 
     # ja3 string
     if ja3:
-        if _SHOW_WARNINGS and impersonate:
-            warnings.warn("JA3 was altered after browser version was set.", stacklevel=1)
+        if impersonate:
+            warnings.warn(
+                "JA3 was altered after browser version was set.", CurlCffiWarning, stacklevel=1
+            )
         permute = False
         if isinstance(extra_fp, ExtraFingerprints) and extra_fp.tls_permute_extensions:
             permute = True
@@ -592,17 +594,20 @@ def set_curl_options(
 
     # akamai string
     if akamai:
-        if _SHOW_WARNINGS and impersonate:
-            warnings.warn("Akamai was altered after browser version was set.", stacklevel=1)
+        if impersonate:
+            warnings.warn(
+                "Akamai was altered after browser version was set.", CurlCffiWarning, stacklevel=1
+            )
         set_akamai_options(c, akamai)
 
     # extra_fp options
     if extra_fp:
         if isinstance(extra_fp, dict):
             extra_fp = ExtraFingerprints(**extra_fp)
-        if _SHOW_WARNINGS and impersonate:
+        if impersonate:
             warnings.warn(
                 "Extra fingerprints was altered after browser version was set.",
+                CurlCffiWarning,
                 stacklevel=1,
             )
         set_extra_fp(c, extra_fp)

curl_cffi/utils.py

@@ -1,11 +1,13 @@
-_SHOW_WARNINGS = False
+import warnings
 
 
-def enable_warnings():
-    global _SHOW_WARNINGS
-    _SHOW_WARNINGS = True
+class CurlCffiWarning(UserWarning, RuntimeWarning):
+    pass
 
 
-def disable_warnings():
-    global _SHOW_WARNINGS
-    _SHOW_WARNINGS = False
+def config_warnings(on: bool = False):
+    if on:
+        warnings.simplefilter("default", category=CurlCffiWarning)
+    else:
+        warnings.simplefilter("ignore", category=CurlCffiWarning)
+

tests/unittest/test_requests.py

@@ -11,6 +11,7 @@
 from curl_cffi.requests.errors import SessionClosed
 from curl_cffi.requests.exceptions import HTTPError
 from curl_cffi.requests.models import Response
+from curl_cffi.utils import CurlCffiWarning, config_warnings
 
 
 def test_head(server):
@@ -302,7 +303,7 @@ def test_cookies(server):
 
 
 def test_secure_cookies(server):
-    with pytest.warns(UserWarning, match="changed"):
+    with pytest.warns(CurlCffiWarning, match="changed"):
         r = requests.get(
             str(server.url.copy_with(path="/echo_cookies")),
             cookies={"__Secure-foo": "bar", "__Host-hello": "world"},