We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[CT-2063] [Regression] Console Logging in dbt 1.4.1 is not interactive
### Is this a regression in a recent version of dbt-core?

- [X] I believe this is a regression in dbt-core functionality
- [X] I have searched the existing issues, and I could not find an existing issue for this regression

### Current Behavior

We recently upgraded to dbt 1.4.1 and noticed the console logging (both locally and when we run commands via Airflow) is not interactive/real-time anymore.

Notice the dbt command started running at 8:01 but didn't log until 9:27 when it was fully complete.

```
2023-02-08, 08:01:12 UTC] {dbt_hook.py:137} INFO - dbt run --project-dir /tmp/dbt --profiles-dir /tmp/dbt/profile --target prod --models analytics_tables
[2023-02-08, 08:01:12 UTC] {dbt_hook.py:150} INFO - Output:
[2023-02-08, 09:27:22 UTC] {dbt_hook.py:154} INFO - [0m08:01:15  Running with dbt=1.4.1
[2023-02-08, 09:27:22 UTC] {dbt_hook.py:154} INFO - [0m08:03:56  Found 100 models, 100 tests, 26 snapshots, 0 analyses, 100 macros, 4 operations, 90 seed files, 100 sources, 0 exposures, 0 metrics
[2023-02-08, 09:27:22 UTC] {dbt_hook.py:154} INFO - [0m08:07:14  Running 1 on-run-start hook
[2023-02-08, 09:27:22 UTC] {dbt_hook.py:154} INFO - [0m08:07:14  1 of 1 START hook: elementary.on-run-start.0 ................................... [RUN]
[2023-02-08, 09:27:22 UTC] {dbt_hook.py:154} INFO - [0m08:07:14  1 of 1 OK hook: elementary.on-run-start.0 ...................................... [[32mOK[0m in 0.00s]
[2023-02-08, 09:27:22 UTC] {dbt_hook.py:154} INFO - [0m08:07:14  Concurrency: 16 threads (target='prod')
[2023-02-08, 09:27:22 UTC] {dbt_hook.py:154} INFO - [0m08:07:14  1 of 27 START sql table model transform.table [RUN]
```

### Expected/Previous Behavior

This is how it logged previously in real-time

```
[2023-02-07, 08:02:32 UTC] {dbt_hook.py:137} INFO - dbt run --project-dir /tmp/dbt --profiles-dir /tmp/dbt/profile --target prod --models analytics_tables
[2023-02-07, 08:02:32 UTC] {dbt_hook.py:150} INFO - Output:
[2023-02-07, 08:02:35 UTC] {dbt_hook.py:154} INFO - [0m08:02:35  Running with dbt=1.3.1
[2023-02-07, 08:05:27 UTC] {dbt_hook.py:154} INFO - [0m08:05:27  Found 100 models, 100 tests, 26 snapshots, 0 analyses, 100 macros, 4 operations, 90 seed files, 100 sources, 0 exposures, 0 metrics
[2023-02-07, 08:09:23 UTC] {dbt_hook.py:154} INFO - [0m08:09:23  Running 1 on-run-start hook
[2023-02-07, 08:09:23 UTC] {dbt_hook.py:154} INFO - [0m08:09:23  1 of 1 START hook: elementary.on-run-start.0 ................................... [RUN]
[2023-02-07, 08:09:23 UTC] {dbt_hook.py:154} INFO - [0m08:09:23  1 of 1 OK hook: elementary.on-run-start.0 ...................................... [[32mOK[0m in 0.00s]
[2023-02-07, 08:09:23 UTC] {dbt_hook.py:154} INFO - [0m08:09:23  Concurrency: 16 threads (target='prod')
[2023-02-07, 08:09:23 UTC] {dbt_hook.py:154} INFO - [0m08:09:23  1 of 27 START sql table model transform.table .... [RUN]

### Steps To Reproduce

All we did was upgrade `dbt-core` and `dbt-snowflake` to the latest versions

### Relevant log output

_No response_

### Environment

```markdown
- OS:
- Python:
- dbt (working version): 1.3.1
- dbt (regression version): 1.4.1
```


### Which database adapter are you using with dbt?

snowflake

### Additional Context

_No response_

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `core/dbt/events/eventmgr.py`
Content:
```
1 from colorama import Style
2 from dataclasses import dataclass
3 from datetime import datetime
4 from enum import Enum
5 import json
6 import logging
7 from logging.handlers import RotatingFileHandler
8 import threading
9 from typing import Any, Callable, List, Optional, TextIO
10 from uuid import uuid4
11 
12 from dbt.events.base_types import BaseEvent, EventLevel, msg_from_base_event, EventMsg
13 
14 
15 # A Filter is a function which takes a BaseEvent and returns True if the event
16 # should be logged, False otherwise.
17 Filter = Callable[[EventMsg], bool]
18 
19 
20 # Default filter which logs every event
21 def NoFilter(_: EventMsg) -> bool:
22     return True
23 
24 
25 # A Scrubber removes secrets from an input string, returning a sanitized string.
26 Scrubber = Callable[[str], str]
27 
28 
29 # Provide a pass-through scrubber implementation, also used as a default
30 def NoScrubber(s: str) -> str:
31     return s
32 
33 
34 class LineFormat(Enum):
35     PlainText = 1
36     DebugText = 2
37     Json = 3
38 
39 
40 # Map from dbt event levels to python log levels
41 _log_level_map = {
42     EventLevel.DEBUG: 10,
43     EventLevel.TEST: 10,
44     EventLevel.INFO: 20,
45     EventLevel.WARN: 30,
46     EventLevel.ERROR: 40,
47 }
48 
49 
50 # We need this function for now because the numeric log severity levels in
51 # Python do not match those for logbook, so we have to explicitly call the
52 # correct function by name.
53 def send_to_logger(l, level: str, log_line: str):
54     if level == "test":
55         l.debug(log_line)
56     elif level == "debug":
57         l.debug(log_line)
58     elif level == "info":
59         l.info(log_line)
60     elif level == "warn":
61         l.warning(log_line)
62     elif level == "error":
63         l.error(log_line)
64     else:
65         raise AssertionError(
66             f"While attempting to log {log_line}, encountered the unhandled level: {level}"
67         )
68 
69 
70 @dataclass
71 class LoggerConfig:
72     name: str
73     filter: Filter = NoFilter
74     scrubber: Scrubber = NoScrubber
75     line_format: LineFormat = LineFormat.PlainText
76     level: EventLevel = EventLevel.WARN
77     use_colors: bool = False
78     output_stream: Optional[TextIO] = None
79     output_file_name: Optional[str] = None
80     logger: Optional[Any] = None
81 
82 
83 class _Logger:
84     def __init__(self, event_manager: "EventManager", config: LoggerConfig) -> None:
85         self.name: str = config.name
86         self.filter: Filter = config.filter
87         self.scrubber: Scrubber = config.scrubber
88         self.level: EventLevel = config.level
89         self.event_manager: EventManager = event_manager
90         self._python_logger: Optional[logging.Logger] = config.logger
91         self._stream: Optional[TextIO] = config.output_stream
92 
93         if config.output_file_name:
94             log = logging.getLogger(config.name)
95             log.setLevel(_log_level_map[config.level])
96             handler = RotatingFileHandler(
97                 filename=str(config.output_file_name),
98                 encoding="utf8",
99                 maxBytes=10 * 1024 * 1024,  # 10 mb
100                 backupCount=5,
101             )
102 
103             handler.setFormatter(logging.Formatter(fmt="%(message)s"))
104             log.handlers.clear()
105             log.addHandler(handler)
106 
107             self._python_logger = log
108 
109     def create_line(self, msg: EventMsg) -> str:
110         raise NotImplementedError()
111 
112     def write_line(self, msg: EventMsg):
113         line = self.create_line(msg)
114         python_level = _log_level_map[EventLevel(msg.info.level)]
115         if self._python_logger is not None:
116             send_to_logger(self._python_logger, msg.info.level, line)
117         elif self._stream is not None and _log_level_map[self.level] <= python_level:
118             self._stream.write(line + "\n")
119 
120     def flush(self):
121         if self._python_logger is not None:
122             for handler in self._python_logger.handlers:
123                 handler.flush()
124         elif self._stream is not None:
125             self._stream.flush()
126 
127 
128 class _TextLogger(_Logger):
129     def __init__(self, event_manager: "EventManager", config: LoggerConfig) -> None:
130         super().__init__(event_manager, config)
131         self.use_colors = config.use_colors
132         self.use_debug_format = config.line_format == LineFormat.DebugText
133 
134     def create_line(self, msg: EventMsg) -> str:
135         return self.create_debug_line(msg) if self.use_debug_format else self.create_info_line(msg)
136 
137     def create_info_line(self, msg: EventMsg) -> str:
138         ts: str = datetime.utcnow().strftime("%H:%M:%S")
139         scrubbed_msg: str = self.scrubber(msg.info.msg)  # type: ignore
140         return f"{self._get_color_tag()}{ts}  {scrubbed_msg}"
141 
142     def create_debug_line(self, msg: EventMsg) -> str:
143         log_line: str = ""
144         # Create a separator if this is the beginning of an invocation
145         # TODO: This is an ugly hack, get rid of it if we can
146         if msg.info.name == "MainReportVersion":
147             separator = 30 * "="
148             log_line = (
149                 f"\n\n{separator} {msg.info.ts} | {self.event_manager.invocation_id} {separator}\n"
150             )
151         ts: str = msg.info.ts.strftime("%H:%M:%S.%f")
152         scrubbed_msg: str = self.scrubber(msg.info.msg)  # type: ignore
153         level = msg.info.level
154         log_line += (
155             f"{self._get_color_tag()}{ts} [{level:<5}]{self._get_thread_name()} {scrubbed_msg}"
156         )
157         return log_line
158 
159     def _get_color_tag(self) -> str:
160         return "" if not self.use_colors else Style.RESET_ALL
161 
162     def _get_thread_name(self) -> str:
163         thread_name = ""
164         if threading.current_thread().name:
165             thread_name = threading.current_thread().name
166             thread_name = thread_name[:10]
167             thread_name = thread_name.ljust(10, " ")
168             thread_name = f" [{thread_name}]:"
169         return thread_name
170 
171 
172 class _JsonLogger(_Logger):
173     def create_line(self, msg: EventMsg) -> str:
174         from dbt.events.functions import msg_to_dict
175 
176         msg_dict = msg_to_dict(msg)
177         raw_log_line = json.dumps(msg_dict, sort_keys=True)
178         line = self.scrubber(raw_log_line)  # type: ignore
179         return line
180 
181 
182 class EventManager:
183     def __init__(self) -> None:
184         self.loggers: List[_Logger] = []
185         self.callbacks: List[Callable[[EventMsg], None]] = []
186         self.invocation_id: str = str(uuid4())
187 
188     def fire_event(self, e: BaseEvent, level: EventLevel = None) -> None:
189         msg = msg_from_base_event(e, level=level)
190         for logger in self.loggers:
191             if logger.filter(msg):  # type: ignore
192                 logger.write_line(msg)
193 
194         for callback in self.callbacks:
195             callback(msg)
196 
197     def add_logger(self, config: LoggerConfig):
198         logger = (
199             _JsonLogger(self, config)
200             if config.line_format == LineFormat.Json
201             else _TextLogger(self, config)
202         )
203         logger.event_manager = self
204         self.loggers.append(logger)
205 
206     def flush(self):
207         for logger in self.loggers:
208             logger.flush()
209
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/core/dbt/events/eventmgr.py b/core/dbt/events/eventmgr.py
--- a/core/dbt/events/eventmgr.py
+++ b/core/dbt/events/eventmgr.py
@@ -88,41 +88,40 @@
         self.level: EventLevel = config.level
         self.event_manager: EventManager = event_manager
         self._python_logger: Optional[logging.Logger] = config.logger
