Repository: nautechsystems/nautilus_trader
File: nautilus_trader/adapters/databento/data.py (Python)

Code before:
            await self._check_live_client_started(dataset, live_client)
        except asyncio.CancelledError:
            self._log.warning("Canceled task 'subscribe_imbalance'")

    async def _subscribe_statistics(self, data_type: DataType) -> None:
        try:
            # TODO: Create `DatabentoTimeSeriesParams`
            instrument_id: InstrumentId = data_type.metadata["instrument_id"]
            dataset: Dataset = self._loader.get_dataset_for_venue(instrument_id.venue)
            live_client = self._get_live_client(dataset)
            live_client.subscribe(
                schema=DatabentoSchema.STATISTICS.value,
                instrument_ids=[instrument_id_to_pyo3(instrument_id)],
            )
            await self._check_live_client_started(dataset, live_client)
        except asyncio.CancelledError:
            self._log.warning("Canceled task 'subscribe_statistics'")

    async def _subscribe_instruments(self, command: SubscribeInstruments) -> None:
        # Replace method in child class, for exchange specific data types.
        raise NotImplementedError("Cannot subscribe to all instruments (not currently supported).")

    async def _subscribe_instrument(self, command: SubscribeInstrument) -> None:
        try:
            dataset: Dataset = self._loader.get_dataset_for_venue(command.instrument_id.venue)
            start = command.params.get("start")

            live_client = self._get_live_client(dataset)
            live_client.subscribe(
                schema=DatabentoSchema.DEFINITION.value,
                instrument_ids=[instrument_id_to_pyo3(command.instrument_id)],
                start=start,
            )
            await self._check_live_client_started(dataset, live_client)
        except asyncio.CancelledError:
            self._log.warning("Canceled task 'subscribe_instrument'")

    async def _subscribe_parent_symbols(
        self,
        dataset: Dataset,
        parent_symbols: set[InstrumentId],
    ) -> None:
        try:
            live_client = self._get_live_client(dataset)
            live_client.subscribe(
                schema=DatabentoSchema.DEFINITION.value,
                instrument_ids=sorted(  # type: ignore[type-var]
                    [instrument_id_to_pyo3(instrument_id) for instrument_id in parent_symbols],
                ),
                stype_in="parent",
            )

Review comment (bug):
I think any time we access `dict[str, Any]` we should use inline type annotations. This helps static analyzers, and declares the expected type to prevent downstream bugs.

Code after:
            await self._check_live_client_started(dataset, live_client)
        except asyncio.CancelledError:
            self._log.warning("Canceled task 'subscribe_imbalance'")

    async def _subscribe_statistics(self, data_type: DataType) -> None:
        try:
            # TODO: Create `DatabentoTimeSeriesParams`
            instrument_id: InstrumentId = data_type.metadata["instrument_id"]
            dataset: Dataset = self._loader.get_dataset_for_venue(instrument_id.venue)
            live_client = self._get_live_client(dataset)
            live_client.subscribe(
                schema=DatabentoSchema.STATISTICS.value,
                instrument_ids=[instrument_id_to_pyo3(instrument_id)],
            )
            await self._check_live_client_started(dataset, live_client)
        except asyncio.CancelledError:
            self._log.warning("Canceled task 'subscribe_statistics'")

    async def _subscribe_instruments(self, command: SubscribeInstruments) -> None:
        # Replace method in child class, for exchange specific data types.
        raise NotImplementedError("Cannot subscribe to all instruments (not currently supported).")

    async def _subscribe_instrument(self, command: SubscribeInstrument) -> None:
        try:
            dataset: Dataset = self._loader.get_dataset_for_venue(command.instrument_id.venue)
            start: int | None = command.params.get("start")

            live_client = self._get_live_client(dataset)
            live_client.subscribe(
                schema=DatabentoSchema.DEFINITION.value,
                instrument_ids=[instrument_id_to_pyo3(command.instrument_id)],
                start=start,
            )
            await self._check_live_client_started(dataset, live_client)
        except asyncio.CancelledError:
            self._log.warning("Canceled task 'subscribe_instrument'")

    async def _subscribe_parent_symbols(
        self,
        dataset: Dataset,
        parent_symbols: set[InstrumentId],
    ) -> None:
        try:
            live_client = self._get_live_client(dataset)
            live_client.subscribe(
                schema=DatabentoSchema.DEFINITION.value,
                instrument_ids=sorted(  # type: ignore[type-var]
                    [instrument_id_to_pyo3(instrument_id) for instrument_id in parent_symbols],
                ),
                stype_in="parent",
            )