pocketutils.tools.sys_tools.SystemTools.env_var_flag() - Code Metrics - Inspection of "feat: env_var_flag" - dmyersturnbull/pocketutils - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — main ( 6412b4...98098e )

by Douglas

created 2021-11-10 19:00 UTC

SystemTools.env_var_flag() A

↳ Parent: pocketutils.tools.sys_tools

Complexity

Conditions

Size

Total Lines	7
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	3
nop	2
dl	0
loc	7
rs	10
c	0
b	0
f	0

"""
Low-level tools (e.g. memory management).
"""
import atexit
import importlib
import locale
import logging
import os
import platform
import socket
import signal
import struct
import sys
import traceback
from dataclasses import dataclass, asdict
from datetime import timezone, datetime
from getpass import getuser
from typing import Any, Callable, Union, Sequence, Mapping, Optional, NamedTuple

from pocketutils.tools.common_tools import CommonTools
from pocketutils.core.input_output import Writeable

from pocketutils.tools.base_tools import BaseTools

logger = logging.getLogger("pocketutils")


@dataclass(frozen=True, repr=True, order=True)

class Frame:
    depth: int
    filename: str
    line: int
    name: str
    repeats: int

    def as_dict(self) -> Mapping[str, Union[int, str]]:

        return asdict(self)


class SerializedException(NamedTuple):

    message: Sequence[str]
    stacktrace: Sequence[Frame]


@dataclass(frozen=True, repr=True)

class SignalHandler:
    name: str
    code: int
    desc: str
    sink: Union[Writeable, Callable[[str], Any]]

    def __call__(self):
        self.sink.write(f"~~{self.name}[{self.code}] ({self.desc})~~")
        traceback.print_stack(file=self.sink)
        for line in traceback.format_stack():
            self.sink.write(line)


@dataclass(frozen=True, repr=True)

class ExitHandler:
    sink: Writeable

    def __call__(self):
        self.sink.write(f"~~EXIT~~")

        traceback.print_stack(file=self.sink)
        for line in traceback.format_stack():
            self.sink.write(line)


class SystemTools(BaseTools):

    @classmethod
    def get_env_info(cls, *, include_insecure: bool = False) -> Mapping[str, str]:
        """
        Get a dictionary of some system and environment information.
        Includes os_release, hostname, username, mem + disk, shell, etc.

        Args:
            include_insecure: Include data like hostname and username

        .. caution ::
            Even with ``include_insecure=False``, avoid exposing this data to untrusted
            sources. For example, this includes the specific OS release, which could
            be used in attack.
        """
        try:
            import psutil

        except ImportError:
            psutil = None
            logger.warning("psutil is not installed, so cannot get extended env info")

        now = datetime.now(timezone.utc).astimezone().isoformat()
        uname = platform.uname()
        lang_code, encoding = locale.getlocale()
        # build up this dict:
        data = {}

        def _try(os_fn, k: str, *args):
            if any((a is None for a in args)):

                return None
            try:
                v = os_fn(*args)

                data[k] = v
                return v
            except (OSError, ImportError):
                return None

        data.update(
            dict(
                platform=platform.platform(),
                python=".".join(str(i) for i in sys.version_info),
                os=uname.system,
                os_release=uname.release,
                os_version=uname.version,
                machine=uname.machine,
                byte_order=sys.byteorder,
                processor=uname.processor,
                build=sys.version,
                python_bits=8 * struct.calcsize("P"),
                environment_info_capture_datetime=now,
                encoding=encoding,
                lang_code=lang_code,
                recursion_limit=sys.getrecursionlimit(),
                float_info=sys.float_info,
                int_info=sys.int_info,
                flags=sys.flags,
                hash_info=sys.hash_info,
                implementation=sys.implementation,
                switch_interval=sys.getswitchinterval(),
                filesystem_encoding=sys.getfilesystemencoding(),
            )
        )
        if "LANG" in os.environ:
            data["lang"] = os.environ["LANG"]
        if "SHELL" in os.environ:
            data["shell"] = os.environ["SHELL"]
        if "LC_ALL" in os.environ:
            data["lc_all"] = os.environ["LC_ALL"]
        if hasattr(sys, "winver"):
            data["win_ver"] = sys.getwindowsversion()

        if hasattr(sys, "mac_ver"):
            data["mac_ver"] = sys.mac_ver()

        if hasattr(sys, "linux_distribution"):
            data["linux_distribution"] = sys.linux_distribution()

        if include_insecure:
            _try(getuser, "username")
            _try(os.getlogin, "login")
            _try(socket.gethostname, "hostname")
            _try(os.getcwd, "cwd")
            pid = _try(os.getpid, "pid")
            ppid = _try(os.getppid, "parent_pid")
            if hasattr(os, "getpriority"):
                _try(os.getpriority, "priority", os.PRIO_PROCESS, pid)
                _try(os.getpriority, "parent_priority", os.PRIO_PROCESS, ppid)
        if psutil is not None:
            data.update(
                dict(
                    disk_used=psutil.disk_usage(".").used,
                    disk_free=psutil.disk_usage(".").free,
                    memory_used=psutil.virtual_memory().used,
                    memory_available=psutil.virtual_memory().available,
                )
            )
        return {k: str(v) for k, v in dict(data).items()}

    @classmethod
    def list_package_versions(cls) -> Mapping[str, str]:
        """
        Returns installed packages and their version numbers.
        Reliable; uses importlib (Python 3.8+).
        """
        # calling .metadata reads the metadata file
        # and .version is an alias to .metadata["version"]
        # so make sure to only read once
        # TODO: get installed extras?

        dct = {}
        for d in importlib.metadata.distributions():

            meta = d.metadata
            dct[meta["name"]] = meta["version"]
        return dct

    @classmethod
    def serialize_exception(cls, e: Optional[BaseException]) -> SerializedException:

        tbe = traceback.TracebackException.from_exception(e)
        msg = [] if e is None else list(tbe.format_exception_only())
        tb = SystemTools.build_traceback(e)

        return SerializedException(msg, tb)

    @classmethod
    def serialize_exception_msg(cls, e: Optional[BaseException]) -> Sequence[str]:

        tbe = traceback.TracebackException.from_exception(e)
        return [] if e is None else list(tbe.format_exception_only())

    @classmethod
    def build_traceback(cls, e: Optional[BaseException]) -> Sequence[Frame]:

        if e is None:
            return []
        tb = []

        current = None
        tbe = traceback.TracebackException.from_exception(e)
        last, repeats = None, 0
        for i, s in enumerate(tbe.stack):

            current = Frame(depth=i, filename=s.filename, line=s.line, name=s.name, repeats=-1)
            if current == last:
                repeats += 1
            else:
                current = Frame(
                    depth=current.depth,
                    filename=current.filename,
                    line=current.line,
                    name=current.name,
                    repeats=repeats,
                )
                tb.append(current)
                repeats = 0
            last = current
        if current is not None and current == last:
            tb.append(current)
        return tb

    @classmethod
    def trace_signals(cls, sink: Writeable = sys.stderr) -> None:
        """
        Registers signal handlers for all signals that log the traceback.
        Uses ``signal.signal``.
        """
        for sig in signal.valid_signals():

            handler = SignalHandler(sig.name, sig.value, signal.strsignal(sig), sink)

            signal.signal(sig.value, handler)

    @classmethod
    def trace_exit(cls, sink: Writeable = sys.stderr) -> None:
        """
        Registers an exit handler via ``atexit.register`` that logs the traceback.
        """
        atexit.register(ExitHandler(sink))

    @classmethod
    def env_var_flag(cls, name: str) -> bool:
        """
        Returns the boolean-parsed value of an environment variable (False if missing).
        Uses :meth:`pocketutils.tools.common_tools.CommonTools.parse_bool_flex`.
        """
        return CommonTools.parse_bool_flex(os.environ.get(name, "false"))


__all__ = ["SignalHandler", "ExitHandler", "SystemTools"]


1			"""
2			Low-level tools (e.g. memory management).
3			"""
4			import atexit
5			import importlib
6			import locale
7			import logging
8			import os
9			import platform
10			import socket
11			import signal
12			import struct
13			import sys
14			import traceback
15			from dataclasses import dataclass, asdict
16			from datetime import timezone, datetime
17			from getpass import getuser
18			from typing import Any, Callable, Union, Sequence, Mapping, Optional, NamedTuple
19
20			from pocketutils.tools.common_tools import CommonTools
21			from pocketutils.core.input_output import Writeable
			0 ignored issues – show Bug introduced 2021-11-01 01:15 UTC by Report Bug Copy Issue Report The name `core` does not seem to exist in module `pocketutils`. Loading history...
22			from pocketutils.tools.base_tools import BaseTools
23
24			logger = logging.getLogger("pocketutils")
25
26
27			@dataclass(frozen=True, repr=True, order=True)
			0 ignored issues – show introduced 2021-11-01 01:15 UTC by Report Bug Copy Issue Report Missing class docstring Loading history...
28			class Frame:
29			depth: int
30			filename: str
31			line: int
32			name: str
33			repeats: int
34
35			def as_dict(self) -> Mapping[str, Union[int, str]]:
			0 ignored issues – show introduced 2021-11-02 18:57 UTC by Report Bug Copy Issue Report Missing function or method docstring Loading history...
36			return asdict(self)
37
38
39			class SerializedException(NamedTuple):
			0 ignored issues – show introduced 2021-11-01 01:15 UTC by Report Bug Copy Issue Report Missing class docstring Loading history...
40			message: Sequence[str]
41			stacktrace: Sequence[Frame]
42
43
44			@dataclass(frozen=True, repr=True)
			0 ignored issues – show introduced 2021-11-01 01:15 UTC by Report Bug Copy Issue Report Missing class docstring Loading history...
45			class SignalHandler:
46			name: str
47			code: int
48			desc: str
49			sink: Union[Writeable, Callable[[str], Any]]
50
51			def __call__(self):
52			self.sink.write(f"~~{self.name}[{self.code}] ({self.desc})~~")
53			traceback.print_stack(file=self.sink)
54			for line in traceback.format_stack():
55			self.sink.write(line)
56
57
58			@dataclass(frozen=True, repr=True)
			0 ignored issues – show introduced 2021-11-02 21:45 UTC by Report Bug Copy Issue Report Missing class docstring Loading history...
59			class ExitHandler:
60			sink: Writeable
61
62			def __call__(self):
63			self.sink.write(f"~~EXIT~~")
			0 ignored issues – show introduced 2021-11-01 01:15 UTC by Report Bug Copy Issue Report Using an f-string that does not have any interpolated variables Loading history...
64			traceback.print_stack(file=self.sink)
65			for line in traceback.format_stack():
66			self.sink.write(line)
67
68
69			class SystemTools(BaseTools):
			0 ignored issues – show introduced 2021-11-02 21:45 UTC by Report Bug Copy Issue Report Missing class docstring Loading history...
70			@classmethod
71			def get_env_info(cls, *, include_insecure: bool = False) -> Mapping[str, str]:
72			"""
73			Get a dictionary of some system and environment information.
74			Includes os_release, hostname, username, mem + disk, shell, etc.
75
76			Args:
77			include_insecure: Include data like hostname and username
78
79			.. caution ::
80			Even with ``include_insecure=False``, avoid exposing this data to untrusted
81			sources. For example, this includes the specific OS release, which could
82			be used in attack.
83			"""
84			try:
85			import psutil
			0 ignored issues – show introduced 2021-01-25 05:25 UTC by Report Bug Copy Issue Report Import outside toplevel (psutil) Loading history...
86			except ImportError:
87			psutil = None
88			logger.warning("psutil is not installed, so cannot get extended env info")
89
90			now = datetime.now(timezone.utc).astimezone().isoformat()
91			uname = platform.uname()
92			lang_code, encoding = locale.getlocale()
93			# build up this dict:
94			data = {}
95
96			def _try(os_fn, k: str, *args):
97			if any((a is None for a in args)):
			0 ignored issues – show Comprehensibility Best Practice introduced 2021-10-05 00:06 UTC by Report Bug Copy Issue Report The variable `a` does not seem to be defined. Loading history...
98			return None
99			try:
100			v = os_fn(*args)
			0 ignored issues – show Coding Style Naming introduced 2021-11-02 21:45 UTC by Report Bug Copy Issue Report Variable name "v" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,\|_[^\\WA-Z]*\|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
101			data[k] = v
102			return v
103			except (OSError, ImportError):
104			return None
105
106			data.update(
107			dict(
108			platform=platform.platform(),
109			python=".".join(str(i) for i in sys.version_info),
110			os=uname.system,
111			os_release=uname.release,
112			os_version=uname.version,
113			machine=uname.machine,
114			byte_order=sys.byteorder,
115			processor=uname.processor,
116			build=sys.version,
117			python_bits=8 * struct.calcsize("P"),
118			environment_info_capture_datetime=now,
119			encoding=encoding,
120			lang_code=lang_code,
121			recursion_limit=sys.getrecursionlimit(),
122			float_info=sys.float_info,
123			int_info=sys.int_info,
124			flags=sys.flags,
125			hash_info=sys.hash_info,
126			implementation=sys.implementation,
127			switch_interval=sys.getswitchinterval(),
128			filesystem_encoding=sys.getfilesystemencoding(),
129			)
130			)
131			if "LANG" in os.environ:
132			data["lang"] = os.environ["LANG"]
133			if "SHELL" in os.environ:
134			data["shell"] = os.environ["SHELL"]
135			if "LC_ALL" in os.environ:
136			data["lc_all"] = os.environ["LC_ALL"]
137			if hasattr(sys, "winver"):
138			data["win_ver"] = sys.getwindowsversion()
			0 ignored issues – show Bug introduced 2021-10-07 00:43 UTC by Report Bug Copy Issue Report The Module `sys` does not seem to have a member named `getwindowsversion`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
139			if hasattr(sys, "mac_ver"):
140			data["mac_ver"] = sys.mac_ver()
			0 ignored issues – show Bug introduced 2021-10-07 00:43 UTC by Report Bug Copy Issue Report The Module `sys` does not seem to have a member named `mac_ver`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
141			if hasattr(sys, "linux_distribution"):
142			data["linux_distribution"] = sys.linux_distribution()
			0 ignored issues – show Bug introduced 2021-10-07 00:43 UTC by Report Bug Copy Issue Report The Module `sys` does not seem to have a member named `linux_distribution`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
143			if include_insecure:
144			_try(getuser, "username")
145			_try(os.getlogin, "login")
146			_try(socket.gethostname, "hostname")
147			_try(os.getcwd, "cwd")
148			pid = _try(os.getpid, "pid")
149			ppid = _try(os.getppid, "parent_pid")
150			if hasattr(os, "getpriority"):
151			_try(os.getpriority, "priority", os.PRIO_PROCESS, pid)
152			_try(os.getpriority, "parent_priority", os.PRIO_PROCESS, ppid)
153			if psutil is not None:
154			data.update(
155			dict(
156			disk_used=psutil.disk_usage(".").used,
157			disk_free=psutil.disk_usage(".").free,
158			memory_used=psutil.virtual_memory().used,
159			memory_available=psutil.virtual_memory().available,
160			)
161			)
162			return {k: str(v) for k, v in dict(data).items()}
163
164			@classmethod
165			def list_package_versions(cls) -> Mapping[str, str]:
166			"""
167			Returns installed packages and their version numbers.
168			Reliable; uses importlib (Python 3.8+).
169			"""
170			# calling .metadata reads the metadata file
171			# and .version is an alias to .metadata["version"]
172			# so make sure to only read once
173			# TODO: get installed extras?
			0 ignored issues – show Coding Style introduced 2021-10-07 00:43 UTC by Report Bug Copy Issue Report `TODO` and `FIXME` comments should generally be avoided. Loading history...
174			dct = {}
175			for d in importlib.metadata.distributions():
			0 ignored issues – show Bug introduced 2021-10-07 00:43 UTC by Report Bug Copy Issue Report The Module `importlib` does not seem to have a member named `metadata`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history... Coding Style Naming introduced 2021-10-07 00:43 UTC by Report Bug Copy Issue Report Variable name "d" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,\|_[^\\WA-Z]*\|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
176			meta = d.metadata
177			dct[meta["name"]] = meta["version"]
178			return dct
179
180			@classmethod
181			def serialize_exception(cls, e: Optional[BaseException]) -> SerializedException:
			0 ignored issues – show Coding Style Naming introduced 2021-11-02 18:57 UTC by Report Bug Copy Issue Report Argument name "e" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,\|_[^\\WA-Z]*\|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history... introduced 2021-11-02 21:45 UTC by Report Bug Copy Issue Report Missing function or method docstring Loading history...
182			tbe = traceback.TracebackException.from_exception(e)
183			msg = [] if e is None else list(tbe.format_exception_only())
184			tb = SystemTools.build_traceback(e)
			0 ignored issues – show Coding Style Naming introduced 2021-11-02 18:57 UTC by Report Bug Copy Issue Report Variable name "tb" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,\|_[^\\WA-Z]*\|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
185			return SerializedException(msg, tb)
186
187			@classmethod
188			def serialize_exception_msg(cls, e: Optional[BaseException]) -> Sequence[str]:
			0 ignored issues – show Coding Style Naming introduced 2021-11-02 21:45 UTC by Report Bug Copy Issue Report Argument name "e" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,\|_[^\\WA-Z]*\|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history... introduced 2021-11-02 21:45 UTC by Report Bug Copy Issue Report Missing function or method docstring Loading history...
189			tbe = traceback.TracebackException.from_exception(e)
190			return [] if e is None else list(tbe.format_exception_only())
191
192			@classmethod
193			def build_traceback(cls, e: Optional[BaseException]) -> Sequence[Frame]:
			0 ignored issues – show introduced 2021-11-02 21:45 UTC by Report Bug Copy Issue Report Missing function or method docstring Loading history... Coding Style Naming introduced 2021-11-02 21:45 UTC by Report Bug Copy Issue Report Argument name "e" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,\|_[^\\WA-Z]*\|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
194			if e is None:
195			return []
196			tb = []
			0 ignored issues – show Coding Style Naming introduced 2021-11-02 21:45 UTC by Report Bug Copy Issue Report Variable name "tb" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,\|_[^\\WA-Z]*\|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
197			current = None
198			tbe = traceback.TracebackException.from_exception(e)
199			last, repeats = None, 0
200			for i, s in enumerate(tbe.stack):
			0 ignored issues – show Coding Style Naming introduced 2021-11-02 18:57 UTC by Report Bug Copy Issue Report Variable name "s" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,\|_[^\\WA-Z]*\|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
201			current = Frame(depth=i, filename=s.filename, line=s.line, name=s.name, repeats=-1)
202			if current == last:
203			repeats += 1
204			else:
205			current = Frame(
206			depth=current.depth,
207			filename=current.filename,
208			line=current.line,
209			name=current.name,
210			repeats=repeats,
211			)
212			tb.append(current)
213			repeats = 0
214			last = current
215			if current is not None and current == last:
216			tb.append(current)
217			return tb
218
219			@classmethod
220			def trace_signals(cls, sink: Writeable = sys.stderr) -> None:
221			"""
222			Registers signal handlers for all signals that log the traceback.
223			Uses ``signal.signal``.
224			"""
225			for sig in signal.valid_signals():
			0 ignored issues – show Bug introduced 2021-11-01 01:15 UTC by Report Bug Copy Issue Report The Module `signal` does not seem to have a member named `valid_signals`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
226			handler = SignalHandler(sig.name, sig.value, signal.strsignal(sig), sink)
			0 ignored issues – show Bug introduced 2021-11-01 01:15 UTC by Report Bug Copy Issue Report The Module `signal` does not seem to have a member named `strsignal`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
227			signal.signal(sig.value, handler)
228
229			@classmethod
230			def trace_exit(cls, sink: Writeable = sys.stderr) -> None:
231			"""
232			Registers an exit handler via ``atexit.register`` that logs the traceback.
233			"""
234			atexit.register(ExitHandler(sink))
235
236			@classmethod
237			def env_var_flag(cls, name: str) -> bool:
238			"""
239			Returns the boolean-parsed value of an environment variable (False if missing).
240			Uses :meth:`pocketutils.tools.common_tools.CommonTools.parse_bool_flex`.
241			"""
242			return CommonTools.parse_bool_flex(os.environ.get(name, "false"))
243
244
245			__all__ = ["SignalHandler", "ExitHandler", "SystemTools"]
246

dmyersturnbull / pocketutils

Push — main ( 6412b4...98098e )

SystemTools.env_var_flag() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like