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::{
18    cache::{CacheConfig, database::CacheDatabaseFactory},
19    live::get_runtime,
20    python::cache::get_global_cache_database_factory_registry,
21};
22use nautilus_core::{
23    UUID4,
24    python::{to_pyruntime_err, to_pyvalue_err},
25};
26use nautilus_model::{
27    data::{CustomData, DataType},
28    identifiers::{AccountId, ClientOrderId, PositionId, TraderId},
29    python::{
30        account::account_any_to_pyobject, instruments::instrument_any_to_pyobject,
31        orders::order_any_to_pyobject,
32    },
33};
34use pyo3::{
35    IntoPyObjectExt,
36    prelude::*,
37    types::{PyBytes, PyDict},
38};
39use serde_json::Value;
40
41use crate::redis::{
42    cache::{RedisCacheConfig, RedisCacheDatabase},
43    queries::DatabaseQueries,
44};
45
46#[pymethods]
47impl RedisCacheDatabase {
48    /// Creates a new `RedisCacheDatabase` instance for the given `trader_id`, `instance_id`, and `config`.
49    ///
50    /// # Errors
51    ///
52    /// Returns an error if:
53    /// - The database configuration is missing in `config`.
54    /// - Establishing the Redis connection fails.
55    /// - The command processing task cannot be spawned.
56    #[new]
57    #[pyo3(signature = (trader_id, instance_id, config_json, database_config_json=None))]
58    fn py_new(
59        trader_id: TraderId,
60        instance_id: UUID4,
61        config_json: &[u8],
62        database_config_json: Option<&[u8]>,
63    ) -> PyResult<Self> {
64        let (config, database) = parse_inputs(config_json, database_config_json)?;
65        let result = get_runtime()
66            .block_on(async { Self::new(trader_id, instance_id, config, database).await });
67        result.map_err(to_pyruntime_err)
68    }
69
70    #[pyo3(name = "close")]
71    fn py_close(&mut self) {
72        self.close();
73    }
74
75    #[pyo3(name = "flushdb")]
76    fn py_flushdb(&mut self) {
77        get_runtime().block_on(async { self.flushdb().await });
78    }
79
80    /// Retrieves all keys matching the given `pattern` from Redis for this trader.
81    ///
82    /// # Errors
83    ///
84    /// Returns an error if the underlying Redis scan operation fails.
85    #[pyo3(name = "keys")]
86    fn py_keys(&mut self, pattern: &str) -> PyResult<Vec<String>> {
87        let result = get_runtime().block_on(async { self.keys(pattern).await });
88        result.map_err(to_pyruntime_err)
89    }
90
91    #[pyo3(name = "load_all")]
92    fn py_load_all(&mut self) -> PyResult<Py<PyAny>> {
93        let result = get_runtime().block_on(async {
94            DatabaseQueries::load_all(&self.con, self.get_encoding(), self.get_trader_key()).await
95        });
96
97        match result {
98            Ok(cache_map) => Python::attach(|py| {
99                let dict = PyDict::new(py);
100
101                // Load currencies
102                let currencies_dict = PyDict::new(py);
103                for (key, value) in cache_map.currencies {
104                    currencies_dict
105                        .set_item(key.to_string(), value)
106                        .map_err(to_pyvalue_err)?;
107                }
108                dict.set_item("currencies", currencies_dict)
109                    .map_err(to_pyvalue_err)?;
110
111                // Load instruments
112                let instruments_dict = PyDict::new(py);
113                for (key, value) in cache_map.instruments {
114                    let py_object = instrument_any_to_pyobject(py, value)?;
115                    instruments_dict
116                        .set_item(key, py_object)
117                        .map_err(to_pyvalue_err)?;
118                }
119                dict.set_item("instruments", instruments_dict)
120                    .map_err(to_pyvalue_err)?;
121
122                // Load synthetics
123                let synthetics_dict = PyDict::new(py);
124                for (key, value) in cache_map.synthetics {
125                    synthetics_dict
126                        .set_item(key, value)
127                        .map_err(to_pyvalue_err)?;
128                }
129                dict.set_item("synthetics", synthetics_dict)
130                    .map_err(to_pyvalue_err)?;
131
132                // Load accounts
133                let accounts_dict = PyDict::new(py);
134                for (key, value) in cache_map.accounts {
135                    let py_object = account_any_to_pyobject(py, value)?;
136                    accounts_dict
137                        .set_item(key, py_object)
138                        .map_err(to_pyvalue_err)?;
139                }
140                dict.set_item("accounts", accounts_dict)
141                    .map_err(to_pyvalue_err)?;
142
143                // Load orders
144                let orders_dict = PyDict::new(py);
145                for (key, value) in cache_map.orders {
146                    let py_object = order_any_to_pyobject(py, value)?;
147                    orders_dict
148                        .set_item(key, py_object)
149                        .map_err(to_pyvalue_err)?;
150                }
151                dict.set_item("orders", orders_dict)
152                    .map_err(to_pyvalue_err)?;
153
154                // Load positions
155                let positions_dict = PyDict::new(py);
156                for (key, value) in cache_map.positions {
157                    positions_dict
158                        .set_item(key, value)
159                        .map_err(to_pyvalue_err)?;
160                }
161                dict.set_item("positions", positions_dict)
162                    .map_err(to_pyvalue_err)?;
163
164                dict.into_py_any(py)
165            }),
166            Err(e) => Err(to_pyruntime_err(e)),
167        }
168    }
169
170    /// Reads the value(s) associated with `key` for this trader from Redis.
171    ///
172    /// # Errors
173    ///
174    /// Returns an error if the underlying Redis read operation fails.
175    #[pyo3(name = "read")]
176    fn py_read(&mut self, py: Python, key: &str) -> PyResult<Vec<Py<PyAny>>> {
177        let result = get_runtime().block_on(async { self.read(key).await });
178        match result {
179            Ok(result) => {
180                let vec_py_bytes = result
181                    .into_iter()
182                    .map(|r| PyBytes::new(py, r.as_ref()).into())
183                    .collect::<Vec<Py<PyAny>>>();
184                Ok(vec_py_bytes)
185            }
186            Err(e) => Err(to_pyruntime_err(e)),
187        }
188    }
189
190    /// Reads multiple values using bulk operations for efficiency.
191    ///
192    /// # Errors
193    ///
194    /// Returns an error if the underlying Redis read operation fails.
195    #[pyo3(name = "read_bulk")]
196    #[expect(clippy::needless_pass_by_value)]
197    fn py_read_bulk(&mut self, py: Python, keys: Vec<String>) -> PyResult<Vec<Option<Py<PyAny>>>> {
198        let result = get_runtime().block_on(async { self.read_bulk(&keys).await });
199        match result {
200            Ok(results) => {
201                let vec_py_bytes = results
202                    .into_iter()
203                    .map(|opt| opt.map(|bytes| PyBytes::new(py, bytes.as_ref()).into()))
204                    .collect::<Vec<Option<Py<PyAny>>>>();
205                Ok(vec_py_bytes)
206            }
207            Err(e) => Err(to_pyruntime_err(e)),
208        }
209    }
210
211    /// Sends an insert 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 = "insert")]
217    fn py_insert(&mut self, key: String, payload: Vec<Vec<u8>>) -> PyResult<()> {
218        let payload: Vec<Bytes> = payload.into_iter().map(Bytes::from).collect();
219        self.insert(key, Some(payload)).map_err(to_pyvalue_err)
220    }
221
222    /// Sends an update 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 = "update")]
228    fn py_update(&mut self, key: String, payload: Vec<Vec<u8>>) -> PyResult<()> {
229        let payload: Vec<Bytes> = payload.into_iter().map(Bytes::from).collect();
230        self.update(key, Some(payload)).map_err(to_pyvalue_err)
231    }
232
233    /// Sends a delete command for `key` with optional `payload` to Redis via the background task.
234    ///
235    /// # Errors
236    ///
237    /// Returns an error if the command cannot be sent to the background task channel.
238    #[pyo3(name = "delete")]
239    #[pyo3(signature = (key, payload=None))]
240    fn py_delete(&mut self, key: String, payload: Option<Vec<Vec<u8>>>) -> PyResult<()> {
241        let payload: Option<Vec<Bytes>> =
242            payload.map(|vec| vec.into_iter().map(Bytes::from).collect());
243        self.delete(key, payload).map_err(to_pyvalue_err)
244    }
245
246    /// Delete the given order 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_order")]
252    fn py_delete_order(&mut self, client_order_id: &str) -> PyResult<()> {
253        let client_order_id = ClientOrderId::new(client_order_id);
254        self.delete_order(&client_order_id).map_err(to_pyvalue_err)
255    }
256
257    /// Delete the given position from the database with full index cleanup.
258    ///
259    /// # Errors
260    ///
261    /// Returns an error if the command cannot be sent to the background task channel.
262    #[pyo3(name = "delete_position")]
263    fn py_delete_position(&mut self, position_id: &str) -> PyResult<()> {
264        let position_id = PositionId::new(position_id);
265        self.delete_position(&position_id).map_err(to_pyvalue_err)
266    }
267
268    /// Delete the given account event from the database.
269    ///
270    /// # Errors
271    ///
272    /// Returns an error if the command cannot be sent to the background task channel.
273    #[pyo3(name = "delete_account_event")]
274    fn py_delete_account_event(&mut self, account_id: &str, event_id: &str) -> PyResult<()> {
275        let account_id = AccountId::new(account_id);
276        self.delete_account_event(&account_id, event_id)
277            .map_err(to_pyvalue_err)
278    }
279
280    /// Stores custom data in Redis (key format: `custom:<ts_init_020>:<uuid>`, value: full JSON).
281    ///
282    /// # Errors
283    ///
284    /// Returns an error if serialization fails or the insert command cannot be sent.
285    #[pyo3(name = "add_custom_data")]
286    #[expect(clippy::needless_pass_by_value)]
287    fn py_add_custom_data(&mut self, data: CustomData) -> PyResult<()> {
288        self.add_custom_data(&data).map_err(to_pyvalue_err)
289    }
290
291    /// Loads custom data from Redis matching the given `data_type` (blocking).
292    ///
293    /// Spawns the async query on the global Nautilus runtime and blocks until
294    /// the result arrives via a channel. Safe from any thread context (Python,
295    /// test runtimes, plain threads).
296    ///
297    /// # Errors
298    ///
299    /// Returns an error if the query fails or the reply channel is closed.
300    #[pyo3(name = "load_custom_data")]
301    #[expect(clippy::needless_pass_by_value)]
302    fn py_load_custom_data(
303        &mut self,
304        py: Python<'_>,
305        data_type: DataType,
306    ) -> PyResult<Vec<CustomData>> {
307        py.detach(|| self.load_custom_data(&data_type).map_err(to_pyvalue_err))
308    }
309}
310
311#[pymethods]
312#[pyo3_stub_gen::derive::gen_stub_pymethods]
313impl RedisCacheConfig {
314    /// Configuration for a Redis-backed cache database.
315    ///
316    /// Redis 6.2 or higher is required for correct operation.
317    #[new]
318    #[expect(clippy::too_many_arguments)]
319    #[pyo3(signature = (host=None, port=None, username=None, password=None, ssl=None, connection_timeout=None, response_timeout=None, number_of_retries=None, exponent_base=None, max_delay=None, factor=None))]
320    fn py_new(
321        host: Option<String>,
322        port: Option<u16>,
323        username: Option<String>,
324        password: Option<String>,
325        ssl: Option<bool>,
326        connection_timeout: Option<u16>,
327        response_timeout: Option<u16>,
328        number_of_retries: Option<usize>,
329        exponent_base: Option<u64>,
330        max_delay: Option<u64>,
331        factor: Option<u64>,
332    ) -> Self {
333        let default = Self::default();
334        Self {
335            host,
336            port,
337            username,
338            password,
339            ssl: ssl.unwrap_or(default.ssl),
340            connection_timeout: connection_timeout.unwrap_or(default.connection_timeout),
341            response_timeout: response_timeout.unwrap_or(default.response_timeout),
342            number_of_retries: number_of_retries.unwrap_or(default.number_of_retries),
343            exponent_base: exponent_base.unwrap_or(default.exponent_base),
344            max_delay: max_delay.unwrap_or(default.max_delay),
345            factor: factor.unwrap_or(default.factor),
346        }
347    }
348
349    #[getter]
350    fn host(&self) -> Option<&str> {
351        self.host.as_deref()
352    }
353
354    #[getter]
355    const fn port(&self) -> Option<u16> {
356        self.port
357    }
358
359    #[getter]
360    fn username(&self) -> Option<&str> {
361        self.username.as_deref()
362    }
363
364    #[getter]
365    fn password(&self) -> Option<&str> {
366        self.password.as_deref()
367    }
368
369    #[getter]
370    const fn ssl(&self) -> bool {
371        self.ssl
372    }
373
374    #[getter]
375    const fn connection_timeout(&self) -> u16 {
376        self.connection_timeout
377    }
378
379    #[getter]
380    const fn response_timeout(&self) -> u16 {
381        self.response_timeout
382    }
383
384    #[getter]
385    const fn number_of_retries(&self) -> usize {
386        self.number_of_retries
387    }
388
389    #[getter]
390    const fn exponent_base(&self) -> u64 {
391        self.exponent_base
392    }
393
394    #[getter]
395    const fn max_delay(&self) -> u64 {
396        self.max_delay
397    }
398
399    #[getter]
400    const fn factor(&self) -> u64 {
401        self.factor
402    }
403}
404
405#[expect(clippy::needless_pass_by_value)]
406fn extract_redis_cache_database_factory(
407    py: Python<'_>,
408    factory: Py<PyAny>,
409) -> PyResult<Box<dyn CacheDatabaseFactory>> {
410    Ok(Box::new(factory.extract::<RedisCacheConfig>(py)?))
411}
412
413pub(in crate::python) fn register_redis_cache_database_factory() -> PyResult<()> {
414    get_global_cache_database_factory_registry()
415        .register(
416            stringify!(RedisCacheConfig).to_string(),
417            extract_redis_cache_database_factory,
418        )
419        .map_err(to_pyruntime_err)
420}
421
422fn parse_inputs(
423    config_json: &[u8],
424    database_config_json: Option<&[u8]>,
425) -> PyResult<(CacheConfig, RedisCacheConfig)> {
426    let mut config_value: Value = serde_json::from_slice(config_json).map_err(to_pyvalue_err)?;
427    // TODO: Remove the legacy embedded database path once Python v2 callers use database_config_json.
428    let legacy_database = config_value
429        .as_object_mut()
430        .and_then(|object| object.remove("database"));
431
432    let config = serde_json::from_value(config_value).map_err(to_pyvalue_err)?;
433    let database = match database_config_json {
434        Some(raw) => serde_json::from_slice(raw).map_err(to_pyvalue_err)?,
435        None => match legacy_database {
436            Some(value) => config_from_legacy_database(value)?,
437            None => RedisCacheConfig::default(),
438        },
439    };
440
441    Ok((config, database))
442}
443
444fn config_from_legacy_database(mut value: Value) -> PyResult<RedisCacheConfig> {
445    if value.is_null() {
446        return Ok(RedisCacheConfig::default());
447    }
448
449    remove_legacy_selector(&mut value, "cache database")?;
450    serde_json::from_value(value).map_err(to_pyvalue_err)
451}
452
453fn remove_legacy_selector(value: &mut Value, label: &str) -> PyResult<()> {
454    let Some(object) = value.as_object_mut() else {
455        return Ok(());
456    };
457
458    let selector = object
459        .remove("database_type")
460        .or_else(|| object.remove("type"));
461    let Some(selector) = selector else {
462        return Ok(());
463    };
464    let Some(selector) = selector.as_str() else {
465        return Err(to_pyvalue_err(format!(
466            "invalid {label} type selector, expected string"
467        )));
468    };
469
470    if selector != "redis" {
471        return Err(to_pyvalue_err(format!(
472            "invalid {label} type selector, expected 'redis', was '{selector}'"
473        )));
474    }
475
476    Ok(())
477}
478
479#[cfg(test)]
480mod tests {
481    use rstest::rstest;
482    use serde_json::json;
483
484    use super::*;
485
486    #[rstest]
487    fn test_parse_inputs_accepts_legacy_database() {
488        let config_json = serde_json::to_vec(&json!({
489            "database": {
490                "type": "redis",
491                "host": "redis.example.com",
492                "port": 6380,
493                "password": "secret",
494                "ssl": true,
495            },
496            "encoding": "json",
497            "buffer_interval_ms": 25,
498        }))
499        .unwrap();
500
501        let (config, database) = parse_inputs(&config_json, None).unwrap();
502
503        assert_eq!(config.buffer_interval_ms, Some(25));
504        assert_eq!(database.host, Some("redis.example.com".to_string()));
505        assert_eq!(database.port, Some(6380));
506        assert_eq!(database.password, Some("secret".to_string()));
507        assert!(database.ssl);
508    }
509
510    #[rstest]
511    fn test_parse_inputs_defaults_null_legacy_database() {
512        let config_json = serde_json::to_vec(&json!({
513            "database": null,
514            "buffer_interval_ms": 50,
515        }))
516        .unwrap();
517
518        let (config, database) = parse_inputs(&config_json, None).unwrap();
519
520        assert_eq!(config.buffer_interval_ms, Some(50));
521        assert_eq!(database, RedisCacheConfig::default());
522    }
523
524    #[rstest]
525    fn test_parse_inputs_prefers_explicit_database_config() {
526        let config_json = serde_json::to_vec(&json!({
527            "database": {
528                "type": "redis",
529                "host": "legacy.example.com",
530            },
531        }))
532        .unwrap();
533        let database_config_json = serde_json::to_vec(&json!({
534            "host": "explicit.example.com",
535            "port": 6381,
536        }))
537        .unwrap();
538
539        let (_, database) = parse_inputs(&config_json, Some(&database_config_json)).unwrap();
540
541        assert_eq!(database.host, Some("explicit.example.com".to_string()));
542        assert_eq!(database.port, Some(6381));
543    }
544
545    #[rstest]
546    fn test_parse_inputs_rejects_non_redis_legacy_database() {
547        Python::initialize();
548        let config_json = serde_json::to_vec(&json!({
549            "database": {
550                "type": "postgres",
551            },
552        }))
553        .unwrap();
554
555        let error = parse_inputs(&config_json, None).unwrap_err();
556
557        assert_eq!(
558            error.to_string(),
559            "ValueError: invalid cache database type selector, expected 'redis', was 'postgres'"
560        );
561    }
562}
nautilus_infrastructure/python/redis/cache.rs

nautilus_infrastructure/python/redis/
cache.rs
