core: synced vendor/ext_json with ce for better compatability

vcsserver/lib/_vendor/jsonlogger/__init__.py

@@ -1,243 +1,237 b''
 '''
 This library is provided to allow standard python logging
 to output log data as JSON formatted strings
 '''
 import logging
-import json
 import re
 from datetime import date, datetime, time, tzinfo, timedelta
 import traceback
 import importlib
 
 from inspect import istraceback
 
 from collections import OrderedDict
 
-
-def _inject_req_id(record, *args, **kwargs):
-    return record
-
-
-ExceptionAwareFormatter = logging.Formatter
-
+from ...logging_formatter import _inject_req_id, ExceptionAwareFormatter
+from ...ext_json import sjson as json
 
 ZERO = timedelta(0)
 HOUR = timedelta(hours=1)
 
 
 class UTC(tzinfo):
     """UTC"""
 
     def utcoffset(self, dt):
         return ZERO
 
     def tzname(self, dt):
         return "UTC"
 
     def dst(self, dt):
         return ZERO
 
 utc = UTC()
 
 
 # skip natural LogRecord attributes
 # http://docs.python.org/library/logging.html#logrecord-attributes
 RESERVED_ATTRS = (
     'args', 'asctime', 'created', 'exc_info', 'exc_text', 'filename',
     'funcName', 'levelname', 'levelno', 'lineno', 'module',
     'msecs', 'message', 'msg', 'name', 'pathname', 'process',
     'processName', 'relativeCreated', 'stack_info', 'thread', 'threadName')
 
 
 def merge_record_extra(record, target, reserved):
     """
     Merges extra attributes from LogRecord object into target dictionary
 
     :param record: logging.LogRecord
     :param target: dict to update
     :param reserved: dict or list with reserved keys to skip
     """
     for key, value in record.__dict__.items():
         # this allows to have numeric keys
         if (key not in reserved
             and not (hasattr(key, "startswith")
                      and key.startswith('_'))):
             target[key] = value
     return target
 
 
 class JsonEncoder(json.JSONEncoder):
     """
     A custom encoder extending the default JSONEncoder
     """
 
     def default(self, obj):
         if isinstance(obj, (date, datetime, time)):
             return self.format_datetime_obj(obj)
 
         elif istraceback(obj):
             return ''.join(traceback.format_tb(obj)).strip()
 
         elif type(obj) == Exception \
                 or isinstance(obj, Exception) \
                 or type(obj) == type:
             return str(obj)
 
         try:
             return super().default(obj)
 
         except TypeError:
             try:
                 return str(obj)
 
             except Exception:
                 return None
 
     def format_datetime_obj(self, obj):
         return obj.isoformat()
 
 
 class JsonFormatter(ExceptionAwareFormatter):
     """
     A custom formatter to format logging records as json strings.
     Extra values will be formatted as str() if not supported by
     json default encoder
     """
 
     def __init__(self, *args, **kwargs):
         """
         :param json_default: a function for encoding non-standard objects
             as outlined in http://docs.python.org/2/library/json.html
         :param json_encoder: optional custom encoder
         :param json_serializer: a :meth:`json.dumps`-compatible callable
             that will be used to serialize the log record.
         :param json_indent: an optional :meth:`json.dumps`-compatible numeric value
             that will be used to customize the indent of the output json.
         :param prefix: an optional string prefix added at the beginning of
             the formatted string
         :param json_indent: indent parameter for json.dumps
         :param json_ensure_ascii: ensure_ascii parameter for json.dumps
         :param reserved_attrs: an optional list of fields that will be skipped when
             outputting json log record. Defaults to all log record attributes:
             http://docs.python.org/library/logging.html#logrecord-attributes
         :param timestamp: an optional string/boolean field to add a timestamp when
             outputting the json log record. If string is passed, timestamp will be added
             to log record using string as key. If True boolean is passed, timestamp key
             will be "timestamp". Defaults to False/off.
         """
         self.json_default = self._str_to_fn(kwargs.pop("json_default", None))
         self.json_encoder = self._str_to_fn(kwargs.pop("json_encoder", None))
         self.json_serializer = self._str_to_fn(kwargs.pop("json_serializer", json.dumps))
         self.json_indent = kwargs.pop("json_indent", None)
         self.json_ensure_ascii = kwargs.pop("json_ensure_ascii", True)
         self.prefix = kwargs.pop("prefix", "")
         reserved_attrs = kwargs.pop("reserved_attrs", RESERVED_ATTRS)
         self.reserved_attrs = dict(list(zip(reserved_attrs, reserved_attrs)))
         self.timestamp = kwargs.pop("timestamp", True)
 
         # super(JsonFormatter, self).__init__(*args, **kwargs)
         logging.Formatter.__init__(self, *args, **kwargs)
         if not self.json_encoder and not self.json_default:
             self.json_encoder = JsonEncoder
 
         self._required_fields = self.parse()
         self._skip_fields = dict(list(zip(self._required_fields,
                                       self._required_fields)))
         self._skip_fields.update(self.reserved_attrs)
 
     def _str_to_fn(self, fn_as_str):
         """
         If the argument is not a string, return whatever was passed in.
         Parses a string such as package.module.function, imports the module
         and returns the function.
 
         :param fn_as_str: The string to parse. If not a string, return it.
         """
         if not isinstance(fn_as_str, str):
             return fn_as_str
 
         path, _, function = fn_as_str.rpartition('.')
         module = importlib.import_module(path)
         return getattr(module, function)
 
     def parse(self):
         """
         Parses format string looking for substitutions
 
         This method is responsible for returning a list of fields (as strings)
         to include in all log messages.
         """
         standard_formatters = re.compile(r'\((.+?)\)', re.IGNORECASE)
         return standard_formatters.findall(self._fmt)
 
     def add_fields(self, log_record, record, message_dict):
         """
         Override this method to implement custom logic for adding fields.
         """
         for field in self._required_fields:
             log_record[field] = record.__dict__.get(field)
         log_record.update(message_dict)
         merge_record_extra(record, log_record, reserved=self._skip_fields)
 
         if self.timestamp:
             key = self.timestamp if type(self.timestamp) == str else 'timestamp'
             log_record[key] = datetime.fromtimestamp(record.created, tz=utc)
 
     def process_log_record(self, log_record):
         """
         Override this method to implement custom logic
         on the possibly ordered dictionary.
         """
         return log_record
 
     def jsonify_log_record(self, log_record):
         """Returns a json string of the log record."""
         return self.json_serializer(log_record,
                                     default=self.json_default,
                                     cls=self.json_encoder,
                                     indent=self.json_indent,
                                     ensure_ascii=self.json_ensure_ascii)
 
     def serialize_log_record(self, log_record):
         """Returns the final representation of the log record."""
         return "{}{}".format(self.prefix, self.jsonify_log_record(log_record))
 
     def format(self, record):
         """Formats a log record and serializes to json"""
         message_dict = {}
         # FIXME: logging.LogRecord.msg and logging.LogRecord.message in typeshed
         #        are always type of str. We shouldn't need to override that.
         if isinstance(record.msg, dict):
             message_dict = record.msg
             record.message = None
         else:
             record.message = record.getMessage()
         # only format time if needed
         if "asctime" in self._required_fields:
             record.asctime = self.formatTime(record, self.datefmt)
 
         # Display formatted exception, but allow overriding it in the
         # user-supplied dict.
         if record.exc_info and not message_dict.get('exc_info'):
             message_dict['exc_info'] = self.formatException(record.exc_info)
         if not message_dict.get('exc_info') and record.exc_text:
             message_dict['exc_info'] = record.exc_text
         # Display formatted record of stack frames
         # default format is a string returned from :func:`traceback.print_stack`
         try:
             if record.stack_info and not message_dict.get('stack_info'):
                 message_dict['stack_info'] = self.formatStack(record.stack_info)
         except AttributeError:
             # Python2.7 doesn't have stack_info.
             pass
 
         try:
             log_record = OrderedDict()
         except NameError:
             log_record = {}
 
         _inject_req_id(record, with_prefix=False)
         self.add_fields(log_record, record, message_dict)
         log_record = self.process_log_record(log_record)
 
         return self.serialize_log_record(log_record)

