nautilus_infrastructure/python/redis/
cache.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 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 bytes::Bytes;
17use nautilus_common::{cache::CacheConfig, live::get_runtime};
18use nautilus_core::{
19    UUID4,
20    python::{to_pyruntime_err, to_pyvalue_err},
21};
22use nautilus_model::{
23    data::{CustomData, DataType},
24    identifiers::{AccountId, ClientOrderId, PositionId, TraderId},
25    python::{
26        account::account_any_to_pyobject, instruments::instrument_any_to_pyobject,
27        orders::order_any_to_pyobject,
28    },
29};
30use pyo3::{
31    IntoPyObjectExt,
32    prelude::*,
33    types::{PyBytes, PyDict},
34};
35use serde_json::Value;
36
37use crate::redis::{
38    cache::{RedisCacheConfig, RedisCacheDatabase},
39    queries::DatabaseQueries,
40};
41
42#[pymethods]
43impl RedisCacheDatabase {
44    /// Creates a new `RedisCacheDatabase` instance for the given `trader_id`, `instance_id`, and `config`.
45    #[new]
46    #[pyo3(signature = (trader_id, instance_id, config_json, database_config_json=None))]
47    fn py_new(
48        trader_id: TraderId,
49        instance_id: UUID4,
50        config_json: &[u8],
51        database_config_json: Option<&[u8]>,
52    ) -> PyResult<Self> {
53        let (config, database) = parse_inputs(config_json, database_config_json)?;
54        let result = get_runtime()
55            .block_on(async { Self::new(trader_id, instance_id, config, database).await });
56        result.map_err(to_pyruntime_err)
57    }
58
59    #[pyo3(name = "close")]
60    fn py_close(&mut self) {
61        self.close();
62    }
63
64    #[pyo3(name = "flushdb")]
65    fn py_flushdb(&mut self) {
66        get_runtime().block_on(async { self.flushdb().await });
67    }
68
69    /// Retrieves all keys matching the given `pattern` from Redis for this trader.
70    ///
71    /// # Errors
72    ///
73    /// Returns an error if the underlying Redis scan operation fails.
74    #[pyo3(name = "keys")]
75    fn py_keys(&mut self, pattern: &str) -> PyResult<Vec<String>> {
76        let result = get_runtime().block_on(async { self.keys(pattern).await });
77        result.map_err(to_pyruntime_err)
78    }
79
80    #[pyo3(name = "load_all")]
81    fn py_load_all(&mut self) -> PyResult<Py<PyAny>> {
82        let result = get_runtime().block_on(async {
83            DatabaseQueries::load_all(&self.con, self.get_encoding(), self.get_trader_key()).await
84        });
85
86        match result {
87            Ok(cache_map) => Python::attach(|py| {
88                let dict = PyDict::new(py);
89
90                // Load currencies
91                let currencies_dict = PyDict::new(py);
92                for (key, value) in cache_map.currencies {
93                    currencies_dict
94                        .set_item(key.to_string(), value)
95                        .map_err(to_pyvalue_err)?;
96                }
97                dict.set_item("currencies", currencies_dict)
98                    .map_err(to_pyvalue_err)?;
99
100                // Load instruments
101                let instruments_dict = PyDict::new(py);
102                for (key, value) in cache_map.instruments {
103                    let py_object = instrument_any_to_pyobject(py, value)?;
104                    instruments_dict
105                        .set_item(key, py_object)
106                        .map_err(to_pyvalue_err)?;
107                }
108                dict.set_item("instruments", instruments_dict)
109                    .map_err(to_pyvalue_err)?;
110
111                // Load synthetics
112                let synthetics_dict = PyDict::new(py);
113                for (key, value) in cache_map.synthetics {
114                    synthetics_dict
115                        .set_item(key, value)
116                        .map_err(to_pyvalue_err)?;
117                }
118                dict.set_item("synthetics", synthetics_dict)
119                    .map_err(to_pyvalue_err)?;
120
121                // Load accounts
122                let accounts_dict = PyDict::new(py);
123                for (key, value) in cache_map.accounts {
124                    let py_object = account_any_to_pyobject(py, value)?;
125                    accounts_dict
126                        .set_item(key, py_object)
127                        .map_err(to_pyvalue_err)?;
128                }
129                dict.set_item("accounts", accounts_dict)
130                    .map_err(to_pyvalue_err)?;
131
132                // Load orders
133                let orders_dict = PyDict::new(py);
134                for (key, value) in cache_map.orders {
135                    let py_object = order_any_to_pyobject(py, value)?;
136                    orders_dict
137                        .set_item(key, py_object)
138                        .map_err(to_pyvalue_err)?;
139                }
140                dict.set_item("orders", orders_dict)
141                    .map_err(to_pyvalue_err)?;
142
143                // Load positions
144                let positions_dict = PyDict::new(py);
145                for (key, value) in cache_map.positions {
146                    positions_dict
147                        .set_item(key, value)
148                        .map_err(to_pyvalue_err)?;
149                }
150                dict.set_item("positions", positions_dict)
151                    .map_err(to_pyvalue_err)?;
152
153                dict.into_py_any(py)
154            }),
155            Err(e) => Err(to_pyruntime_err(e)),
156        }
157    }
158
159    /// Reads the value(s) associated with `key` for this trader from Redis.
160    ///
161    /// # Errors
162    ///
163    /// Returns an error if the underlying Redis read operation fails.
164    #[pyo3(name = "read")]
165    fn py_read(&mut self, py: Python, key: &str) -> PyResult<Vec<Py<PyAny>>> {
166        let result = get_runtime().block_on(async { self.read(key).await });
167        match result {
168            Ok(result) => {
169                let vec_py_bytes = result
170                    .into_iter()
171                    .map(|r| PyBytes::new(py, r.as_ref()).into())
172                    .collect::<Vec<Py<PyAny>>>();
173                Ok(vec_py_bytes)
174            }
175            Err(e) => Err(to_pyruntime_err(e)),
176        }
177    }
178
179    /// Reads multiple values using bulk operations for efficiency.
180    ///
181    /// # Errors
182    ///
183    /// Returns an error if the underlying Redis read operation fails.
184    #[pyo3(name = "read_bulk")]
185    #[expect(clippy::needless_pass_by_value)]
186    fn py_read_bulk(&mut self, py: Python, keys: Vec<String>) -> PyResult<Vec<Option<Py<PyAny>>>> {
187        let result = get_runtime().block_on(async { self.read_bulk(&keys).await });
188        match result {
189            Ok(results) => {
190                let vec_py_bytes = results
191                    .into_iter()
192                    .map(|opt| opt.map(|bytes| PyBytes::new(py, bytes.as_ref()).into()))
193                    .collect::<Vec<Option<Py<PyAny>>>>();
194                Ok(vec_py_bytes)
195            }
196            Err(e) => Err(to_pyruntime_err(e)),
197        }
198    }
199
200    /// Sends an insert command for `key` with optional `payload` to Redis via the background task.
201    ///
202    /// # Errors
203    ///
204    /// Returns an error if the command cannot be sent to the background task channel.
205    #[pyo3(name = "insert")]
206    fn py_insert(&mut self, key: String, payload: Vec<Vec<u8>>) -> PyResult<()> {
207        let payload: Vec<Bytes> = payload.into_iter().map(Bytes::from).collect();
208        self.insert(key, Some(payload)).map_err(to_pyvalue_err)
209    }
210
211    /// Sends an update command for `key` with optional `payload` to Redis via the background task.
212    ///
213    /// # Errors
214    ///
215    /// Returns an error if the command cannot be sent to the background task channel.
216    #[pyo3(name = "update")]
217    fn py_update(&mut self, key: String, payload: Vec<Vec<u8>>) -> PyResult<()> {
218        let payload: Vec<Bytes> = payload.into_iter().map(Bytes::from).collect();
219        self.update(key, Some(payload)).map_err(to_pyvalue_err)
220    }
221
222    /// Sends a delete command for `key` with optional `payload` to Redis via the background task.
223    ///
224    /// # Errors
225    ///
226    /// Returns an error if the command cannot be sent to the background task channel.
227    #[pyo3(name = "delete")]
228    #[pyo3(signature = (key, payload=None))]
229    fn py_delete(&mut self, key: String, payload: Option<Vec<Vec<u8>>>) -> PyResult<()> {
230        let payload: Option<Vec<Bytes>> =
231            payload.map(|vec| vec.into_iter().map(Bytes::from).collect());
232        self.delete(key, payload).map_err(to_pyvalue_err)
233    }
234
235    /// Delete the given order from the database with full index cleanup.
236    ///
237    /// # Errors
238    ///
239    /// Returns an error if the command cannot be sent to the background task channel.
240    #[pyo3(name = "delete_order")]
241    fn py_delete_order(&mut self, client_order_id: &str) -> PyResult<()> {
242        let client_order_id = ClientOrderId::new(client_order_id);
243        self.delete_order(&client_order_id).map_err(to_pyvalue_err)
244    }
245
246    /// Delete the given position from the database with full index cleanup.
247    ///
248    /// # Errors
249    ///
250    /// Returns an error if the command cannot be sent to the background task channel.
251    #[pyo3(name = "delete_position")]
252    fn py_delete_position(&mut self, position_id: &str) -> PyResult<()> {
253        let position_id = PositionId::new(position_id);
254        self.delete_position(&position_id).map_err(to_pyvalue_err)
255    }
256
257    /// Delete the given account event from the database.
258    ///
259    /// # Errors
260    ///
261    /// Returns an error if the command cannot be sent to the background task channel.
262    #[pyo3(name = "delete_account_event")]
263    fn py_delete_account_event(&mut self, account_id: &str, event_id: &str) -> PyResult<()> {
264        let account_id = AccountId::new(account_id);
265        self.delete_account_event(&account_id, event_id)
266            .map_err(to_pyvalue_err)
267    }
268
269    /// Stores custom data in Redis (key format: `custom:<ts_init_020>:<uuid>`, value: full JSON).
270    ///
271    /// # Errors
272    ///
273    /// Returns an error if serialization fails or the insert command cannot be sent.
274    #[pyo3(name = "add_custom_data")]
275    #[expect(clippy::needless_pass_by_value)]
276    fn py_add_custom_data(&mut self, data: CustomData) -> PyResult<()> {
277        self.add_custom_data(&data).map_err(to_pyvalue_err)
278    }
279
280    /// Loads custom data from Redis matching the given `data_type` (blocking).
281    ///
282    /// Spawns the async query on the global Nautilus runtime and blocks until
283    /// the result arrives via a channel. Safe from any thread context (Python,
284    /// test runtimes, plain threads).
285    #[pyo3(name = "load_custom_data")]
286    #[expect(clippy::needless_pass_by_value)]
287    fn py_load_custom_data(
288        &mut self,
289        py: Python<'_>,
290        data_type: DataType,
291    ) -> PyResult<Vec<CustomData>> {
292        py.detach(|| self.load_custom_data(&data_type).map_err(to_pyvalue_err))
293    }
294}
295
296fn parse_inputs(
297    config_json: &[u8],
298    database_config_json: Option<&[u8]>,
299) -> PyResult<(CacheConfig, RedisCacheConfig)> {
300    let mut config_value: Value = serde_json::from_slice(config_json).map_err(to_pyvalue_err)?;
301    // TODO: Remove the legacy embedded database path once Python v2 callers use database_config_json.
302    let legacy_database = config_value
303        .as_object_mut()
304        .and_then(|object| object.remove("database"));
305
306    let config = serde_json::from_value(config_value).map_err(to_pyvalue_err)?;
307    let database = match database_config_json {
308        Some(raw) => serde_json::from_slice(raw).map_err(to_pyvalue_err)?,
309        None => match legacy_database {
310            Some(value) => config_from_legacy_database(value)?,
311            None => RedisCacheConfig::default(),
312        },
313    };
314
315    Ok((config, database))
316}
317
318fn config_from_legacy_database(mut value: Value) -> PyResult<RedisCacheConfig> {
319    if value.is_null() {
320        return Ok(RedisCacheConfig::default());
321    }
322
323    remove_legacy_selector(&mut value, "cache database")?;
324    serde_json::from_value(value).map_err(to_pyvalue_err)
325}
326
327fn remove_legacy_selector(value: &mut Value, label: &str) -> PyResult<()> {
328    let Some(object) = value.as_object_mut() else {
329        return Ok(());
330    };
331
332    let selector = object
333        .remove("database_type")
334        .or_else(|| object.remove("type"));
335    let Some(selector) = selector else {
336        return Ok(());
337    };
338    let Some(selector) = selector.as_str() else {
339        return Err(to_pyvalue_err(format!(
340            "invalid {label} type selector, expected string"
341        )));
342    };
343
344    if selector != "redis" {
345        return Err(to_pyvalue_err(format!(
346            "invalid {label} type selector, expected 'redis', was '{selector}'"
347        )));
348    }
349
350    Ok(())
351}
352
353#[cfg(test)]
354mod tests {
355    use rstest::rstest;
356    use serde_json::json;
357
358    use super::*;
359
360    #[rstest]
361    fn test_parse_inputs_accepts_legacy_database() {
362        let config_json = serde_json::to_vec(&json!({
363            "database": {
364                "type": "redis",
365                "host": "redis.example.com",
366                "port": 6380,
367                "password": "secret",
368                "ssl": true,
369            },
370            "encoding": "json",
371            "buffer_interval_ms": 25,
372        }))
373        .unwrap();
374
375        let (config, database) = parse_inputs(&config_json, None).unwrap();
376
377        assert_eq!(config.buffer_interval_ms, Some(25));
378        assert_eq!(database.host, Some("redis.example.com".to_string()));
379        assert_eq!(database.port, Some(6380));
380        assert_eq!(database.password, Some("secret".to_string()));
381        assert!(database.ssl);
382    }
383
384    #[rstest]
385    fn test_parse_inputs_defaults_null_legacy_database() {
386        let config_json = serde_json::to_vec(&json!({
387            "database": null,
388            "buffer_interval_ms": 50,
389        }))
390        .unwrap();
391
392        let (config, database) = parse_inputs(&config_json, None).unwrap();
393
394        assert_eq!(config.buffer_interval_ms, Some(50));
395        assert_eq!(database, RedisCacheConfig::default());
396    }
397
398    #[rstest]
399    fn test_parse_inputs_prefers_explicit_database_config() {
400        let config_json = serde_json::to_vec(&json!({
401            "database": {
402                "type": "redis",
403                "host": "legacy.example.com",
404            },
405        }))
406        .unwrap();
407        let database_config_json = serde_json::to_vec(&json!({
408            "host": "explicit.example.com",
409            "port": 6381,
410        }))
411        .unwrap();
412
413        let (_, database) = parse_inputs(&config_json, Some(&database_config_json)).unwrap();
414
415        assert_eq!(database.host, Some("explicit.example.com".to_string()));
416        assert_eq!(database.port, Some(6381));
417    }
418
419    #[rstest]
420    fn test_parse_inputs_rejects_non_redis_legacy_database() {
421        Python::initialize();
422        let config_json = serde_json::to_vec(&json!({
423            "database": {
424                "type": "postgres",
425            },
426        }))
427        .unwrap();
428
429        let error = parse_inputs(&config_json, None).unwrap_err();
430
431        assert_eq!(
432            error.to_string(),
433            "ValueError: invalid cache database type selector, expected 'redis', was 'postgres'"
434        );
435    }
436}
nautilus_infrastructure/python/redis/cache.rs

nautilus_infrastructure/python/redis/
cache.rs
