nautilus_persistence/python/
config.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
16//! Python bindings for persistence configuration types.
17
18use std::str::FromStr;
19
20use nautilus_common::python::config_error_to_pyvalue_err;
21use nautilus_core::{
22    DurationNanos, UnixNanos, from_pydict,
23    python::{enums::parse_enum, params::params_to_pydict, to_pytype_err, to_pyvalue_err},
24};
25use nautilus_model::{
26    data::{NautilusDataType, NautilusRecordType},
27    instruments::NautilusInstrumentType,
28    python::{
29        common::EnumIterator,
30        data::{PyNautilusDataType, PyNautilusRecordType},
31        instruments::PyNautilusInstrumentType,
32    },
33};
34use pyo3::{
35    Bound, Py, PyAny, PyRef, PyResult, PyTypeInfo, Python,
36    types::{PyAnyMethods, PyDict, PyType},
37};
38
39use crate::config::{
40    CatalogBackendType, CatalogCompression, DEFAULT_ROTATION_TIMEZONE, DataCatalogConfig,
41    RotationConfig, RotationMode, StreamingConfig, StreamingRecordFilterConfig,
42};
43
44#[derive(Clone, Debug, Eq, Hash, PartialEq)]
45#[pyo3::pyclass(
46    frozen,
47    name = "CatalogBackend",
48    module = "nautilus_trader.persistence",
49    skip_from_py_object
50)]
51#[pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.persistence")]
52pub struct PyCatalogBackend {
53    inner: CatalogBackendType,
54}
55
56impl PyCatalogBackend {
57    #[must_use]
58    pub const fn new(inner: CatalogBackendType) -> Self {
59        Self { inner }
60    }
61
62    #[must_use]
63    pub fn inner(&self) -> CatalogBackendType {
64        self.inner.clone()
65    }
66}
67
68fn parse_catalog_backend(value: &str) -> PyResult<CatalogBackendType> {
69    value.parse::<CatalogBackendType>().map_err(to_pyvalue_err)
70}
71
72#[pyo3_stub_gen::derive::gen_stub_pymethods]
73#[pyo3::pymethods]
74impl PyCatalogBackend {
75    #[classattr]
76    #[expect(
77        non_snake_case,
78        clippy::use_self,
79        reason = "PyO3 stub generation needs the concrete Python enum type"
80    )]
81    fn Parquet() -> PyCatalogBackend {
82        Self::new(CatalogBackendType::Parquet)
83    }
84
85    #[staticmethod]
86    fn from_str(value: &str) -> PyResult<Self> {
87        parse_catalog_backend(value).map(Self::new)
88    }
89
90    #[staticmethod]
91    #[pyo3(name = "External")]
92    fn py_external(name: &str) -> PyResult<Self> {
93        let backend = name.parse::<CatalogBackendType>().map_err(to_pyvalue_err)?;
94        match backend {
95            CatalogBackendType::External(_) => Ok(Self::new(backend)),
96            CatalogBackendType::Parquet => Err(to_pyvalue_err(format!(
97                "Catalog backend name '{name}' is reserved for a built-in backend"
98            ))),
99        }
100    }
101
102    #[getter]
103    fn name(&self) -> &str {
104        match self.inner {
105            CatalogBackendType::Parquet => "Parquet",
106            CatalogBackendType::External(_) => "External",
107        }
108    }
109
110    #[getter]
111    fn value(&self) -> String {
112        self.inner.to_string()
113    }
114
115    #[getter]
116    fn external_name(&self) -> Option<&str> {
117        match &self.inner {
118            CatalogBackendType::External(name) => Some(name),
119            CatalogBackendType::Parquet => None,
120        }
121    }
122
123    fn __str__(&self) -> String {
124        self.inner.to_string()
125    }
126
127    fn __repr__(&self) -> String {
128        match &self.inner {
129            CatalogBackendType::Parquet => "CatalogBackend.Parquet".to_string(),
130            CatalogBackendType::External(name) => {
131                format!("CatalogBackend.External({name:?})")
132            }
133        }
134    }
135
136    fn __richcmp__(&self, other: &Self, op: pyo3::pyclass::CompareOp, py: Python<'_>) -> Py<PyAny> {
137        use nautilus_core::python::IntoPyObjectNautilusExt;
138
139        match op {
140            pyo3::pyclass::CompareOp::Eq => (self.inner == other.inner).into_py_any_unwrap(py),
141            pyo3::pyclass::CompareOp::Ne => (self.inner != other.inner).into_py_any_unwrap(py),
142            _ => py.NotImplemented(),
143        }
144    }
145
146    #[expect(
147        clippy::cast_possible_truncation,
148        clippy::cast_possible_wrap,
149        reason = "Python hashes use the platform signed integer width"
150    )]
151    fn __hash__(&self) -> isize {
152        use std::{
153            collections::hash_map::DefaultHasher,
154            hash::{Hash, Hasher},
155        };
156
157        let mut hasher = DefaultHasher::new();
158        self.inner.hash(&mut hasher);
159        hasher.finish() as isize
160    }
161}
162
163#[derive(Clone, Debug)]
164#[pyo3::pyclass(
165    frozen,
166    name = "RotationConfig",
167    module = "nautilus_trader.persistence",
168    from_py_object
169)]
170#[pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.persistence")]
171pub struct PyRotationConfig {
172    inner: RotationConfig,
173}
174
175impl From<PyRotationConfig> for RotationConfig {
176    fn from(config: PyRotationConfig) -> Self {
177        config.inner
178    }
179}
180
181impl From<RotationConfig> for PyRotationConfig {
182    fn from(config: RotationConfig) -> Self {
183        Self { inner: config }
184    }
185}
186
187impl PyRotationConfig {
188    fn validated(inner: RotationConfig) -> PyResult<Self> {
189        inner.validate().map_err(config_error_to_pyvalue_err)?;
190        Ok(Self { inner })
191    }
192}
193
194#[pyo3_stub_gen::derive::gen_stub_pymethods]
195#[pyo3::pymethods]
196impl PyRotationConfig {
197    #[staticmethod]
198    fn no_rotation() -> Self {
199        Self {
200            inner: RotationConfig::NoRotation,
201        }
202    }
203
204    #[staticmethod]
205    fn size(max_size: u64) -> PyResult<Self> {
206        Self::validated(RotationConfig::Size { max_size })
207    }
208
209    #[staticmethod]
210    fn interval(interval_ns: u64) -> PyResult<Self> {
211        Self::validated(RotationConfig::Interval {
212            interval_ns: DurationNanos::new(interval_ns),
213        })
214    }
215
216    #[staticmethod]
217    #[pyo3(signature = (interval_ns, schedule_ns, timezone = DEFAULT_ROTATION_TIMEZONE))]
218    fn scheduled_dates(interval_ns: u64, schedule_ns: u64, timezone: &str) -> PyResult<Self> {
219        Self::validated(RotationConfig::ScheduledDates {
220            interval_ns: DurationNanos::new(interval_ns),
221            schedule_ns: UnixNanos::from(schedule_ns),
222            timezone: timezone.to_string(),
223        })
224    }
225
226    #[getter]
227    const fn mode(&self) -> RotationMode {
228        self.inner.mode()
229    }
230
231    #[getter]
232    fn max_size(&self) -> Option<u64> {
233        match self.inner {
234            RotationConfig::Size { max_size } => Some(max_size),
235            _ => None,
236        }
237    }
238
239    #[getter]
240    fn interval_ns(&self) -> Option<u64> {
241        match self.inner {
242            RotationConfig::Interval { interval_ns }
243            | RotationConfig::ScheduledDates { interval_ns, .. } => Some(interval_ns.as_u64()),
244            _ => None,
245        }
246    }
247
248    #[getter]
249    fn schedule_ns(&self) -> Option<u64> {
250        match self.inner {
251            RotationConfig::ScheduledDates { schedule_ns, .. } => Some(schedule_ns.as_u64()),
252            _ => None,
253        }
254    }
255
256    #[getter]
257    fn timezone(&self) -> Option<&str> {
258        match &self.inner {
259            RotationConfig::ScheduledDates { timezone, .. } => Some(timezone),
260            _ => None,
261        }
262    }
263
264    fn __repr__(&self) -> String {
265        format!("{:?}", self.inner)
266    }
267}
268
269#[pyo3_stub_gen::derive::gen_stub_pymethods]
270#[pyo3::pymethods]
271#[allow(
272    clippy::trivially_copy_pass_by_ref,
273    reason = "PyO3 enum methods take self by reference for Python API parity"
274)]
275impl RotationMode {
276    /// The rotation policy of a streaming writer, without its parameters.
277    #[new]
278    fn py_new(py: Python<'_>, value: &Bound<'_, PyAny>) -> PyResult<Self> {
279        let t = Self::type_object(py);
280        Self::py_from_str(&t, value)
281    }
282
283    const fn __hash__(&self) -> isize {
284        *self as isize
285    }
286
287    fn __str__(&self) -> String {
288        self.to_string()
289    }
290
291    #[getter]
292    #[must_use]
293    pub fn name(&self) -> String {
294        self.to_string()
295    }
296
297    #[getter]
298    #[must_use]
299    pub const fn value(&self) -> u8 {
300        *self as u8
301    }
302
303    #[classmethod]
304    fn variants(_: &Bound<'_, PyType>, py: Python<'_>) -> EnumIterator {
305        EnumIterator::new::<Self>(py)
306    }
307
308    #[classmethod]
309    #[pyo3(name = "from_str")]
310    fn py_from_str(_: &Bound<'_, PyType>, data: &Bound<'_, PyAny>) -> PyResult<Self> {
311        let data_str: &str = data.extract()?;
312        Self::from_str(data_str).map_err(to_pyvalue_err)
313    }
314}
315
316#[pyo3_stub_gen::derive::gen_stub_pymethods]
317#[pyo3::pymethods]
318impl StreamingConfig {
319    /// Configuration streaming live or backtest runs to a persistence writer.
320    ///
321    /// The writer appends Feather files under the local `writer_path`, in one
322    /// `{backtest|sandbox|live}/{instance_id}` directory per run. With a `catalog`, the writer for
323    /// that catalog's backend also promotes the files into it; without one, the Feather files are the
324    /// only output.
325    #[new]
326    #[expect(
327        clippy::too_many_arguments,
328        clippy::fn_params_excessive_bools,
329        reason = "the PyO3 constructor mirrors the public Python configuration signature"
330    )]
331    #[pyo3(signature = (
332        writer_path,
333        catalog = None,
334        flush_interval_ms = 1000,
335        replace_existing = false,
336        rotation_config = None,
337        promotion_interval_ms = None,
338        promote_on_close = true,
339        delete_feather_after_promotion = false,
340        use_ts_event_for_ts_init = false,
341        data_types = None,
342        record_types = None,
343        instrument_types = None,
344        record_filters = None,
345        params = None,
346    ))]
347    fn py_new(
348        writer_path: String,
349        catalog: Option<DataCatalogConfig>,
350        flush_interval_ms: u64,
351        replace_existing: bool,
352        rotation_config: Option<PyRotationConfig>,
353        promotion_interval_ms: Option<u64>,
354        promote_on_close: bool,
355        delete_feather_after_promotion: bool,
356        use_ts_event_for_ts_init: bool,
357        data_types: Option<&Bound<'_, PyAny>>,
358        record_types: Option<&Bound<'_, PyAny>>,
359        instrument_types: Option<&Bound<'_, PyAny>>,
360        record_filters: Option<&Bound<'_, PyAny>>,
361        params: Option<Py<PyDict>>,
362    ) -> pyo3::PyResult<Self> {
363        let mut config = Self::new(
364            writer_path,
365            catalog,
366            flush_interval_ms,
367            replace_existing,
368            rotation_config.map_or(RotationConfig::NoRotation, Into::into),
369        );
370        config.promotion_interval_ms = promotion_interval_ms;
371        config.promote_on_close = promote_on_close;
372        config.delete_feather_after_promotion = delete_feather_after_promotion;
373        config.use_ts_event_for_ts_init = use_ts_event_for_ts_init;
374        let mut parsed_types = py_streaming_types_from_any(data_types)?;
375        parsed_types
376            .records
377            .extend(py_record_types_from_any(record_types)?.unwrap_or_default());
378        parsed_types
379            .instruments
380            .extend(py_instrument_types_from_any(instrument_types)?.unwrap_or_default());
381        config.data_types = (!parsed_types.data.is_empty()).then_some(parsed_types.data);
382        config.record_types = (!parsed_types.records.is_empty()).then_some(parsed_types.records);
383        config.instrument_types =
384            (!parsed_types.instruments.is_empty()).then_some(parsed_types.instruments);
385        config.record_filters = py_record_filters_from_any(record_filters)?;
386        config.params = Python::attach(|py| match params {
387            Some(params) => from_pydict(py, &params),
388            None => Ok(None),
389        })?;
390
391        Ok(config)
392    }
393
394    #[getter]
395    fn writer_path(&self) -> &str {
396        &self.writer_path
397    }
398
399    #[getter]
400    fn catalog(&self) -> Option<DataCatalogConfig> {
401        self.catalog.clone()
402    }
403
404    #[getter]
405    const fn flush_interval_ms(&self) -> u64 {
406        self.flush_interval_ms
407    }
408
409    #[getter]
410    const fn replace_existing(&self) -> bool {
411        self.replace_existing
412    }
413
414    #[getter]
415    fn rotation_config(&self) -> PyRotationConfig {
416        self.rotation_config.clone().into()
417    }
418
419    #[getter]
420    const fn promotion_interval_ms(&self) -> Option<u64> {
421        self.promotion_interval_ms
422    }
423
424    #[getter]
425    const fn promote_on_close(&self) -> bool {
426        self.promote_on_close
427    }
428
429    #[getter]
430    const fn delete_feather_after_promotion(&self) -> bool {
431        self.delete_feather_after_promotion
432    }
433
434    #[getter]
435    const fn use_ts_event_for_ts_init(&self) -> bool {
436        self.use_ts_event_for_ts_init
437    }
438
439    /// Returns the writer backend: `Feather` without a catalog, otherwise the catalog's backend.
440    #[getter]
441    #[pyo3(name = "writer_backend")]
442    fn py_writer_backend(&self) -> String {
443        self.writer_backend().to_string()
444    }
445
446    #[getter]
447    #[pyo3(name = "params")]
448    fn py_params(&self, py: Python<'_>) -> PyResult<Option<Py<PyDict>>> {
449        self.params
450            .as_ref()
451            .map(|params| params_to_pydict(py, params))
452            .transpose()
453    }
454
455    #[getter]
456    fn data_types(&self) -> Option<Vec<PyNautilusDataType>> {
457        self.data_types
458            .clone()
459            .map(|values| values.into_iter().map(PyNautilusDataType::new).collect())
460    }
461
462    #[getter]
463    fn record_types(&self) -> Option<Vec<PyNautilusRecordType>> {
464        self.record_types
465            .clone()
466            .map(|values| values.into_iter().map(PyNautilusRecordType::new).collect())
467    }
468
469    #[getter]
470    fn instrument_types(&self) -> Option<Vec<PyNautilusInstrumentType>> {
471        self.instrument_types.clone().map(|values| {
472            values
473                .into_iter()
474                .map(PyNautilusInstrumentType::new)
475                .collect()
476        })
477    }
478
479    #[getter]
480    fn record_filters(&self, py: Python<'_>) -> PyResult<Option<Py<PyDict>>> {
481        let Some(filters) = &self.record_filters else {
482            return Ok(None);
483        };
484
485        let result = PyDict::new(py);
486        for filter in filters {
487            result.set_item(
488                PyNautilusRecordType::new(filter.record_type),
489                filter.identifiers.clone(),
490            )?;
491        }
492
493        Ok(Some(result.unbind()))
494    }
495
496    fn __repr__(&self) -> String {
497        format!("{self:?}")
498    }
499}
500
501#[pyo3_stub_gen::derive::gen_stub_pymethods]
502#[pyo3::pymethods]
503impl DataCatalogConfig {
504    /// Configuration for a catalog available to request-time historical data loading.
505    #[new]
506    #[expect(
507        clippy::too_many_arguments,
508        reason = "the PyO3 constructor mirrors the public Python configuration signature"
509    )]
510    #[pyo3(signature = (path, fs_protocol = None, catalog_backend = None, params = None, name = None, read_only = false, fs_rust_storage_options = None, batch_size = None, compression = None, max_row_group_size = None))]
511    fn py_new(
512        path: String,
513        fs_protocol: Option<String>,
514        catalog_backend: Option<PyRef<'_, PyCatalogBackend>>,
515        params: Option<Py<PyDict>>,
516        name: Option<String>,
517        read_only: bool,
518        fs_rust_storage_options: Option<std::collections::HashMap<String, String>>,
519        batch_size: Option<usize>,
520        compression: Option<&str>,
521        max_row_group_size: Option<usize>,
522    ) -> pyo3::PyResult<Self> {
523        let catalog_backend = catalog_backend.map(|backend| backend.inner());
524        let compression = compression
525            .map(|compression| parse_enum::<CatalogCompression>(compression, "compression"))
526            .transpose()?;
527
528        let params = Python::attach(|py| match params {
529            Some(params) => from_pydict(py, &params),
530            None => Ok(None),
531        })?;
532
533        Self::builder()
534            .path(path)
535            .maybe_name(name)
536            .maybe_fs_protocol(fs_protocol)
537            .maybe_catalog_backend(catalog_backend)
538            .maybe_batch_size(batch_size)
539            .maybe_compression(compression)
540            .maybe_max_row_group_size(max_row_group_size)
541            .maybe_params(params)
542            .maybe_fs_rust_storage_options(fs_rust_storage_options.map(|options| {
543                options
544                    .into_iter()
545                    .map(|(key, value)| (key, value.into()))
546                    .collect()
547            }))
548            .read_only(read_only)
549            .build()
550            .map_err(config_error_to_pyvalue_err)
551    }
552
553    /// Returns the path to the data catalog.
554    #[getter]
555    #[pyo3(name = "path")]
556    fn py_path(&self) -> &str {
557        self.path()
558    }
559
560    /// Returns the catalog registration name.
561    #[getter]
562    #[pyo3(name = "name")]
563    fn py_name(&self) -> Option<&str> {
564        self.name()
565    }
566
567    /// Returns whether the catalog rejects response write-back.
568    #[getter]
569    #[pyo3(name = "read_only")]
570    fn py_read_only(&self) -> bool {
571        self.read_only()
572    }
573
574    /// Returns the fsspec file system protocol for the data catalog.
575    #[getter]
576    #[pyo3(name = "fs_protocol")]
577    fn py_fs_protocol(&self) -> &str {
578        self.fs_protocol()
579    }
580
581    /// Returns the catalog backend implementation to use.
582    #[getter]
583    #[pyo3(name = "catalog_backend")]
584    fn py_catalog_backend(&self) -> PyCatalogBackend {
585        PyCatalogBackend::new(self.catalog_backend().clone())
586    }
587
588    /// Returns the number of rows per batch the catalog reads and writes.
589    #[getter]
590    #[pyo3(name = "batch_size")]
591    fn py_batch_size(&self) -> Option<usize> {
592        self.batch_size()
593    }
594
595    /// Returns the compression codec of written data files.
596    #[getter]
597    #[pyo3(name = "compression")]
598    fn py_compression(&self) -> Option<String> {
599        self.compression()
600            .map(|compression| compression.to_string())
601    }
602
603    /// Returns the maximum number of rows per written row group.
604    #[getter]
605    #[pyo3(name = "max_row_group_size")]
606    fn py_max_row_group_size(&self) -> Option<usize> {
607        self.max_row_group_size()
608    }
609
610    /// Returns backend-specific catalog parameters.
611    #[getter]
612    #[pyo3(name = "params")]
613    fn py_params(&self, py: Python<'_>) -> PyResult<Option<Py<PyDict>>> {
614        self.params()
615            .map(|params| params_to_pydict(py, params))
616            .transpose()
617    }
618
619    #[getter]
620    fn fs_rust_storage_option_keys(&self) -> Option<Vec<String>> {
621        self.fs_rust_storage_options().map(|options| {
622            let mut keys = options.keys().cloned().collect::<Vec<_>>();
623            keys.sort();
624            keys
625        })
626    }
627
628    fn __repr__(&self) -> String {
629        format!("{self:?}")
630    }
631}
632
633#[derive(Default)]
634struct ParsedStreamingTypes {
635    data: Vec<NautilusDataType>,
636    records: Vec<NautilusRecordType>,
637    instruments: Vec<NautilusInstrumentType>,
638}
639
640fn py_streaming_type_from_any(
641    value: &Bound<'_, PyAny>,
642    parsed: &mut ParsedStreamingTypes,
643) -> pyo3::PyResult<()> {
644    if let Ok(data_type) = value.extract::<PyRef<'_, PyNautilusDataType>>() {
645        parsed.data.push(data_type.inner());
646        return Ok(());
647    }
648
649    if let Ok(record_type) = value.extract::<PyRef<'_, PyNautilusRecordType>>() {
650        parsed.records.push(record_type.inner());
651        return Ok(());
652    }
653
654    if let Ok(instrument_type) = value.extract::<PyRef<'_, PyNautilusInstrumentType>>() {
655        parsed.instruments.push(instrument_type.inner());
656        return Ok(());
657    }
658
659    Err(to_pytype_err(
660        "streaming type must be NautilusDataType, NautilusRecordType, or NautilusInstrumentType",
661    ))
662}
663
664fn py_streaming_types_from_any(
665    values: Option<&Bound<'_, PyAny>>,
666) -> pyo3::PyResult<ParsedStreamingTypes> {
667    let mut parsed = ParsedStreamingTypes::default();
668
669    if let Some(values) = values {
670        for item in values.try_iter()? {
671            py_streaming_type_from_any(&item?, &mut parsed)?;
672        }
673    }
674
675    Ok(parsed)
676}
677
678fn py_record_type_from_any(record_type: &Bound<'_, PyAny>) -> pyo3::PyResult<NautilusRecordType> {
679    if let Ok(record_type) = record_type.extract::<PyRef<'_, PyNautilusRecordType>>() {
680        return Ok(record_type.inner());
681    }
682
683    Err(to_pytype_err("record_type must be NautilusRecordType"))
684}
685
686fn py_record_types_from_any(
687    record_types: Option<&Bound<'_, PyAny>>,
688) -> pyo3::PyResult<Option<Vec<NautilusRecordType>>> {
689    record_types
690        .map(|record_types| {
691            record_types
692                .try_iter()?
693                .map(|item| py_record_type_from_any(&item?))
694                .collect::<pyo3::PyResult<Vec<_>>>()
695        })
696        .transpose()
697}
698
699pub(crate) fn py_instrument_type_from_any(
700    instrument_type: &Bound<'_, PyAny>,
701) -> pyo3::PyResult<NautilusInstrumentType> {
702    if let Ok(instrument_type) = instrument_type.extract::<PyRef<'_, PyNautilusInstrumentType>>() {
703        return Ok(instrument_type.inner());
704    }
705
706    Err(to_pytype_err(
707        "instrument_type must be NautilusInstrumentType",
708    ))
709}
710
711fn py_instrument_types_from_any(
712    instrument_types: Option<&Bound<'_, PyAny>>,
713) -> pyo3::PyResult<Option<Vec<NautilusInstrumentType>>> {
714    instrument_types
715        .map(|instrument_types| {
716            instrument_types
717                .try_iter()?
718                .map(|item| py_instrument_type_from_any(&item?))
719                .collect::<pyo3::PyResult<Vec<_>>>()
720        })
721        .transpose()
722}
723
724fn py_record_filters_from_any(
725    record_filters: Option<&Bound<'_, PyAny>>,
726) -> pyo3::PyResult<Option<Vec<StreamingRecordFilterConfig>>> {
727    let Some(record_filters) = record_filters else {
728        return Ok(None);
729    };
730
731    let record_filters = record_filters.cast::<PyDict>()?;
732    let mut filters = Vec::with_capacity(record_filters.len()?);
733    for (record_type, identifiers) in record_filters {
734        let record_type = py_record_type_from_any(&record_type)?;
735
736        let identifiers = if identifiers.is_none() {
737            None
738        } else if let Ok(identifier) = identifiers.extract::<String>() {
739            Some(vec![identifier])
740        } else {
741            Some(identifiers.extract::<Vec<String>>()?)
742        };
743
744        filters.push(StreamingRecordFilterConfig {
745            record_type,
746            identifiers,
747        });
748    }
749
750    Ok((!filters.is_empty()).then_some(filters))
751}
752
753#[cfg(test)]
754mod tests {
755    use ahash::AHashMap;
756    use rstest::rstest;
757
758    use super::*;
759
760    #[rstest]
761    fn test_data_catalog_config_py_new_keeps_storage_option_values() {
762        Python::initialize();
763
764        let config = DataCatalogConfig::py_new(
765            "bucket/catalog".to_string(),
766            Some("s3".to_string()),
767            None,
768            None,
769            None,
770            false,
771            Some(std::collections::HashMap::from([(
772                "aws_secret_access_key".to_string(),
773                "catalog-secret".to_string(),
774            )])),
775            None,
776            None,
777            None,
778        )
779        .unwrap();
780
781        assert_eq!(
782            config.connect_config().storage_options,
783            Some(AHashMap::from([(
784                "aws_secret_access_key".to_string(),
785                "catalog-secret".to_string(),
786            )]))
787        );
788    }
789}
nautilus_persistence/python/config.rs

nautilus_persistence/python/
config.rs