vcsserver/lib/_vendor/redis_lock/__init__.py

@@ -1,394 +1,402 b''
 
 import threading
 import weakref
 from base64 import b64encode
 from logging import getLogger
 from os import urandom
 from typing import Union
 
 from redis import StrictRedis
 
 __version__ = '4.0.0'
 
 loggers = {
     k: getLogger("vcsserver." + ".".join((__name__, k)))
     for k in [
         "acquire",
         "refresh.thread.start",
         "refresh.thread.stop",
         "refresh.thread.exit",
         "refresh.start",
         "refresh.shutdown",
         "refresh.exit",
         "release",
     ]
 }
 
 text_type = str
 binary_type = bytes
 
 
 # Check if the id match. If not, return an error code.
 UNLOCK_SCRIPT = b"""
     if redis.call("get", KEYS[1]) ~= ARGV[1] then
         return 1
     else
         redis.call("del", KEYS[2])
         redis.call("lpush", KEYS[2], 1)
         redis.call("pexpire", KEYS[2], ARGV[2])
         redis.call("del", KEYS[1])
         return 0
     end
 """
 
 # Covers both cases when key doesn't exist and doesn't equal to lock's id
 EXTEND_SCRIPT = b"""
     if redis.call("get", KEYS[1]) ~= ARGV[1] then
         return 1
     elseif redis.call("ttl", KEYS[1]) < 0 then
         return 2
     else
         redis.call("expire", KEYS[1], ARGV[2])
         return 0
     end
 """
 
 RESET_SCRIPT = b"""
     redis.call('del', KEYS[2])
     redis.call('lpush', KEYS[2], 1)
     redis.call('pexpire', KEYS[2], ARGV[2])
     return redis.call('del', KEYS[1])
 """
 
 RESET_ALL_SCRIPT = b"""
     local locks = redis.call('keys', 'lock:*')
     local signal
     for _, lock in pairs(locks) do
         signal = 'lock-signal:' .. string.sub(lock, 6)
         redis.call('del', signal)
         redis.call('lpush', signal, 1)
         redis.call('expire', signal, 1)
         redis.call('del', lock)
     end
     return #locks
 """
 
 
 class AlreadyAcquired(RuntimeError):
     pass
 
 
 class NotAcquired(RuntimeError):
     pass
 
 
 class AlreadyStarted(RuntimeError):
     pass
 
 
 class TimeoutNotUsable(RuntimeError):
     pass
 
 
 class InvalidTimeout(RuntimeError):
     pass
 
 
 class TimeoutTooLarge(RuntimeError):
     pass
 
 
 class NotExpirable(RuntimeError):
     pass
 
 
 class Lock:
     """
     A Lock context manager implemented via redis SETNX/BLPOP.
     """
 
     unlock_script = None
     extend_script = None
     reset_script = None
     reset_all_script = None
