jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-18 02:37:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use dict instead of MutableMapping [recorder] (#115318)

Browse Source
This commit is contained in:
Marc Mueller 2024-04-10 03:08:31 +02:00 committed by GitHub
parent d8c8d1a297
commit 2decf6c023
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 32 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
homeassistant/components/history/websocket_api.py
View File

@ -3,7 +3,7 @@
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
from collections.abc import Callable, Iterable, MutableMapping from collections.abc import Callable, Iterable
from dataclasses import dataclass from dataclasses import dataclass
from datetime import datetime as dt, timedelta from datetime import datetime as dt, timedelta
import logging import logging
@ -173,7 +173,7 @@ async def ws_get_history_during_period(
def _generate_stream_message( def _generate_stream_message(
states: MutableMapping[str, list[dict[str, Any]]], states: dict[str, list[dict[str, Any]]],
start_day: dt, start_day: dt,
end_day: dt, end_day: dt,
) -> dict[str, Any]: ) -> dict[str, Any]:
@ -201,7 +201,7 @@ def _generate_websocket_response(
msg_id: int, msg_id: int,
start_time: dt, start_time: dt,
end_time: dt, end_time: dt,
states: MutableMapping[str, list[dict[str, Any]]], states: dict[str, list[dict[str, Any]]],
) -> bytes: ) -> bytes:
"""Generate a websocket response.""" """Generate a websocket response."""
return json_bytes( return json_bytes(
@ -225,7 +225,7 @@ def _generate_historical_response(
) -> tuple[float, dt | None, bytes | None]: ) -> tuple[float, dt | None, bytes | None]:
"""Generate a historical response.""" """Generate a historical response."""
states = cast( states = cast(
MutableMapping[str, list[dict[str, Any]]], dict[str, list[dict[str, Any]]],
history.get_significant_states( history.get_significant_states(
hass, hass,
start_time, start_time,
@ -311,7 +311,7 @@ def _history_compressed_state(state: State, no_attributes: bool) -> dict[str, An
def _events_to_compressed_states( def _events_to_compressed_states(
events: Iterable[Event], no_attributes: bool events: Iterable[Event], no_attributes: bool
) -> MutableMapping[str, list[dict[str, Any]]]: ) -> dict[str, list[dict[str, Any]]]:
"""Convert events to a compressed states.""" """Convert events to a compressed states."""
states_by_entity_ids: dict[str, list[dict[str, Any]]] = {} states_by_entity_ids: dict[str, list[dict[str, Any]]] = {}
for event in events: for event in events:

11
homeassistant/components/recorder/history/__init__.py
View File

@ -2,7 +2,6 @@
from __future__ import annotations from __future__ import annotations
from collections.abc import MutableMapping
from datetime import datetime from datetime import datetime
from typing import Any from typing import Any
@ -43,7 +42,7 @@ def get_full_significant_states_with_session(
include_start_time_state: bool = True, include_start_time_state: bool = True,
significant_changes_only: bool = True, significant_changes_only: bool = True,
no_attributes: bool = False, no_attributes: bool = False,
) -> MutableMapping[str, list[State]]: ) -> dict[str, list[State]]:
"""Return a dict of significant states during a time period.""" """Return a dict of significant states during a time period."""
if not recorder.get_instance(hass).states_meta_manager.active: if not recorder.get_instance(hass).states_meta_manager.active:
from .legacy import ( # pylint: disable=import-outside-toplevel from .legacy import ( # pylint: disable=import-outside-toplevel
@ -68,7 +67,7 @@ def get_full_significant_states_with_session(
def get_last_state_changes( def get_last_state_changes(
hass: HomeAssistant, number_of_states: int, entity_id: str hass: HomeAssistant, number_of_states: int, entity_id: str
) -> MutableMapping[str, list[State]]: ) -> dict[str, list[State]]:
"""Return the last number_of_states.""" """Return the last number_of_states."""
if not recorder.get_instance(hass).states_meta_manager.active: if not recorder.get_instance(hass).states_meta_manager.active:
from .legacy import ( # pylint: disable=import-outside-toplevel from .legacy import ( # pylint: disable=import-outside-toplevel
@ -92,7 +91,7 @@ def get_significant_states(
minimal_response: bool = False, minimal_response: bool = False,
no_attributes: bool = False, no_attributes: bool = False,
compressed_state_format: bool = False, compressed_state_format: bool = False,
) -> MutableMapping[str, list[State | dict[str, Any]]]: ) -> dict[str, list[State | dict[str, Any]]]:
"""Return a dict of significant states during a time period.""" """Return a dict of significant states during a time period."""
if not recorder.get_instance(hass).states_meta_manager.active: if not recorder.get_instance(hass).states_meta_manager.active:
from .legacy import ( # pylint: disable=import-outside-toplevel from .legacy import ( # pylint: disable=import-outside-toplevel
@ -128,7 +127,7 @@ def get_significant_states_with_session(
minimal_response: bool = False, minimal_response: bool = False,
no_attributes: bool = False, no_attributes: bool = False,
compressed_state_format: bool = False, compressed_state_format: bool = False,
) -> MutableMapping[str, list[State | dict[str, Any]]]: ) -> dict[str, list[State | dict[str, Any]]]:
"""Return a dict of significant states during a time period.""" """Return a dict of significant states during a time period."""
if not recorder.get_instance(hass).states_meta_manager.active: if not recorder.get_instance(hass).states_meta_manager.active:
from .legacy import ( # pylint: disable=import-outside-toplevel from .legacy import ( # pylint: disable=import-outside-toplevel
@ -162,7 +161,7 @@ def state_changes_during_period(
descending: bool = False, descending: bool = False,
limit: int | None = None, limit: int | None = None,
include_start_time_state: bool = True, include_start_time_state: bool = True,
) -> MutableMapping[str, list[State]]: ) -> dict[str, list[State]]:
"""Return a list of states that changed during a time period.""" """Return a list of states that changed during a time period."""
if not recorder.get_instance(hass).states_meta_manager.active: if not recorder.get_instance(hass).states_meta_manager.active:
from .legacy import ( # pylint: disable=import-outside-toplevel from .legacy import ( # pylint: disable=import-outside-toplevel

20
homeassistant/components/recorder/history/legacy.py
View File

@ -3,7 +3,7 @@
from __future__ import annotations from __future__ import annotations
from collections import defaultdict from collections import defaultdict
from collections.abc import Callable, Iterable, Iterator, MutableMapping from collections.abc import Callable, Iterable, Iterator
from datetime import datetime from datetime import datetime
from itertools import groupby from itertools import groupby
from operator import attrgetter from operator import attrgetter
@ -209,7 +209,7 @@ def get_significant_states(
minimal_response: bool = False, minimal_response: bool = False,
no_attributes: bool = False, no_attributes: bool = False,
compressed_state_format: bool = False, compressed_state_format: bool = False,
) -> MutableMapping[str, list[State | dict[str, Any]]]: ) -> dict[str, list[State | dict[str, Any]]]:
"""Wrap get_significant_states_with_session with an sql session.""" """Wrap get_significant_states_with_session with an sql session."""
with session_scope(hass=hass, read_only=True) as session: with session_scope(hass=hass, read_only=True) as session:
return get_significant_states_with_session( return get_significant_states_with_session(
@ -317,7 +317,7 @@ def get_significant_states_with_session(
minimal_response: bool = False, minimal_response: bool = False,
no_attributes: bool = False, no_attributes: bool = False,
compressed_state_format: bool = False, compressed_state_format: bool = False,
) -> MutableMapping[str, list[State | dict[str, Any]]]: ) -> dict[str, list[State | dict[str, Any]]]:
"""Return states changes during UTC period start_time - end_time. """Return states changes during UTC period start_time - end_time.
entity_ids is an optional iterable of entities to include in the results. entity_ids is an optional iterable of entities to include in the results.
@ -365,14 +365,14 @@ def get_full_significant_states_with_session(
include_start_time_state: bool = True, include_start_time_state: bool = True,
significant_changes_only: bool = True, significant_changes_only: bool = True,
no_attributes: bool = False, no_attributes: bool = False,
) -> MutableMapping[str, list[State]]: ) -> dict[str, list[State]]:
"""Variant of get_significant_states_with_session. """Variant of get_significant_states_with_session.
Difference with get_significant_states_with_session is that it does not Difference with get_significant_states_with_session is that it does not
return minimal responses. return minimal responses.
""" """
return cast( return cast(
MutableMapping[str, list[State]], dict[str, list[State]],
get_significant_states_with_session( get_significant_states_with_session(
hass=hass, hass=hass,
session=session, session=session,
@ -454,7 +454,7 @@ def state_changes_during_period(
descending: bool = False, descending: bool = False,
limit: int | None = None, limit: int | None = None,
include_start_time_state: bool = True, include_start_time_state: bool = True,
) -> MutableMapping[str, list[State]]: ) -> dict[str, list[State]]:
"""Return states changes during UTC period start_time - end_time.""" """Return states changes during UTC period start_time - end_time."""
if not entity_id: if not entity_id:
raise ValueError("entity_id must be provided") raise ValueError("entity_id must be provided")
@ -471,7 +471,7 @@ def state_changes_during_period(
) )
states = execute_stmt_lambda_element(session, stmt, None, end_time) states = execute_stmt_lambda_element(session, stmt, None, end_time)
return cast( return cast(
MutableMapping[str, list[State]], dict[str, list[State]],
_sorted_states_to_dict( _sorted_states_to_dict(
hass, hass,
session, session,
@ -522,7 +522,7 @@ def _get_last_state_changes_stmt(
def get_last_state_changes( def get_last_state_changes(
hass: HomeAssistant, number_of_states: int, entity_id: str hass: HomeAssistant, number_of_states: int, entity_id: str
) -> MutableMapping[str, list[State]]: ) -> dict[str, list[State]]:
"""Return the last number_of_states.""" """Return the last number_of_states."""
entity_id_lower = entity_id.lower() entity_id_lower = entity_id.lower()
entity_ids = [entity_id_lower] entity_ids = [entity_id_lower]
@ -533,7 +533,7 @@ def get_last_state_changes(
) )
states = list(execute_stmt_lambda_element(session, stmt)) states = list(execute_stmt_lambda_element(session, stmt))
return cast( return cast(
MutableMapping[str, list[State]], dict[str, list[State]],
_sorted_states_to_dict( _sorted_states_to_dict(
hass, hass,
session, session,
@ -693,7 +693,7 @@ def _sorted_states_to_dict(
minimal_response: bool = False, minimal_response: bool = False,
no_attributes: bool = False, no_attributes: bool = False,
compressed_state_format: bool = False, compressed_state_format: bool = False,
) -> MutableMapping[str, list[State | dict[str, Any]]]: ) -> dict[str, list[State | dict[str, Any]]]:
"""Convert SQL results into JSON friendly data structure. """Convert SQL results into JSON friendly data structure.
This takes our state list and turns it into a JSON friendly data This takes our state list and turns it into a JSON friendly data

20
homeassistant/components/recorder/history/modern.py
View File

@ -2,7 +2,7 @@
from __future__ import annotations from __future__ import annotations
from collections.abc import Callable, Iterable, Iterator, MutableMapping from collections.abc import Callable, Iterable, Iterator
from datetime import datetime from datetime import datetime
from itertools import groupby from itertools import groupby
from operator import itemgetter from operator import itemgetter
@ -117,7 +117,7 @@ def get_significant_states(
minimal_response: bool = False, minimal_response: bool = False,
no_attributes: bool = False, no_attributes: bool = False,
compressed_state_format: bool = False, compressed_state_format: bool = False,
) -> MutableMapping[str, list[State | dict[str, Any]]]: ) -> dict[str, list[State | dict[str, Any]]]:
"""Wrap get_significant_states_with_session with an sql session.""" """Wrap get_significant_states_with_session with an sql session."""
with session_scope(hass=hass, read_only=True) as session: with session_scope(hass=hass, read_only=True) as session:
return get_significant_states_with_session( return get_significant_states_with_session(
@ -213,7 +213,7 @@ def get_significant_states_with_session(
minimal_response: bool = False, minimal_response: bool = False,
no_attributes: bool = False, no_attributes: bool = False,
compressed_state_format: bool = False, compressed_state_format: bool = False,
) -> MutableMapping[str, list[State | dict[str, Any]]]: ) -> dict[str, list[State | dict[str, Any]]]:
"""Return states changes during UTC period start_time - end_time. """Return states changes during UTC period start_time - end_time.
entity_ids is an optional iterable of entities to include in the results. entity_ids is an optional iterable of entities to include in the results.
@ -296,14 +296,14 @@ def get_full_significant_states_with_session(
include_start_time_state: bool = True, include_start_time_state: bool = True,
significant_changes_only: bool = True, significant_changes_only: bool = True,
no_attributes: bool = False, no_attributes: bool = False,
) -> MutableMapping[str, list[State]]: ) -> dict[str, list[State]]:
"""Variant of get_significant_states_with_session. """Variant of get_significant_states_with_session.
Difference with get_significant_states_with_session is that it does not Difference with get_significant_states_with_session is that it does not
return minimal responses. return minimal responses.
""" """
return cast( return cast(
MutableMapping[str, list[State]], dict[str, list[State]],
get_significant_states_with_session( get_significant_states_with_session(
hass=hass, hass=hass,
session=session, session=session,
@ -390,7 +390,7 @@ def state_changes_during_period(
descending: bool = False, descending: bool = False,
limit: int | None = None, limit: int | None = None,
include_start_time_state: bool = True, include_start_time_state: bool = True,
) -> MutableMapping[str, list[State]]: ) -> dict[str, list[State]]:
"""Return states changes during UTC period start_time - end_time.""" """Return states changes during UTC period start_time - end_time."""
has_last_reported = ( has_last_reported = (
recorder.get_instance(hass).schema_version >= LAST_REPORTED_SCHEMA_VERSION recorder.get_instance(hass).schema_version >= LAST_REPORTED_SCHEMA_VERSION
@ -438,7 +438,7 @@ def state_changes_during_period(
], ],
) )
return cast( return cast(
MutableMapping[str, list[State]], dict[str, list[State]],
_sorted_states_to_dict( _sorted_states_to_dict(
execute_stmt_lambda_element( execute_stmt_lambda_element(
session, stmt, None, end_time, orm_rows=False session, stmt, None, end_time, orm_rows=False
@ -504,7 +504,7 @@ def _get_last_state_changes_multiple_stmt(
def get_last_state_changes( def get_last_state_changes(
hass: HomeAssistant, number_of_states: int, entity_id: str hass: HomeAssistant, number_of_states: int, entity_id: str
) -> MutableMapping[str, list[State]]: ) -> dict[str, list[State]]:
"""Return the last number_of_states.""" """Return the last number_of_states."""
has_last_reported = ( has_last_reported = (
recorder.get_instance(hass).schema_version >= LAST_REPORTED_SCHEMA_VERSION recorder.get_instance(hass).schema_version >= LAST_REPORTED_SCHEMA_VERSION
@ -539,7 +539,7 @@ def get_last_state_changes(
) )
states = list(execute_stmt_lambda_element(session, stmt, orm_rows=False)) states = list(execute_stmt_lambda_element(session, stmt, orm_rows=False))
return cast( return cast(
MutableMapping[str, list[State]], dict[str, list[State]],
_sorted_states_to_dict( _sorted_states_to_dict(
reversed(states), reversed(states),
None, None,
@ -680,7 +680,7 @@ def _sorted_states_to_dict(
compressed_state_format: bool = False, compressed_state_format: bool = False,
descending: bool = False, descending: bool = False,
no_attributes: bool = False, no_attributes: bool = False,
) -> MutableMapping[str, list[State | dict[str, Any]]]: ) -> dict[str, list[State | dict[str, Any]]]:
"""Convert SQL results into JSON friendly data structure. """Convert SQL results into JSON friendly data structure.
This takes our state list and turns it into a JSON friendly data This takes our state list and turns it into a JSON friendly data

4
homeassistant/components/sensor/recorder.py
View File

@ -3,7 +3,7 @@
from __future__ import annotations from __future__ import annotations
from collections import defaultdict from collections import defaultdict
from collections.abc import Callable, Iterable, MutableMapping from collections.abc import Callable, Iterable
import datetime import datetime
import itertools import itertools
import logging import logging
@ -402,7 +402,7 @@ def compile_statistics( # noqa: C901
entities_full_history = [ entities_full_history = [
i.entity_id for i in sensor_states if "sum" in wanted_statistics[i.entity_id] i.entity_id for i in sensor_states if "sum" in wanted_statistics[i.entity_id]
] ]
history_list: MutableMapping[str, list[State]] = {} history_list: dict[str, list[State]] = {}
if entities_full_history: if entities_full_history:
history_list = history.get_full_significant_states_with_session( history_list = history.get_full_significant_states_with_session(
hass, hass,