async-sync Rust cache and Postgres adapter refactoring

Author: filipmacek

nautilus_core/common/src/cache/core.rs

@@ -2404,10 +2404,10 @@ impl Cache {
 
     /// Returns references to all instrument IDs for the given `venue`.
     #[must_use]
-    pub fn instrument_ids(&self, venue: &Venue) -> Vec<&InstrumentId> {
+    pub fn instrument_ids(&self, venue: Option<&Venue>) -> Vec<&InstrumentId> {
         self.instruments
             .keys()
-            .filter(|i| &i.venue == venue)
+            .filter(|i| venue.is_none() || &i.venue == venue.unwrap())
             .collect()
     }
 

nautilus_core/common/src/cache/database.rs

@@ -59,9 +59,12 @@ pub trait CacheDatabaseAdapter {
 
     fn load_index_order_client(&mut self) -> anyhow::Result<HashMap<ClientOrderId, ClientId>>;
 
-    fn load_currency(&mut self, code: &Ustr) -> anyhow::Result<Currency>;
+    fn load_currency(&mut self, code: &Ustr) -> anyhow::Result<Option<Currency>>;
 
-    fn load_instrument(&mut self, instrument_id: &InstrumentId) -> anyhow::Result<InstrumentAny>;
+    fn load_instrument(
+        &mut self,
+        instrument_id: &InstrumentId,
+    ) -> anyhow::Result<Option<InstrumentAny>>;
 
     fn load_synthetic(
         &mut self,
@@ -70,7 +73,7 @@ pub trait CacheDatabaseAdapter {
 
     fn load_account(&mut self, account_id: &AccountId) -> anyhow::Result<Box<dyn Account>>;
 
-    fn load_order(&mut self, client_order_id: &ClientOrderId) -> anyhow::Result<OrderAny>;
+    fn load_order(&mut self, client_order_id: &ClientOrderId) -> anyhow::Result<Option<OrderAny>>;
 
     fn load_position(&mut self, position_id: &PositionId) -> anyhow::Result<Position>;
 

nautilus_core/infrastructure/src/python/sql/cache_database.rs

@@ -15,7 +15,7 @@
 
 use std::collections::HashMap;
 
-use nautilus_common::runtime::get_runtime;
+use nautilus_common::{cache::database::CacheDatabaseAdapter, runtime::get_runtime};
 use nautilus_core::python::to_pyruntime_err;
 use nautilus_model::{
     identifiers::{ClientOrderId, InstrumentId},
@@ -70,15 +70,13 @@ impl PostgresCacheDatabase {
     }
 
     #[pyo3(name = "add")]
-    fn py_add(slf: PyRef<'_, Self>, key: String, value: Vec<u8>) -> PyResult<()> {
-        let result = get_runtime().block_on(async { slf.add(key, value).await });
-        result.map_err(to_pyruntime_err)
+    fn py_add(mut slf: PyRefMut<'_, Self>, key: String, value: Vec<u8>) -> PyResult<()> {
+        slf.add(key, value).map_err(to_pyruntime_err)
     }
 
     #[pyo3(name = "add_currency")]
-    fn py_add_currency(slf: PyRef<'_, Self>, currency: Currency) -> PyResult<()> {
-        let result = get_runtime().block_on(async { slf.add_currency(currency).await });
-        result.map_err(to_pyruntime_err)
+    fn py_add_currency(mut slf: PyRefMut<'_, Self>, currency: Currency) -> PyResult<()> {
+        slf.add_currency(&currency).map_err(to_pyruntime_err)
     }
 
     #[pyo3(name = "load_instrument")]
@@ -88,7 +86,7 @@ impl PostgresCacheDatabase {
         py: Python<'_>,
     ) -> PyResult<Option<PyObject>> {
         get_runtime().block_on(async {
-            let result = DatabaseQueries::load_instrument(&slf.pool, instrument_id)
+            let result = DatabaseQueries::load_instrument(&slf.pool, &instrument_id)
                 .await
                 .unwrap();
             match result {
@@ -116,27 +114,29 @@ impl PostgresCacheDatabase {
 
     #[pyo3(name = "add_instrument")]
     fn py_add_instrument(
-        slf: PyRef<'_, Self>,
+        mut slf: PyRefMut<'_, Self>,
         instrument: PyObject,
         py: Python<'_>,
     ) -> PyResult<()> {
         let instrument_any = pyobject_to_instrument_any(py, instrument)?;
-        let result = get_runtime().block_on(async { slf.add_instrument(instrument_any).await });
-        result.map_err(to_pyruntime_err)
+        slf.add_instrument(&instrument_any)
+            .map_err(to_pyruntime_err)
     }
 
     #[pyo3(name = "add_order")]
-    fn py_add_order(slf: PyRef<'_, Self>, order: PyObject, py: Python<'_>) -> PyResult<()> {
+    fn py_add_order(mut slf: PyRefMut<'_, Self>, order: PyObject, py: Python<'_>) -> PyResult<()> {
         let order_any = convert_pyobject_to_order_any(py, order)?;
-        let result = get_runtime().block_on(async { slf.add_order(order_any).await });
-        result.map_err(to_pyruntime_err)
+        slf.add_order(&order_any).map_err(to_pyruntime_err)
     }
 
     #[pyo3(name = "update_order")]
-    fn py_update_order(slf: PyRef<'_, Self>, order: PyObject, py: Python<'_>) -> PyResult<()> {
+    fn py_update_order(
+        mut slf: PyRefMut<'_, Self>,
+        order: PyObject,
+        py: Python<'_>,
+    ) -> PyResult<()> {
         let order_any = convert_pyobject_to_order_any(py, order)?;
-        let result = get_runtime().block_on(async { slf.update_order(order_any).await });
-        result.map_err(to_pyruntime_err)
+        slf.update_order(&order_any).map_err(to_pyruntime_err)
     }
 
     #[pyo3(name = "load_order")]

nautilus_core/infrastructure/src/redis/cache.rs

@@ -703,10 +703,16 @@ impl CacheDatabaseAdapter for RedisCacheDatabaseAdapter {
         for key in self.database.keys(&format!("{CURRENCIES}*"))? {
             let parts: Vec<&str> = key.as_str().rsplitn(2, ':').collect();
             let currency_code = Ustr::from(parts.first().unwrap());
-            let currency = self.load_currency(&currency_code)?;
-            currencies.insert(currency_code, currency);
+            let result = self.load_currency(&currency_code)?;
+            match result {
+                Some(currency) => {
+                    currencies.insert(currency_code, currency);
+                }
+                None => {
+                    error!("Currency not found: {currency_code}");
+                }
+            }
         }
-
         Ok(currencies)
     }
 
@@ -716,8 +722,15 @@ impl CacheDatabaseAdapter for RedisCacheDatabaseAdapter {
         for key in self.database.keys(&format!("{INSTRUMENTS}*"))? {
             let parts: Vec<&str> = key.as_str().rsplitn(2, ':').collect();
             let instrument_id = InstrumentId::from_str(parts.first().unwrap())?;
-            let instrument = self.load_instrument(&instrument_id)?;
-            instruments.insert(instrument_id, instrument);
+            let result = self.load_instrument(&instrument_id)?;
+            match result {
+                Some(instrument) => {
+                    instruments.insert(instrument_id, instrument);
+                }
+                None => {
+                    error!("Instrument not found: {instrument_id}");
+                }
+            }
         }
 
         Ok(instruments)
@@ -755,10 +768,16 @@ impl CacheDatabaseAdapter for RedisCacheDatabaseAdapter {
         for key in self.database.keys(&format!("{ORDERS}*"))? {
             let parts: Vec<&str> = key.as_str().rsplitn(2, ':').collect();
             let client_order_id = ClientOrderId::from(*parts.first().unwrap());
-            let order = self.load_order(&client_order_id)?;
-            orders.insert(client_order_id, order);
+            let result = self.load_order(&client_order_id)?;
+            match result {
+                Some(order) => {
+                    orders.insert(client_order_id, order);
+                }
+                None => {
+                    error!("Order not found: {client_order_id}");
+                }
+            }
         }
-
         Ok(orders)
     }
 
@@ -783,11 +802,14 @@ impl CacheDatabaseAdapter for RedisCacheDatabaseAdapter {
         todo!()
     }
 
-    fn load_currency(&mut self, code: &Ustr) -> anyhow::Result<Currency> {
+    fn load_currency(&mut self, code: &Ustr) -> anyhow::Result<Option<Currency>> {
         todo!()
     }
 
-    fn load_instrument(&mut self, instrument_id: &InstrumentId) -> anyhow::Result<InstrumentAny> {
+    fn load_instrument(
+        &mut self,
+        instrument_id: &InstrumentId,
+    ) -> anyhow::Result<Option<InstrumentAny>> {
         todo!()
     }
 
@@ -802,7 +824,7 @@ impl CacheDatabaseAdapter for RedisCacheDatabaseAdapter {
         todo!()
     }
 
-    fn load_order(&mut self, client_order_id: &ClientOrderId) -> anyhow::Result<OrderAny> {
+    fn load_order(&mut self, client_order_id: &ClientOrderId) -> anyhow::Result<Option<OrderAny>> {
         todo!()
     }