+    blocking = None
 
     _lock_renewal_interval: float
     _lock_renewal_thread: Union[threading.Thread, None]
 
-    def __init__(self, redis_client, name, expire=None, id=None, auto_renewal=False, strict=True, signal_expire=1000):
+    def __init__(self, redis_client, name, expire=None, id=None, auto_renewal=False, strict=True, signal_expire=1000, blocking=True):
         """
         :param redis_client:
             An instance of :class:`~StrictRedis`.
         :param name:
             The name (redis key) the lock should have.
         :param expire:
             The lock expiry time in seconds. If left at the default (None)
             the lock will not expire.
         :param id:
             The ID (redis value) the lock should have. A random value is
             generated when left at the default.
 
             Note that if you specify this then the lock is marked as "held". Acquires
             won't be possible.
         :param auto_renewal:
             If set to ``True``, Lock will automatically renew the lock so that it
             doesn't expire for as long as the lock is held (acquire() called
             or running in a context manager).
 
             Implementation note: Renewal will happen using a daemon thread with
             an interval of ``expire*2/3``. If wishing to use a different renewal
             time, subclass Lock, call ``super().__init__()`` then set
             ``self._lock_renewal_interval`` to your desired interval.
         :param strict:
             If set ``True`` then the ``redis_client`` needs to be an instance of ``redis.StrictRedis``.
         :param signal_expire:
             Advanced option to override signal list expiration in milliseconds. Increase it for very slow clients. Default: ``1000``.
+        :param blocking:
+            Boolean value specifying whether lock should be blocking or not.
+            Used in `__enter__` method.
         """
         if strict and not isinstance(redis_client, StrictRedis):
             raise ValueError("redis_client must be instance of StrictRedis. "
                              "Use strict=False if you know what you're doing.")
         if auto_renewal and expire is None:
             raise ValueError("Expire may not be None when auto_renewal is set")
 
         self._client = redis_client
 
         if expire:
             expire = int(expire)
             if expire < 0:
                 raise ValueError("A negative expire is not acceptable.")
         else:
             expire = None
         self._expire = expire
 
         self._signal_expire = signal_expire
         if id is None:
             self._id = b64encode(urandom(18)).decode('ascii')
         elif isinstance(id, binary_type):
             try:
                 self._id = id.decode('ascii')
             except UnicodeDecodeError:
                 self._id = b64encode(id).decode('ascii')
         elif isinstance(id, text_type):
             self._id = id
         else:
             raise TypeError(f"Incorrect type for `id`. Must be bytes/str not {type(id)}.")
         self._name = 'lock:' + name
         self._signal = 'lock-signal:' + name
         self._lock_renewal_interval = (float(expire) * 2 / 3
                                        if auto_renewal
                                        else None)
         self._lock_renewal_thread = None
 
+        self.blocking = blocking
+
         self.register_scripts(redis_client)
 
     @classmethod
     def register_scripts(cls, redis_client):
         global reset_all_script
         if reset_all_script is None:
             cls.unlock_script = redis_client.register_script(UNLOCK_SCRIPT)
             cls.extend_script = redis_client.register_script(EXTEND_SCRIPT)
             cls.reset_script = redis_client.register_script(RESET_SCRIPT)
             cls.reset_all_script = redis_client.register_script(RESET_ALL_SCRIPT)
             reset_all_script = redis_client.register_script(RESET_ALL_SCRIPT)
 
     @property
     def _held(self):
         return self.id == self.get_owner_id()
 
     def reset(self):
         """
         Forcibly deletes the lock. Use this with care.
         """
         self.reset_script(client=self._client, keys=(self._name, self._signal), args=(self.id, self._signal_expire))
 
     @property
     def id(self):
         return self._id
 
     def get_owner_id(self):
         owner_id = self._client.get(self._name)
         if isinstance(owner_id, binary_type):
             owner_id = owner_id.decode('ascii', 'replace')
         return owner_id
 
     def acquire(self, blocking=True, timeout=None):
         """
         :param blocking:
             Boolean value specifying whether lock should be blocking or not.
         :param timeout:
             An integer value specifying the maximum number of seconds to block.
         """
         logger = loggers["acquire"]
 
         logger.debug("Getting blocking: %s acquire on %r ...", blocking, self._name)
 
         if self._held:
             owner_id = self.get_owner_id()
             raise AlreadyAcquired("Already acquired from this Lock instance. Lock id: {}".format(owner_id))
 
         if not blocking and timeout is not None:
             raise TimeoutNotUsable("Timeout cannot be used if blocking=False")
 
         if timeout:
             timeout = int(timeout)
             if timeout < 0:
                 raise InvalidTimeout(f"Timeout ({timeout}) cannot be less than or equal to 0")
 
             if self._expire and not self._lock_renewal_interval and timeout > self._expire:
                 raise TimeoutTooLarge(f"Timeout ({timeout}) cannot be greater than expire ({self._expire})")
 
         busy = True
         blpop_timeout = timeout or self._expire or 0
         timed_out = False
         while busy:
             busy = not self._client.set(self._name, self._id, nx=True, ex=self._expire)
             if busy:
                 if timed_out:
                     return False
                 elif blocking:
                     timed_out = not self._client.blpop(self._signal, blpop_timeout) and timeout
                 else:
                     logger.warning("Failed to acquire Lock(%r).", self._name)
                     return False
 
         logger.debug("Acquired Lock(%r).", self._name)
         if self._lock_renewal_interval is not None:
             self._start_lock_renewer()
         return True
 
     def extend(self, expire=None):
         """
         Extends expiration time of the lock.
 
         :param expire:
             New expiration time. If ``None`` - `expire` provided during
             lock initialization will be taken.
         """
         if expire:
             expire = int(expire)
             if expire < 0:
                 raise ValueError("A negative expire is not acceptable.")
         elif self._expire is not None:
             expire = self._expire
         else:
             raise TypeError(
                 "To extend a lock 'expire' must be provided as an "
                 "argument to extend() method or at initialization time."
             )
 
         error = self.extend_script(client=self._client, keys=(self._name, self._signal), args=(self._id, expire))
         if error == 1:
             raise NotAcquired(f"Lock {self._name} is not acquired or it already expired.")
         elif error == 2:
             raise NotExpirable(f"Lock {self._name} has no assigned expiration time")
         elif error:
             raise RuntimeError(f"Unsupported error code {error} from EXTEND script")
 
     @staticmethod
     def _lock_renewer(name, lockref, interval, stop):
         """
         Renew the lock key in redis every `interval` seconds for as long
         as `self._lock_renewal_thread.should_exit` is False.
         """
         while not stop.wait(timeout=interval):
             loggers["refresh.thread.start"].debug("Refreshing Lock(%r).", name)
             lock: "Lock" = lockref()
             if lock is None:
                 loggers["refresh.thread.stop"].debug(
                     "Stopping loop because Lock(%r) was garbage collected.", name
                 )
                 break
             lock.extend(expire=lock._expire)
             del lock
         loggers["refresh.thread.exit"].debug("Exiting renewal thread for Lock(%r).", name)
 
     def _start_lock_renewer(self):
         """
         Starts the lock refresher thread.
         """
         if self._lock_renewal_thread is not None:
             raise AlreadyStarted("Lock refresh thread already started")
 
         loggers["refresh.start"].debug(
             "Starting renewal thread for Lock(%r). Refresh interval: %s seconds.",
             self._name, self._lock_renewal_interval
         )
         self._lock_renewal_stop = threading.Event()
         self._lock_renewal_thread = threading.Thread(
             group=None,
             target=self._lock_renewer,
             kwargs={
                 'name': self._name,
                 'lockref': weakref.ref(self),
                 'interval': self._lock_renewal_interval,
                 'stop': self._lock_renewal_stop,
             },
         )
         self._lock_renewal_thread.daemon = True
         self._lock_renewal_thread.start()
 
     def _stop_lock_renewer(self):
         """
         Stop the lock renewer.
 
         This signals the renewal thread and waits for its exit.
         """
         if self._lock_renewal_thread is None or not self._lock_renewal_thread.is_alive():
             return
         loggers["refresh.shutdown"].debug("Signaling renewal thread for Lock(%r) to exit.", self._name)
         self._lock_renewal_stop.set()
         self._lock_renewal_thread.join()
         self._lock_renewal_thread = None
         loggers["refresh.exit"].debug("Renewal thread for Lock(%r) exited.", self._name)
 
     def __enter__(self):
-        acquired = self.acquire(blocking=True)
+        acquired = self.acquire(blocking=self.blocking)
         if not acquired:
-            raise AssertionError(f"Lock({self._name}) wasn't acquired, but blocking=True was used!")
+            if self.blocking:
+                raise AssertionError(f"Lock({self._name}) wasn't acquired, but blocking=True was used!")
+            raise NotAcquired(f"Lock({self._name}) is not acquired or it already expired.")
         return self
 
     def __exit__(self, exc_type=None, exc_value=None, traceback=None):
         self.release()
 
     def release(self):
         """Releases the lock, that was acquired with the same object.
 
         .. note::
 
             If you want to release a lock that you acquired in a different place you have two choices:
 
             * Use ``Lock("name", id=id_from_other_place).release()``
             * Use ``Lock("name").reset()``
         """
         if self._lock_renewal_thread is not None:
             self._stop_lock_renewer()
         loggers["release"].debug("Releasing Lock(%r).", self._name)
         error = self.unlock_script(client=self._client, keys=(self._name, self._signal), args=(self._id, self._signal_expire))
         if error == 1:
             raise NotAcquired(f"Lock({self._name}) is not acquired or it already expired.")
         elif error:
             raise RuntimeError(f"Unsupported error code {error} from EXTEND script.")
 
     def locked(self):
         """
         Return true if the lock is acquired.
 
         Checks that lock with same name already exists. This method returns true, even if
         lock have another id.
         """
         return self._client.exists(self._name) == 1
 
 
 reset_all_script = None
 
 
 def reset_all(redis_client):
     """
     Forcibly deletes all locks if its remains (like a crash reason). Use this with care.
 
     :param redis_client:
         An instance of :class:`~StrictRedis`.
     """
     Lock.register_scripts(redis_client)
 
     reset_all_script(client=redis_client)  # noqa

