diff --git a/CHANGELOG.md b/CHANGELOG.md index e83d40a..aa0c3e9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,5 +3,6 @@ 2. add `utils.is_running`, `utils.lock_pid_file` 3. add `request.make_response` 4. add `utils.get_paste`(tk), `utils.set_clip`(win32) -5. add `custom_dns` -6. +5. add `utils.custom_dns` +6. update `utils.format_error` support slice index +7. diff --git a/README.md b/README.md index 29b9160..2438674 100644 --- a/README.md +++ b/README.md @@ -130,6 +130,8 @@ Module Docs - https://github.com/ClericPy/morebuiltins/blob/master/doc.md 4.6 `make_response` - Generates an HTTP response based on the provided parameters. +4.7 `custom_dns` - Custom the DNS of socket.getaddrinfo, only effect current thread. + ## 5. morebuiltins.download_python diff --git a/doc.md b/doc.md index 93d7fdc..e4c62b6 100644 --- a/doc.md +++ b/doc.md @@ -323,42 +323,49 @@ ... 1 / 0 ... except Exception as e: ... format_error(e) - '[::3] 1 / 0 >>> ZeroDivisionError(division by zero)' + '[::3] 1 / 0 >>> ZeroDivisionError(division by zero)' >>> try: ... # test in function ... def func1(): 1 / 0 ... func1() ... except Exception as e: ... format_error(e) - '[:func1:3] def func1(): 1 / 0 >>> ZeroDivisionError(division by zero)' + '[::4 | :func1:3] def func1(): 1 / 0 >>> ZeroDivisionError(division by zero)' >>> try: ... # test index ... def func2(): 1 / 0 ... func2() ... except Exception as e: ... format_error(e, index=0) - '[::4] func2() >>> ZeroDivisionError(division by zero)' + '[::4] func2() >>> ZeroDivisionError(division by zero)' + >>> try: + ... # test slice index + ... def func2(): 1 / 0 + ... func2() + ... except Exception as e: + ... format_error(e, index=slice(-1, None, None)) + '[:func2:3] def func2(): 1 / 0 >>> ZeroDivisionError(division by zero)' >>> try: ... # test with default filter(filename skip site-packages) ... from pip._internal.utils.compatibility_tags import version_info_to_nodot ... version_info_to_nodot(0) ... except Exception as e: ... format_error(e) - "[::4] version_info_to_nodot(0) >>> TypeError('int' object is not subscriptable)" + "[::4] version_info_to_nodot(0) >>> TypeError('int' object is not subscriptable)" >>> try: ... # test without filter ... from pip._internal.utils.compatibility_tags import version_info_to_nodot ... version_info_to_nodot(0) ... except Exception as e: ... format_error(e, filter=None) - '[compatibility_tags.py:version_info_to_nodot:23] return "".join(map(str, version_info[:2])) >>> TypeError(\'int\' object is not subscriptable)' + '[::4 | compatibility_tags.py:version_info_to_nodot:23] return "".join(map(str, version_info[:2])) >>> TypeError(\'int\' object is not subscriptable)' >>> try: ... # test with custom filter. ... from pip._internal.utils.compatibility_tags import version_info_to_nodot ... version_info_to_nodot(0) ... except Exception as e: ... format_error(e, filter=lambda i: '::4] version_info_to_nodot(0) >>> TypeError('int' object is not subscriptable)" + "[::4] version_info_to_nodot(0) >>> TypeError('int' object is not subscriptable)" --- @@ -472,10 +479,12 @@ Examples: - >>> set_pid_file().name # Assuming this is the first run, should succeed + >>> path = set_pid_file() # Assuming this is the first run, should succeed + >>> path.name 'Lib__doctest.py.pid' >>> set_pid_file() # Simulating second instance trying to start, should raise error if raise_error=True False + >>> path.unlink() --- @@ -909,6 +918,37 @@ :return: A byte sequence representing the constructed HTTP response. +--- + + + 4.7 `custom_dns` - Custom the DNS of socket.getaddrinfo, only effect current thread. + + [WARNING] This will modify the global socket.getaddrinfo. + + >>> from concurrent.futures import ThreadPoolExecutor + >>> # this only effect current thread + >>> custom_dns({"1.1.1.1": ("127.0.0.1", 80), ("1.1.1.1", 80): ("192.168.0.1", 443)}) + >>> socket.getaddrinfo('1.1.1.1', 80)[0][-1] + ('192.168.0.1', 443) + >>> socket.getaddrinfo('1.1.1.1', 8888)[0][-1] + ('127.0.0.1', 80) + >>> ThreadPoolExecutor().submit(lambda : socket.getaddrinfo('1.1.1.1', 8888)[0][-1]).result() + ('1.1.1.1', 8888) + >>> # this effect global socket.getaddrinfo + >>> custom_dns({"1.1.1.1": ("127.0.0.1", 80), ("1.1.1.1", 80): ("192.168.0.1", 443)}, thread=False) + >>> ThreadPoolExecutor().submit(lambda : socket.getaddrinfo('1.1.1.1', 8888)[0][-1]).result() + ('127.0.0.1', 80) + + Demo: + + custom_dns(custom={("MY_PROXY_HOST", 80): ("xxxxxxxxx", 43532)}) + print( + requests.get( + "https://www.github.com/", proxies={"all": "http://MY_PROXY_HOST"} + ).text + ) + + --- ====================== diff --git a/morebuiltins/request.py b/morebuiltins/request.py index 675e08a..eee2543 100644 --- a/morebuiltins/request.py +++ b/morebuiltins/request.py @@ -21,6 +21,7 @@ "update_url", "get_lan_ip", "make_response", + "custom_dns", ] diff --git a/morebuiltins/utils.py b/morebuiltins/utils.py index 2559a60..ef91cc4 100644 --- a/morebuiltins/utils.py +++ b/morebuiltins/utils.py @@ -14,7 +14,7 @@ from enum import IntEnum from functools import wraps from itertools import groupby, islice -from os.path import basename +from os.path import basename, exists from pathlib import Path from time import gmtime, mktime, strftime, strptime, time, timezone from typing import ( @@ -635,63 +635,89 @@ def _tb_filter(tb: traceback.FrameSummary): def format_error( error: BaseException, - index=-1, + index: Union[int, slice] = slice(-3, None, None), filter: Optional[Callable] = _tb_filter, - template="[{filename}:{tb.name}:{tb.lineno}] {tb.line} >>> {error.__class__.__name__}({error!s})", + template="[{trace_routes}] {error_line} >>> {error.__class__.__name__}({error!s:.100})", **kwargs, ) -> str: - """Extracts frame information from an exception, with an option to filter out “site-packages” details by default. + r"""Extracts frame information from an exception, with an option to filter out “site-packages” details by default. >>> try: ... # test default ... 1 / 0 ... except Exception as e: ... format_error(e) - '[::3] 1 / 0 >>> ZeroDivisionError(division by zero)' + '[::3] 1 / 0 >>> ZeroDivisionError(division by zero)' >>> try: ... # test in function ... def func1(): 1 / 0 ... func1() ... except Exception as e: ... format_error(e) - '[:func1:3] def func1(): 1 / 0 >>> ZeroDivisionError(division by zero)' + '[::4 | :func1:3] def func1(): 1 / 0 >>> ZeroDivisionError(division by zero)' >>> try: ... # test index ... def func2(): 1 / 0 ... func2() ... except Exception as e: ... format_error(e, index=0) - '[::4] func2() >>> ZeroDivisionError(division by zero)' + '[::4] func2() >>> ZeroDivisionError(division by zero)' + >>> try: + ... # test slice index + ... def func2(): 1 / 0 + ... func2() + ... except Exception as e: + ... format_error(e, index=slice(-1, None, None)) + '[:func2:3] def func2(): 1 / 0 >>> ZeroDivisionError(division by zero)' >>> try: ... # test with default filter(filename skip site-packages) ... from pip._internal.utils.compatibility_tags import version_info_to_nodot ... version_info_to_nodot(0) ... except Exception as e: ... format_error(e) - "[::4] version_info_to_nodot(0) >>> TypeError('int' object is not subscriptable)" + "[::4] version_info_to_nodot(0) >>> TypeError('int' object is not subscriptable)" >>> try: ... # test without filter ... from pip._internal.utils.compatibility_tags import version_info_to_nodot ... version_info_to_nodot(0) ... except Exception as e: ... format_error(e, filter=None) - '[compatibility_tags.py:version_info_to_nodot:23] return "".join(map(str, version_info[:2])) >>> TypeError(\\'int\\' object is not subscriptable)' + '[::4 | compatibility_tags.py:version_info_to_nodot:23] return "".join(map(str, version_info[:2])) >>> TypeError(\'int\' object is not subscriptable)' >>> try: ... # test with custom filter. ... from pip._internal.utils.compatibility_tags import version_info_to_nodot ... version_info_to_nodot(0) ... except Exception as e: ... format_error(e, filter=lambda i: '::4] version_info_to_nodot(0) >>> TypeError('int' object is not subscriptable)" + "[::4] version_info_to_nodot(0) >>> TypeError('int' object is not subscriptable)" """ try: filter = filter or always_return_value(True) tbs = [tb for tb in traceback.extract_tb(error.__traceback__) if filter(tb)] - tb = tbs[index] - _kwargs = dict(locals()) + if isinstance(index, slice): + tbs = tbs[index] + elif isinstance(index, int): + tbs = [tbs[index]] + else: + raise ValueError("Invalid index type") + trace_route_list = [] + for tb in tbs: + filename = tb.filename + if exists(filename): + _basename = basename(filename) + elif filename[0] == "<": + _basename = f"{filename.split()[0]}>" + else: + _basename = filename + trace_route_list.append(f"{_basename}:{tb.name}:{tb.lineno}") + trace_routes = " | ".join(trace_route_list) + _kwargs = { + "tbs": tbs, + "error": error, + "error_line": tbs[-1].line, + "trace_routes": trace_routes, + } _kwargs.update(kwargs) - if "filename" not in _kwargs: - _kwargs["filename"] = basename(tb.filename) return template.format_map(_kwargs) except IndexError: return "" @@ -917,10 +943,12 @@ def set_pid_file( Examples: - >>> set_pid_file().name # Assuming this is the first run, should succeed + >>> path = set_pid_file() # Assuming this is the first run, should succeed + >>> path.name 'Lib__doctest.py.pid' >>> set_pid_file() # Simulating second instance trying to start, should raise error if raise_error=True False + >>> path.unlink() """ if path is None: if not default_name: