codekingpro/portable-devtools

1import functools2import json3import multiprocessing4import os5import threading6from contextlib import contextmanager7from threading import Thread8 9from ._colorizer import Colorizer10from ._locks_machinery import create_handler_lock11 12 13def prepare_colored_format(format_, ansi_level):14    colored = Colorizer.prepare_format(format_)15    return colored, colored.colorize(ansi_level)16 17 18def prepare_stripped_format(format_):19    colored = Colorizer.prepare_format(format_)20    return colored.strip()21 22 23def memoize(function):24    return functools.lru_cache(maxsize=64)(function)25 26 27class Message(str):28    __slots__ = ("record",)29 30 31class Handler:32    def __init__(33        self,34        *,35        sink,36        name,37        levelno,38        formatter,39        is_formatter_dynamic,40        filter_,41        colorize,42        serialize,43        enqueue,44        multiprocessing_context,45        error_interceptor,46        exception_formatter,47        id_,48        levels_ansi_codes49    ):50        self._name = name51        self._sink = sink52        self._levelno = levelno53        self._formatter = formatter54        self._is_formatter_dynamic = is_formatter_dynamic55        self._filter = filter_56        self._colorize = colorize57        self._serialize = serialize58        self._enqueue = enqueue59        self._multiprocessing_context = multiprocessing_context60        self._error_interceptor = error_interceptor61        self._exception_formatter = exception_formatter62        self._id = id_63        self._levels_ansi_codes = levels_ansi_codes  # Warning, reference shared among handlers64 65        self._decolorized_format = None66        self._precolorized_formats = {}67        self._memoize_dynamic_format = None68 69        self._stopped = False70        self._lock = create_handler_lock()71        self._lock_acquired = threading.local()72        self._queue = None73        self._queue_lock = None74        self._confirmation_event = None75        self._confirmation_lock = None76        self._owner_process_pid = None77        self._thread = None78 79        if self._is_formatter_dynamic:80            if self._colorize:81                self._memoize_dynamic_format = memoize(prepare_colored_format)82            else:83                self._memoize_dynamic_format = memoize(prepare_stripped_format)84        else:85            if self._colorize:86                for level_name in self._levels_ansi_codes:87                    self.update_format(level_name)88            else:89                self._decolorized_format = self._formatter.strip()90 91        if self._enqueue:92            if self._multiprocessing_context is None:93                self._queue = multiprocessing.SimpleQueue()94                self._confirmation_event = multiprocessing.Event()95                self._confirmation_lock = multiprocessing.Lock()96            else:97                self._queue = self._multiprocessing_context.SimpleQueue()98                self._confirmation_event = self._multiprocessing_context.Event()99                self._confirmation_lock = self._multiprocessing_context.Lock()100            self._queue_lock = create_handler_lock()101            self._owner_process_pid = os.getpid()102            self._thread = Thread(103                target=self._queued_writer, daemon=True, name="loguru-writer-%d" % self._id104            )105            self._thread.start()106 107    def __repr__(self):108        return "(id=%d, level=%d, sink=%s)" % (self._id, self._levelno, self._name)109 110    @contextmanager111    def _protected_lock(self):112        """Acquire the lock, but fail fast if its already acquired by the current thread."""113        if getattr(self._lock_acquired, "acquired", False):114            raise RuntimeError(115                "Could not acquire internal lock because it was already in use (deadlock avoided). "116                "This likely happened because the logger was re-used inside a sink, a signal "117                "handler or a '__del__' method. This is not permitted because the logger and its "118                "handlers are not re-entrant."119            )120        self._lock_acquired.acquired = True121        try:122            with self._lock:123                yield124        finally:125            self._lock_acquired.acquired = False126 127    def emit(self, record, level_id, from_decorator, is_raw, colored_message):128        try:129            if self._levelno > record["level"].no:130                return131 132            if self._filter is not None:133                if not self._filter(record):134                    return135 136            if self._is_formatter_dynamic:137                dynamic_format = self._formatter(record)138 139            formatter_record = record.copy()140 141            if not record["exception"]:142                formatter_record["exception"] = ""143            else:144                type_, value, tb = record["exception"]145                formatter = self._exception_formatter146                lines = formatter.format_exception(type_, value, tb, from_decorator=from_decorator)147                formatter_record["exception"] = "".join(lines)148 149            if colored_message is not None and colored_message.stripped != record["message"]:150                colored_message = None151 152            if is_raw:153                if colored_message is None or not self._colorize:154                    formatted = record["message"]155                else:156                    ansi_level = self._levels_ansi_codes[level_id]157                    formatted = colored_message.colorize(ansi_level)158            elif self._is_formatter_dynamic:159                if not self._colorize:160                    precomputed_format = self._memoize_dynamic_format(dynamic_format)161                    formatted = precomputed_format.format_map(formatter_record)162                elif colored_message is None:163                    ansi_level = self._levels_ansi_codes[level_id]164                    _, precomputed_format = self._memoize_dynamic_format(dynamic_format, ansi_level)165                    formatted = precomputed_format.format_map(formatter_record)166                else:167                    ansi_level = self._levels_ansi_codes[level_id]168                    formatter, precomputed_format = self._memoize_dynamic_format(169                        dynamic_format, ansi_level170                    )171                    coloring_message = formatter.make_coloring_message(172                        record["message"], ansi_level=ansi_level, colored_message=colored_message173                    )174                    formatter_record["message"] = coloring_message175                    formatted = precomputed_format.format_map(formatter_record)176 177            else:178                if not self._colorize:179                    precomputed_format = self._decolorized_format180                    formatted = precomputed_format.format_map(formatter_record)181                elif colored_message is None:182                    ansi_level = self._levels_ansi_codes[level_id]183                    precomputed_format = self._precolorized_formats[level_id]184                    formatted = precomputed_format.format_map(formatter_record)185                else:186                    ansi_level = self._levels_ansi_codes[level_id]187                    precomputed_format = self._precolorized_formats[level_id]188                    coloring_message = self._formatter.make_coloring_message(189                        record["message"], ansi_level=ansi_level, colored_message=colored_message190                    )191                    formatter_record["message"] = coloring_message192                    formatted = precomputed_format.format_map(formatter_record)193 194            if self._serialize:195                formatted = self._serialize_record(formatted, record)196 197            str_record = Message(formatted)198            str_record.record = record199 200            with self._protected_lock():201                if self._stopped:202                    return203                if self._enqueue:204                    self._queue.put(str_record)205                else:206                    self._sink.write(str_record)207        except Exception:208            if not self._error_interceptor.should_catch():209                raise210            self._error_interceptor.print(record)211 212    def stop(self):213        with self._protected_lock():214            self._stopped = True215            if self._enqueue:216                if self._owner_process_pid != os.getpid():217                    return218                self._queue.put(None)219                self._thread.join()220                if hasattr(self._queue, "close"):221                    self._queue.close()222 223            self._sink.stop()224 225    def complete_queue(self):226        if not self._enqueue:227            return228 229        with self._confirmation_lock:230            self._queue.put(True)231            self._confirmation_event.wait()232            self._confirmation_event.clear()233 234    def tasks_to_complete(self):235        if self._enqueue and self._owner_process_pid != os.getpid():236            return []237        lock = self._queue_lock if self._enqueue else self._protected_lock()238        with lock:239            return self._sink.tasks_to_complete()240 241    def update_format(self, level_id):242        if not self._colorize or self._is_formatter_dynamic:243            return244        ansi_code = self._levels_ansi_codes[level_id]245        self._precolorized_formats[level_id] = self._formatter.colorize(ansi_code)246 247    @property248    def levelno(self):249        return self._levelno250 251    @staticmethod252    def _serialize_record(text, record):253        exception = record["exception"]254 255        if exception is not None:256            exception = {257                "type": None if exception.type is None else exception.type.__name__,258                "value": exception.value,259                "traceback": bool(exception.traceback),260            }261 262        serializable = {263            "text": text,264            "record": {265                "elapsed": {266                    "repr": record["elapsed"],267                    "seconds": record["elapsed"].total_seconds(),268                },269                "exception": exception,270                "extra": record["extra"],271                "file": {"name": record["file"].name, "path": record["file"].path},272                "function": record["function"],273                "level": {274                    "icon": record["level"].icon,275                    "name": record["level"].name,276                    "no": record["level"].no,277                },278                "line": record["line"],279                "message": record["message"],280                "module": record["module"],281                "name": record["name"],282                "process": {"id": record["process"].id, "name": record["process"].name},283                "thread": {"id": record["thread"].id, "name": record["thread"].name},284                "time": {"repr": record["time"], "timestamp": record["time"].timestamp()},285            },286        }287 288        return json.dumps(serializable, default=str, ensure_ascii=False) + "\n"289 290    def _queued_writer(self):291        message = None292        queue = self._queue293 294        # We need to use a lock to protect sink during fork.295        # Particularly, writing to stderr may lead to deadlock in child process.296        lock = self._queue_lock297 298        while True:299            try:300                message = queue.get()301            except Exception:302                with lock:303                    self._error_interceptor.print(None)304                continue305 306            if message is None:307                break308 309            if message is True:310                self._confirmation_event.set()311                continue312 313            with lock:314                try:315                    self._sink.write(message)316                except Exception:317                    self._error_interceptor.print(message.record)318 319    def __getstate__(self):320        state = self.__dict__.copy()321        state["_lock"] = None322        state["_lock_acquired"] = None323        state["_memoize_dynamic_format"] = None324        if self._enqueue:325            state["_sink"] = None326            state["_thread"] = None327            state["_owner_process"] = None328            state["_queue_lock"] = None329        return state330 331    def __setstate__(self, state):332        self.__dict__.update(state)333        self._lock = create_handler_lock()334        self._lock_acquired = threading.local()335        if self._enqueue:336            self._queue_lock = create_handler_lock()337        if self._is_formatter_dynamic:338            if self._colorize:339                self._memoize_dynamic_format = memoize(prepare_colored_format)340            else:341                self._memoize_dynamic_format = memoize(prepare_stripped_format)342