vcsserver/lib/_vendor/statsd/__init__.py

@@ -1,50 +1,51 b''
 import logging
 
 from .stream import TCPStatsClient, UnixSocketStatsClient  # noqa
 from .udp import StatsClient  # noqa
 
 HOST = 'localhost'
 PORT = 8125
 IPV6 = False
 PREFIX = None
 MAXUDPSIZE = 512
 
 log = logging.getLogger('rhodecode.statsd')
 
 
 def statsd_config(config, prefix='statsd.'):
     _config = {}
-    for key in config.keys():
+    for key in list(config.keys()):
         if key.startswith(prefix):
             _config[key[len(prefix):]] = config[key]
     return _config
 
 
 def client_from_config(configuration, prefix='statsd.', **kwargs):
     from pyramid.settings import asbool
 
     _config = statsd_config(configuration, prefix)
+    statsd_flag = _config.get('enabled')
     statsd_enabled = asbool(_config.pop('enabled', False))
     if not statsd_enabled:
-        log.debug('statsd client not enabled by statsd.enabled =  flag, skipping...')
+        log.debug('statsd client not enabled by statsd.enabled = %s flag, skipping...', statsd_flag)
         return
 
     host = _config.pop('statsd_host', HOST)
     port = _config.pop('statsd_port', PORT)
     prefix = _config.pop('statsd_prefix', PREFIX)
     maxudpsize = _config.pop('statsd_maxudpsize', MAXUDPSIZE)
     ipv6 = asbool(_config.pop('statsd_ipv6', IPV6))
     log.debug('configured statsd client %s:%s', host, port)
 
     try:
         client = StatsClient(
             host=host, port=port, prefix=prefix, maxudpsize=maxudpsize, ipv6=ipv6)
     except Exception:
         log.exception('StatsD is enabled, but failed to connect to statsd server, fallback: disable statsd')
         client = None
 
     return client
 
 
 def get_statsd_client(request):
     return client_from_config(request.registry.settings)

vcsserver/lib/ext_json.py

@@ -1,2 +1,14 b''
-# use orjson by default
-import orjson as json
+import json as stdlib_json
+
+try:
+    # we keep simplejson for having dump functionality still
+    # orjson doesn't support it
+    import simplejson as sjson
+except ImportError:
+    sjson = stdlib_json
+
+try:
+    import orjson
+    import orjson as json
+except ImportError:
+    json = stdlib_json

vcsserver/lib/logging_formatter.py

@@ -1,53 +1,63 b''
 # Copyright (C) 2010-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import sys
 import logging
 
 
 BLACK, RED, GREEN, YELLOW, BLUE, MAGENTA, CYAN, WHITE = list(range(30, 38))
 
 # Sequences
 RESET_SEQ = "\033[0m"
 COLOR_SEQ = "\033[0;%dm"
 BOLD_SEQ = "\033[1m"
 
 COLORS = {
     'CRITICAL': MAGENTA,
     'ERROR': RED,
     'WARNING': CYAN,
     'INFO': GREEN,
     'DEBUG': BLUE,
     'SQL': YELLOW
 }
 
 
+def _inject_req_id(record, *args, **kwargs):
+    return record
+
+
+class ExceptionAwareFormatter(logging.Formatter):
+    pass
+
+
 class ColorFormatter(logging.Formatter):
 
     def format(self, record):
         """
-        Change record's levelname to use with COLORS enum
+        Changes record's levelname to use with COLORS enum
         """
         def_record = super().format(record)
 
         levelname = record.levelname
         start = COLOR_SEQ % (COLORS[levelname])
         end = RESET_SEQ
 
         colored_record = ''.join([start, def_record, end])
         return colored_record
+
+