nautilus_common/python/
logging.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2025 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16use log::LevelFilter;
17use nautilus_core::{UUID4, python::to_pyvalue_err};
18use nautilus_model::identifiers::TraderId;
19use pyo3::prelude::*;
20use ustr::Ustr;
21
22use crate::{
23    enums::{LogColor, LogLevel},
24    logging::{
25        self, headers,
26        logger::{self, LogGuard, LoggerConfig},
27        logging_clock_set_realtime_mode, logging_clock_set_static_mode,
28        logging_clock_set_static_time, logging_set_bypass, map_log_level_to_filter,
29        parse_level_filter_str,
30        writer::FileWriterConfig,
31    },
32};
33
34#[pymethods]
35impl LoggerConfig {
36    /// Creates a [`LoggerConfig`] from a spec string.
37    ///
38    /// # Errors
39    ///
40    /// Returns a Python exception if the spec string is invalid.
41    #[staticmethod]
42    #[pyo3(name = "from_spec")]
43    pub fn py_from_spec(spec: String) -> PyResult<Self> {
44        Self::from_spec(&spec).map_err(to_pyvalue_err)
45    }
46}
47
48#[pymethods]
49impl FileWriterConfig {
50    #[new]
51    #[pyo3(signature = (directory=None, file_name=None, file_format=None, file_rotate=None))]
52    #[must_use]
53    pub fn py_new(
54        directory: Option<String>,
55        file_name: Option<String>,
56        file_format: Option<String>,
57        file_rotate: Option<(u64, u32)>,
58    ) -> Self {
59        Self::new(directory, file_name, file_format, file_rotate)
60    }
61}
62
63/// Initialize tracing.
64///
65/// Tracing is meant to be used to trace/debug async Rust code. It can be
66/// configured to filter modules and write up to a specific level only using
67/// by passing a configuration using the `RUST_LOG` environment variable.
68///
69/// # Safety
70///
71/// Should only be called once during an applications run, ideally at the
72/// beginning of the run.
73///
74/// # Errors
75///
76/// Returns an error if tracing subscriber fails to initialize.
77#[pyfunction()]
78#[pyo3(name = "init_tracing")]
79pub fn py_init_tracing() -> PyResult<()> {
80    logging::init_tracing().map_err(to_pyvalue_err)
81}
82
83/// Initialize logging.
84///
85/// Logging should be used for Python and sync Rust logic which is most of
86/// the components in the [nautilus_trader](https://pypi.org/project/nautilus_trader) package.
87/// Logging can be configured to filter components and write up to a specific level only
88/// by passing a configuration using the `NAUTILUS_LOG` environment variable.
89///
90/// # Safety
91///
92/// Should only be called once during an applications run, ideally at the
93/// beginning of the run.
94/// Initializes logging via Python interface.
95///
96/// # Errors
97///
98/// Returns a Python exception if logger initialization fails.
99#[pyfunction]
100#[pyo3(name = "init_logging")]
101#[allow(clippy::too_many_arguments)]
102#[pyo3(signature = (trader_id, instance_id, level_stdout, level_file=None, component_levels=None, directory=None, file_name=None, file_format=None, file_rotate=None, is_colored=None, is_bypassed=None, print_config=None, log_components_only=None))]
103pub fn py_init_logging(
104    trader_id: TraderId,
105    instance_id: UUID4,
106    level_stdout: LogLevel,
107    level_file: Option<LogLevel>,
108    component_levels: Option<std::collections::HashMap<String, String>>,
109    directory: Option<String>,
110    file_name: Option<String>,
111    file_format: Option<String>,
112    file_rotate: Option<(u64, u32)>,
113    is_colored: Option<bool>,
114    is_bypassed: Option<bool>,
115    print_config: Option<bool>,
116    log_components_only: Option<bool>,
117) -> PyResult<LogGuard> {
118    let level_file = level_file.map_or(LevelFilter::Off, map_log_level_to_filter);
119
120    let component_levels = parse_component_levels(component_levels).map_err(to_pyvalue_err)?;
121
122    let config = LoggerConfig::new(
123        map_log_level_to_filter(level_stdout),
124        level_file,
125        component_levels,
126        log_components_only.unwrap_or(false),
127        is_colored.unwrap_or(true),
128        print_config.unwrap_or(false),
129    );
130
131    let file_config = FileWriterConfig::new(directory, file_name, file_format, file_rotate);
132
133    if is_bypassed.unwrap_or(false) {
134        logging_set_bypass();
135    }
136
137    logging::init_logging(trader_id, instance_id, config, file_config).map_err(to_pyvalue_err)
138}
139
140#[pyfunction()]
141#[pyo3(name = "logger_flush")]
142pub fn py_logger_flush() {
143    log::logger().flush();
144}
145
146fn parse_component_levels(
147    original_map: Option<std::collections::HashMap<String, String>>,
148) -> anyhow::Result<std::collections::HashMap<Ustr, LevelFilter>> {
149    match original_map {
150        Some(map) => {
151            let mut new_map = std::collections::HashMap::new();
152            for (key, value) in map {
153                let ustr_key = Ustr::from(&key);
154                let level = parse_level_filter_str(&value)?;
155                new_map.insert(ustr_key, level);
156            }
157            Ok(new_map)
158        }
159        None => Ok(std::collections::HashMap::new()),
160    }
161}
162
163/// Create a new log event.
164#[pyfunction]
165#[pyo3(name = "logger_log")]
166pub fn py_logger_log(level: LogLevel, color: LogColor, component: &str, message: &str) {
167    logger::log(level, color, Ustr::from(component), message);
168}
169
170/// Logs the standard Nautilus system header.
171#[pyfunction]
172#[pyo3(name = "log_header")]
173pub fn py_log_header(trader_id: TraderId, machine_id: &str, instance_id: UUID4, component: &str) {
174    headers::log_header(trader_id, machine_id, instance_id, Ustr::from(component));
175}
176
177/// Logs system information.
178#[pyfunction]
179#[pyo3(name = "log_sysinfo")]
180pub fn py_log_sysinfo(component: &str) {
181    headers::log_sysinfo(Ustr::from(component));
182}
183
184#[pyfunction]
185#[pyo3(name = "logging_clock_set_static_mode")]
186pub fn py_logging_clock_set_static_mode() {
187    logging_clock_set_static_mode();
188}
189
190#[pyfunction]
191#[pyo3(name = "logging_clock_set_realtime_mode")]
192pub fn py_logging_clock_set_realtime_mode() {
193    logging_clock_set_realtime_mode();
194}
195
196#[pyfunction]
197#[pyo3(name = "logging_clock_set_static_time")]
198pub fn py_logging_clock_set_static_time(time_ns: u64) {
199    logging_clock_set_static_time(time_ns);
200}
201
202/// A thin wrapper around the global Rust logger which exposes ergonomic
203/// logging helpers for Python code.
204///
205/// It mirrors the familiar Python `logging` interface while forwarding
206/// all records through the Nautilus logging infrastructure so that log levels
207/// and formatting remain consistent across Rust and Python.
208#[pyclass(
209    module = "nautilus_trader.core.nautilus_pyo3.common",
210    name = "Logger",
211    unsendable
212)]
213#[derive(Debug, Clone)]
214pub struct PyLogger {
215    name: Ustr,
216}
217
218impl PyLogger {
219    pub fn new(name: &str) -> Self {
220        Self {
221            name: Ustr::from(name),
222        }
223    }
224}
225
226#[pymethods]
227impl PyLogger {
228    /// Create a new `Logger` instance.
229    #[new]
230    #[pyo3(signature = (name="Python"))]
231    fn py_new(name: &str) -> Self {
232        Self::new(name)
233    }
234
235    /// The component identifier carried by this logger.
236    #[getter]
237    fn name(&self) -> &str {
238        &self.name
239    }
240
241    /// Emit a TRACE level record.
242    #[pyo3(name = "trace")]
243    fn py_trace(&self, message: &str, color: Option<LogColor>) {
244        self._log(LogLevel::Trace, color, message);
245    }
246
247    /// Emit a DEBUG level record.
248    #[pyo3(name = "debug")]
249    fn py_debug(&self, message: &str, color: Option<LogColor>) {
250        self._log(LogLevel::Debug, color, message);
251    }
252
253    /// Emit an INFO level record.
254    #[pyo3(name = "info")]
255    fn py_info(&self, message: &str, color: Option<LogColor>) {
256        self._log(LogLevel::Info, color, message);
257    }
258
259    /// Emit a WARNING level record.
260    #[pyo3(name = "warning")]
261    fn py_warning(&self, message: &str, color: Option<LogColor>) {
262        self._log(LogLevel::Warning, color, message);
263    }
264
265    /// Emit an ERROR level record.
266    #[pyo3(name = "error")]
267    fn py_error(&self, message: &str, color: Option<LogColor>) {
268        self._log(LogLevel::Error, color, message);
269    }
270
271    /// Emit an ERROR level record with the active Python exception info.
272    #[pyo3(name = "exception")]
273    #[pyo3(signature = (message="", color=None))]
274    fn py_exception(&self, py: Python, message: &str, color: Option<LogColor>) {
275        let mut full_msg = message.to_owned();
276
277        if pyo3::PyErr::occurred(py) {
278            let err = PyErr::fetch(py);
279            let err_str = err.to_string();
280            if full_msg.is_empty() {
281                full_msg = err_str;
282            } else {
283                full_msg = format!("{full_msg}: {err_str}");
284            }
285        }
286
287        self._log(LogLevel::Error, color, &full_msg);
288    }
289
290    /// Flush buffered log records.
291    #[pyo3(name = "flush")]
292    fn py_flush(&self) {
293        log::logger().flush();
294    }
295
296    fn _log(&self, level: LogLevel, color: Option<LogColor>, message: &str) {
297        let color = color.unwrap_or(LogColor::Normal);
298        logger::log(level, color, self.name, message);
299    }
300}
nautilus_common/python/logging.rs

nautilus_common/python/
logging.rs