-        self._stream: Optional[TextIO] = config.output_stream
+
+        if config.output_stream is not None:
+            stream_handler = logging.StreamHandler(config.output_stream)
+            self._python_logger = self._get_python_log_for_handler(stream_handler)
 
         if config.output_file_name:
-            log = logging.getLogger(config.name)
-            log.setLevel(_log_level_map[config.level])
-            handler = RotatingFileHandler(
+            file_handler = RotatingFileHandler(
                 filename=str(config.output_file_name),
                 encoding="utf8",
                 maxBytes=10 * 1024 * 1024,  # 10 mb
                 backupCount=5,
             )
+            self._python_logger = self._get_python_log_for_handler(file_handler)
 
-            handler.setFormatter(logging.Formatter(fmt="%(message)s"))
-            log.handlers.clear()
-            log.addHandler(handler)
-
-            self._python_logger = log
+    def _get_python_log_for_handler(self, handler: logging.Handler):
+        log = logging.getLogger(self.name)
+        log.setLevel(_log_level_map[self.level])
+        handler.setFormatter(logging.Formatter(fmt="%(message)s"))
+        log.handlers.clear()
+        log.addHandler(handler)
+        return log
 
     def create_line(self, msg: EventMsg) -> str:
         raise NotImplementedError()
 
     def write_line(self, msg: EventMsg):
         line = self.create_line(msg)
-        python_level = _log_level_map[EventLevel(msg.info.level)]
         if self._python_logger is not None:
             send_to_logger(self._python_logger, msg.info.level, line)
-        elif self._stream is not None and _log_level_map[self.level] <= python_level:
-            self._stream.write(line + "\n")
 
     def flush(self):
         if self._python_logger is not None:
             for handler in self._python_logger.handlers:
                 handler.flush()
-        elif self._stream is not None:
-            self._stream.flush()
 
 
 class _TextLogger(_Logger):

{"golden_diff": "diff --git a/core/dbt/events/eventmgr.py b/core/dbt/events/eventmgr.py\n--- a/core/dbt/events/eventmgr.py\n+++ b/core/dbt/events/eventmgr.py\n@@ -88,41 +88,40 @@\n         self.level: EventLevel = config.level\n         self.event_manager: EventManager = event_manager\n         self._python_logger: Optional[logging.Logger] = config.logger\n-        self._stream: Optional[TextIO] = config.output_stream\n+\n+        if config.output_stream is not None:\n+            stream_handler = logging.StreamHandler(config.output_stream)\n+            self._python_logger = self._get_python_log_for_handler(stream_handler)\n \n         if config.output_file_name:\n-            log = logging.getLogger(config.name)\n-            log.setLevel(_log_level_map[config.level])\n-            handler = RotatingFileHandler(\n+            file_handler = RotatingFileHandler(\n                 filename=str(config.output_file_name),\n                 encoding=\"utf8\",\n                 maxBytes=10 * 1024 * 1024,  # 10 mb\n                 backupCount=5,\n             )\n+            self._python_logger = self._get_python_log_for_handler(file_handler)\n \n-            handler.setFormatter(logging.Formatter(fmt=\"%(message)s\"))\n-            log.handlers.clear()\n-            log.addHandler(handler)\n-\n-            self._python_logger = log\n+    def _get_python_log_for_handler(self, handler: logging.Handler):\n+        log = logging.getLogger(self.name)\n+        log.setLevel(_log_level_map[self.level])\n+        handler.setFormatter(logging.Formatter(fmt=\"%(message)s\"))\n+        log.handlers.clear()\n+        log.addHandler(handler)\n+        return log\n \n     def create_line(self, msg: EventMsg) -> str:\n         raise NotImplementedError()\n \n     def write_line(self, msg: EventMsg):\n         line = self.create_line(msg)\n-        python_level = _log_level_map[EventLevel(msg.info.level)]\n         if self._python_logger is not None:\n             send_to_logger(self._python_logger, msg.info.level, line)\n-        elif self._stream is not None and _log_level_map[self.level] <= python_level:\n-            self._stream.write(line + \"\\n\")\n \n     def flush(self):\n         if self._python_logger is not None:\n             for handler in self._python_logger.handlers:\n                 handler.flush()\n-        elif self._stream is not None:\n-            self._stream.flush()\n \n \n class _TextLogger(_Logger):\n", "issue": "[CT-2063] [Regression] Console Logging in dbt 1.4.1 is not interactive\n### Is this a regression in a recent version of dbt-core?\n\n- [X] I believe this is a regression in dbt-core functionality\n- [X] I have searched the existing issues, and I could not find an existing issue for this regression\n\n### Current Behavior\n\nWe recently upgraded to dbt 1.4.1 and noticed the console logging (both locally and when we run commands via Airflow) is not interactive/real-time anymore.\r\n\r\nNotice the dbt command started running at 8:01 but didn't log until 9:27 when it was fully complete.\r\n\r\n```\r\n2023-02-08, 08:01:12 UTC] {dbt_hook.py:137} INFO - dbt run --project-dir /tmp/dbt --profiles-dir /tmp/dbt/profile --target prod --models analytics_tables\r\n[2023-02-08, 08:01:12 UTC] {dbt_hook.py:150} INFO - Output:\r\n[2023-02-08, 09:27:22 UTC] {dbt_hook.py:154} INFO - [0m08:01:15  Running with dbt=1.4.1\r\n[2023-02-08, 09:27:22 UTC] {dbt_hook.py:154} INFO - [0m08:03:56  Found 100 models, 100 tests, 26 snapshots, 0 analyses, 100 macros, 4 operations, 90 seed files, 100 sources, 0 exposures, 0 metrics\r\n[2023-02-08, 09:27:22 UTC] {dbt_hook.py:154} INFO - [0m08:07:14  Running 1 on-run-start hook\r\n[2023-02-08, 09:27:22 UTC] {dbt_hook.py:154} INFO - [0m08:07:14  1 of 1 START hook: elementary.on-run-start.0 ................................... [RUN]\r\n[2023-02-08, 09:27:22 UTC] {dbt_hook.py:154} INFO - [0m08:07:14  1 of 1 OK hook: elementary.on-run-start.0 ...................................... [[32mOK[0m in 0.00s]\r\n[2023-02-08, 09:27:22 UTC] {dbt_hook.py:154} INFO - [0m08:07:14  Concurrency: 16 threads (target='prod')\r\n[2023-02-08, 09:27:22 UTC] {dbt_hook.py:154} INFO - [0m08:07:14  1 of 27 START sql table model transform.table [RUN]\r\n```\n\n### Expected/Previous Behavior\n\nThis is how it logged previously in real-time\r\n\r\n```\r\n[2023-02-07, 08:02:32 UTC] {dbt_hook.py:137} INFO - dbt run --project-dir /tmp/dbt --profiles-dir /tmp/dbt/profile --target prod --models analytics_tables\r\n[2023-02-07, 08:02:32 UTC] {dbt_hook.py:150} INFO - Output:\r\n[2023-02-07, 08:02:35 UTC] {dbt_hook.py:154} INFO - \u001b[0m08:02:35  Running with dbt=1.3.1\r\n[2023-02-07, 08:05:27 UTC] {dbt_hook.py:154} INFO - \u001b[0m08:05:27  Found 100 models, 100 tests, 26 snapshots, 0 analyses, 100 macros, 4 operations, 90 seed files, 100 sources, 0 exposures, 0 metrics\r\n[2023-02-07, 08:09:23 UTC] {dbt_hook.py:154} INFO - \u001b[0m08:09:23  Running 1 on-run-start hook\r\n[2023-02-07, 08:09:23 UTC] {dbt_hook.py:154} INFO - \u001b[0m08:09:23  1 of 1 START hook: elementary.on-run-start.0 ................................... [RUN]\r\n[2023-02-07, 08:09:23 UTC] {dbt_hook.py:154} INFO - \u001b[0m08:09:23  1 of 1 OK hook: elementary.on-run-start.0 ...................................... [\u001b[32mOK\u001b[0m in 0.00s]\r\n[2023-02-07, 08:09:23 UTC] {dbt_hook.py:154} INFO - \u001b[0m08:09:23  Concurrency: 16 threads (target='prod')\r\n[2023-02-07, 08:09:23 UTC] {dbt_hook.py:154} INFO - \u001b[0m08:09:23  1 of 27 START sql table model transform.table .... [RUN]\n\n### Steps To Reproduce\n\nAll we did was upgrade `dbt-core` and `dbt-snowflake` to the latest versions\n\n### Relevant log output\n\n_No response_\n\n### Environment\n\n```markdown\n- OS:\r\n- Python:\r\n- dbt (working version): 1.3.1\r\n- dbt (regression version): 1.4.1\n```\n\n\n### Which database adapter are you using with dbt?\n\nsnowflake\n\n### Additional Context\n\n_No response_\n", "before_files": [{"content": "from colorama import Style\nfrom dataclasses import dataclass\nfrom datetime import datetime\nfrom enum import Enum\nimport json\nimport logging\nfrom logging.handlers import RotatingFileHandler\nimport threading\nfrom typing import Any, Callable, List, Optional, TextIO\nfrom uuid import uuid4\n\nfrom dbt.events.base_types import BaseEvent, EventLevel, msg_from_base_event, EventMsg\n\n\n# A Filter is a function which takes a BaseEvent and returns True if the event\n# should be logged, False otherwise.\nFilter = Callable[[EventMsg], bool]\n\n\n# Default filter which logs every event\ndef NoFilter(_: EventMsg) -> bool:\n    return True\n\n\n# A Scrubber removes secrets from an input string, returning a sanitized string.\nScrubber = Callable[[str], str]\n\n\n# Provide a pass-through scrubber implementation, also used as a default\ndef NoScrubber(s: str) -> str:\n    return s\n\n\nclass LineFormat(Enum):\n    PlainText = 1\n    DebugText = 2\n    Json = 3\n\n\n# Map from dbt event levels to python log levels\n_log_level_map = {\n    EventLevel.DEBUG: 10,\n    EventLevel.TEST: 10,\n    EventLevel.INFO: 20,\n    EventLevel.WARN: 30,\n    EventLevel.ERROR: 40,\n}\n\n\n# We need this function for now because the numeric log severity levels in\n# Python do not match those for logbook, so we have to explicitly call the\n# correct function by name.\ndef send_to_logger(l, level: str, log_line: str):\n    if level == \"test\":\n        l.debug(log_line)\n    elif level == \"debug\":\n        l.debug(log_line)\n    elif level == \"info\":\n        l.info(log_line)\n    elif level == \"warn\":\n        l.warning(log_line)\n    elif level == \"error\":\n        l.error(log_line)\n    else:\n        raise AssertionError(\n            f\"While attempting to log {log_line}, encountered the unhandled level: {level}\"\n        )\n\n\n@dataclass\nclass LoggerConfig:\n    name: str\n    filter: Filter = NoFilter\n    scrubber: Scrubber = NoScrubber\n    line_format: LineFormat = LineFormat.PlainText\n    level: EventLevel = EventLevel.WARN\n    use_colors: bool = False\n    output_stream: Optional[TextIO] = None\n    output_file_name: Optional[str] = None\n    logger: Optional[Any] = None\n\n\nclass _Logger:\n    def __init__(self, event_manager: \"EventManager\", config: LoggerConfig) -> None:\n        self.name: str = config.name\n        self.filter: Filter = config.filter\n        self.scrubber: Scrubber = config.scrubber\n        self.level: EventLevel = config.level\n        self.event_manager: EventManager = event_manager\n        self._python_logger: Optional[logging.Logger] = config.logger\n        self._stream: Optional[TextIO] = config.output_stream\n\n        if config.output_file_name:\n            log = logging.getLogger(config.name)\n            log.setLevel(_log_level_map[config.level])\n            handler = RotatingFileHandler(\n                filename=str(config.output_file_name),\n                encoding=\"utf8\",\n                maxBytes=10 * 1024 * 1024,  # 10 mb\n                backupCount=5,\n            )\n\n            handler.setFormatter(logging.Formatter(fmt=\"%(message)s\"))\n            log.handlers.clear()\n            log.addHandler(handler)\n\n            self._python_logger = log\n\n    def create_line(self, msg: EventMsg) -> str:\n        raise NotImplementedError()\n\n    def write_line(self, msg: EventMsg):\n        line = self.create_line(msg)\n        python_level = _log_level_map[EventLevel(msg.info.level)]\n        if self._python_logger is not None:\n            send_to_logger(self._python_logger, msg.info.level, line)\n        elif self._stream is not None and _log_level_map[self.level] <= python_level:\n            self._stream.write(line + \"\\n\")\n\n    def flush(self):\n        if self._python_logger is not None:\n            for handler in self._python_logger.handlers:\n                handler.flush()\n        elif self._stream is not None:\n            self._stream.flush()\n\n\nclass _TextLogger(_Logger):\n    def __init__(self, event_manager: \"EventManager\", config: LoggerConfig) -> None:\n        super().__init__(event_manager, config)\n        self.use_colors = config.use_colors\n        self.use_debug_format = config.line_format == LineFormat.DebugText\n\n    def create_line(self, msg: EventMsg) -> str:\n        return self.create_debug_line(msg) if self.use_debug_format else self.create_info_line(msg)\n\n    def create_info_line(self, msg: EventMsg) -> str:\n        ts: str = datetime.utcnow().strftime(\"%H:%M:%S\")\n        scrubbed_msg: str = self.scrubber(msg.info.msg)  # type: ignore\n        return f\"{self._get_color_tag()}{ts}  {scrubbed_msg}\"\n\n    def create_debug_line(self, msg: EventMsg) -> str:\n        log_line: str = \"\"\n        # Create a separator if this is the beginning of an invocation\n        # TODO: This is an ugly hack, get rid of it if we can\n        if msg.info.name == \"MainReportVersion\":\n            separator = 30 * \"=\"\n            log_line = (\n                f\"\\n\\n{separator} {msg.info.ts} | {self.event_manager.invocation_id} {separator}\\n\"\n            )\n        ts: str = msg.info.ts.strftime(\"%H:%M:%S.%f\")\n        scrubbed_msg: str = self.scrubber(msg.info.msg)  # type: ignore\n        level = msg.info.level\n        log_line += (\n            f\"{self._get_color_tag()}{ts} [{level:<5}]{self._get_thread_name()} {scrubbed_msg}\"\n        )\n        return log_line\n\n    def _get_color_tag(self) -> str:\n        return \"\" if not self.use_colors else Style.RESET_ALL\n\n    def _get_thread_name(self) -> str:\n        thread_name = \"\"\n        if threading.current_thread().name:\n            thread_name = threading.current_thread().name\n            thread_name = thread_name[:10]\n            thread_name = thread_name.ljust(10, \" \")\n            thread_name = f\" [{thread_name}]:\"\n        return thread_name\n\n\nclass _JsonLogger(_Logger):\n    def create_line(self, msg: EventMsg) -> str:\n        from dbt.events.functions import msg_to_dict\n\n        msg_dict = msg_to_dict(msg)\n        raw_log_line = json.dumps(msg_dict, sort_keys=True)\n        line = self.scrubber(raw_log_line)  # type: ignore\n        return line\n\n\nclass EventManager:\n    def __init__(self) -> None:\n        self.loggers: List[_Logger] = []\n        self.callbacks: List[Callable[[EventMsg], None]] = []\n        self.invocation_id: str = str(uuid4())\n\n    def fire_event(self, e: BaseEvent, level: EventLevel = None) -> None:\n        msg = msg_from_base_event(e, level=level)\n        for logger in self.loggers:\n            if logger.filter(msg):  # type: ignore\n                logger.write_line(msg)\n\n        for callback in self.callbacks:\n            callback(msg)\n\n    def add_logger(self, config: LoggerConfig):\n        logger = (\n            _JsonLogger(self, config)\n            if config.line_format == LineFormat.Json\n            else _TextLogger(self, config)\n        )\n        logger.event_manager = self\n        self.loggers.append(logger)\n\n    def flush(self):\n        for logger in self.loggers:\n            logger.flush()\n", "path": "core/dbt/events/eventmgr.py"}], "after_files": [{"content": "from colorama import Style\nfrom dataclasses import dataclass\nfrom datetime import datetime\nfrom enum import Enum\nimport json\nimport logging\nfrom logging.handlers import RotatingFileHandler\nimport threading\nfrom typing import Any, Callable, List, Optional, TextIO\nfrom uuid import uuid4\n\nfrom dbt.events.base_types import BaseEvent, EventLevel, msg_from_base_event, EventMsg\n\n\n# A Filter is a function which takes a BaseEvent and returns True if the event\n# should be logged, False otherwise.\nFilter = Callable[[EventMsg], bool]\n\n\n# Default filter which logs every event\ndef NoFilter(_: EventMsg) -> bool:\n    return True\n\n\n# A Scrubber removes secrets from an input string, returning a sanitized string.\nScrubber = Callable[[str], str]\n\n\n# Provide a pass-through scrubber implementation, also used as a default\ndef NoScrubber(s: str) -> str:\n    return s\n\n\nclass LineFormat(Enum):\n    PlainText = 1\n    DebugText = 2\n    Json = 3\n\n\n# Map from dbt event levels to python log levels\n_log_level_map = {\n    EventLevel.DEBUG: 10,\n    EventLevel.TEST: 10,\n    EventLevel.INFO: 20,\n    EventLevel.WARN: 30,\n    EventLevel.ERROR: 40,\n}\n\n\n# We need this function for now because the numeric log severity levels in\n# Python do not match those for logbook, so we have to explicitly call the\n# correct function by name.\ndef send_to_logger(l, level: str, log_line: str):\n    if level == \"test\":\n        l.debug(log_line)\n    elif level == \"debug\":\n        l.debug(log_line)\n    elif level == \"info\":\n        l.info(log_line)\n    elif level == \"warn\":\n        l.warning(log_line)\n    elif level == \"error\":\n        l.error(log_line)\n    else:\n        raise AssertionError(\n            f\"While attempting to log {log_line}, encountered the unhandled level: {level}\"\n        )\n\n\n@dataclass\nclass LoggerConfig:\n    name: str\n    filter: Filter = NoFilter\n    scrubber: Scrubber = NoScrubber\n    line_format: LineFormat = LineFormat.PlainText\n    level: EventLevel = EventLevel.WARN\n    use_colors: bool = False\n    output_stream: Optional[TextIO] = None\n    output_file_name: Optional[str] = None\n    logger: Optional[Any] = None\n\n\nclass _Logger:\n    def __init__(self, event_manager: \"EventManager\", config: LoggerConfig) -> None:\n        self.name: str = config.name\n        self.filter: Filter = config.filter\n        self.scrubber: Scrubber = config.scrubber\n        self.level: EventLevel = config.level\n        self.event_manager: EventManager = event_manager\n        self._python_logger: Optional[logging.Logger] = config.logger\n\n        if config.output_stream is not None:\n            stream_handler = logging.StreamHandler(config.output_stream)\n            self._python_logger = self._get_python_log_for_handler(stream_handler)\n\n        if config.output_file_name:\n            file_handler = RotatingFileHandler(\n                filename=str(config.output_file_name),\n                encoding=\"utf8\",\n                maxBytes=10 * 1024 * 1024,  # 10 mb\n                backupCount=5,\n            )\n            self._python_logger = self._get_python_log_for_handler(file_handler)\n\n    def _get_python_log_for_handler(self, handler: logging.Handler):\n        log = logging.getLogger(self.name)\n        log.setLevel(_log_level_map[self.level])\n        handler.setFormatter(logging.Formatter(fmt=\"%(message)s\"))\n        log.handlers.clear()\n        log.addHandler(handler)\n        return log\n\n    def create_line(self, msg: EventMsg) -> str:\n        raise NotImplementedError()\n\n    def write_line(self, msg: EventMsg):\n        line = self.create_line(msg)\n        if self._python_logger is not None:\n            send_to_logger(self._python_logger, msg.info.level, line)\n\n    def flush(self):\n        if self._python_logger is not None:\n            for handler in self._python_logger.handlers:\n                handler.flush()\n\n\nclass _TextLogger(_Logger):\n    def __init__(self, event_manager: \"EventManager\", config: LoggerConfig) -> None:\n        super().__init__(event_manager, config)\n        self.use_colors = config.use_colors\n        self.use_debug_format = config.line_format == LineFormat.DebugText\n\n    def create_line(self, msg: EventMsg) -> str:\n        return self.create_debug_line(msg) if self.use_debug_format else self.create_info_line(msg)\n\n    def create_info_line(self, msg: EventMsg) -> str:\n        ts: str = datetime.utcnow().strftime(\"%H:%M:%S\")\n        scrubbed_msg: str = self.scrubber(msg.info.msg)  # type: ignore\n        return f\"{self._get_color_tag()}{ts}  {scrubbed_msg}\"\n\n    def create_debug_line(self, msg: EventMsg) -> str:\n        log_line: str = \"\"\n        # Create a separator if this is the beginning of an invocation\n        # TODO: This is an ugly hack, get rid of it if we can\n        if msg.info.name == \"MainReportVersion\":\n            separator = 30 * \"=\"\n            log_line = (\n                f\"\\n\\n{separator} {msg.info.ts} | {self.event_manager.invocation_id} {separator}\\n\"\n            )\n        ts: str = msg.info.ts.strftime(\"%H:%M:%S.%f\")\n        scrubbed_msg: str = self.scrubber(msg.info.msg)  # type: ignore\n        level = msg.info.level\n        log_line += (\n            f\"{self._get_color_tag()}{ts} [{level:<5}]{self._get_thread_name()} {scrubbed_msg}\"\n        )\n        return log_line\n\n    def _get_color_tag(self) -> str:\n        return \"\" if not self.use_colors else Style.RESET_ALL\n\n    def _get_thread_name(self) -> str:\n        thread_name = \"\"\n        if threading.current_thread().name:\n            thread_name = threading.current_thread().name\n            thread_name = thread_name[:10]\n            thread_name = thread_name.ljust(10, \" \")\n            thread_name = f\" [{thread_name}]:\"\n        return thread_name\n\n\nclass _JsonLogger(_Logger):\n    def create_line(self, msg: EventMsg) -> str:\n        from dbt.events.functions import msg_to_dict\n\n        msg_dict = msg_to_dict(msg)\n        raw_log_line = json.dumps(msg_dict, sort_keys=True)\n        line = self.scrubber(raw_log_line)  # type: ignore\n        return line\n\n\nclass EventManager:\n    def __init__(self) -> None:\n        self.loggers: List[_Logger] = []\n        self.callbacks: List[Callable[[EventMsg], None]] = []\n        self.invocation_id: str = str(uuid4())\n\n    def fire_event(self, e: BaseEvent, level: EventLevel = None) -> None:\n        msg = msg_from_base_event(e, level=level)\n        for logger in self.loggers:\n            if logger.filter(msg):  # type: ignore\n                logger.write_line(msg)\n\n        for callback in self.callbacks:\n            callback(msg)\n\n    def add_logger(self, config: LoggerConfig):\n        logger = (\n            _JsonLogger(self, config)\n            if config.line_format == LineFormat.Json\n            else _TextLogger(self, config)\n        )\n        logger.event_manager = self\n        self.loggers.append(logger)\n\n    def flush(self):\n        for logger in self.loggers:\n            logger.flush()\n", "path": "core/dbt/events/eventmgr.py"}]}