evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Simplify imports in recorder (#126248)

Browse source
This commit is contained in:
epenet 2024-09-20 03:43:21 +02:00 • committed by GitHub
parent 72065768f3
commit bb5640b41b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 15 additions and 15 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
homeassistant/components/recorder/history/__init__.py
View file

@ -8,8 +8,8 @@ from typing import Any
from sqlalchemy.orm.session import Session
from homeassistant.core import HomeAssistant, State
from homeassistant.helpers.recorder import get_instance
from ... import recorder
from ..filters import Filters
from .const import NEED_ATTRIBUTE_DOMAINS, SIGNIFICANT_DOMAINS
from .modern import (
@ -44,7 +44,7 @@ def get_full_significant_states_with_session(
no_attributes: bool = False,
) -> dict[str, list[State]]:
"""Return a dict of significant states during a time period."""
if not recorder.get_instance(hass).states_meta_manager.active:
if not get_instance(hass).states_meta_manager.active:
from .legacy import ( # pylint: disable=import-outside-toplevel
get_full_significant_states_with_session as _legacy_get_full_significant_states_with_session,
)
@ -69,7 +69,7 @@ def get_last_state_changes(
hass: HomeAssistant, number_of_states: int, entity_id: str
) -> dict[str, list[State]]:
"""Return the last number_of_states."""
if not recorder.get_instance(hass).states_meta_manager.active:
if not get_instance(hass).states_meta_manager.active:
from .legacy import ( # pylint: disable=import-outside-toplevel
get_last_state_changes as _legacy_get_last_state_changes,
)
@ -93,7 +93,7 @@ def get_significant_states(
compressed_state_format: bool = False,
) -> dict[str, list[State | dict[str, Any]]]:
"""Return a dict of significant states during a time period."""
if not recorder.get_instance(hass).states_meta_manager.active:
if not get_instance(hass).states_meta_manager.active:
from .legacy import ( # pylint: disable=import-outside-toplevel
get_significant_states as _legacy_get_significant_states,
)
@ -129,7 +129,7 @@ def get_significant_states_with_session(
compressed_state_format: bool = False,
) -> dict[str, list[State | dict[str, Any]]]:
"""Return a dict of significant states during a time period."""
if not recorder.get_instance(hass).states_meta_manager.active:
if not get_instance(hass).states_meta_manager.active:
from .legacy import ( # pylint: disable=import-outside-toplevel
get_significant_states_with_session as _legacy_get_significant_states_with_session,
)
@ -163,7 +163,7 @@ def state_changes_during_period(
include_start_time_state: bool = True,
) -> dict[str, list[State]]:
"""Return a list of states that changed during a time period."""
if not recorder.get_instance(hass).states_meta_manager.active:
if not get_instance(hass).states_meta_manager.active:
from .legacy import ( # pylint: disable=import-outside-toplevel
state_changes_during_period as _legacy_state_changes_during_period,
)

4
homeassistant/components/recorder/history/legacy.py
View file

@ -19,9 +19,9 @@ from sqlalchemy.sql.lambdas import StatementLambdaElement
from homeassistant.const import COMPRESSED_STATE_LAST_UPDATED, COMPRESSED_STATE_STATE
from homeassistant.core import HomeAssistant, State, split_entity_id
from homeassistant.helpers.recorder import get_instance
import homeassistant.util.dt as dt_util
from ... import recorder
from ..db_schema import RecorderRuns, StateAttributes, States
from ..filters import Filters
from ..models import process_timestamp, process_timestamp_to_utc_isoformat
@ -496,7 +496,7 @@ def _get_rows_with_session(
)
if run is None:
run = recorder.get_instance(hass).recorder_runs_manager.get(utc_point_in_time)
run = get_instance(hass).recorder_runs_manager.get(utc_point_in_time)
if run is None or process_timestamp(run.start) > utc_point_in_time:
# History did not run before utc_point_in_time

14
homeassistant/components/recorder/history/modern.py
View file

@ -24,9 +24,9 @@ from sqlalchemy.orm.session import Session
from homeassistant.const import COMPRESSED_STATE_LAST_UPDATED, COMPRESSED_STATE_STATE
from homeassistant.core import HomeAssistant, State, split_entity_id
from homeassistant.helpers.recorder import get_instance
import homeassistant.util.dt as dt_util
from ... import recorder
from ..const import LAST_REPORTED_SCHEMA_VERSION
from ..db_schema import SHARED_ATTR_OR_LEGACY_ATTRIBUTES, StateAttributes, States
from ..filters import Filters
@ -231,7 +231,7 @@ def get_significant_states_with_session(
raise ValueError("entity_ids must be provided")
entity_id_to_metadata_id: dict[str, int | None] | None = None
metadata_ids_in_significant_domains: list[int] = []
instance = recorder.get_instance(hass)
instance = get_instance(hass)
if not (
entity_id_to_metadata_id := instance.states_meta_manager.get_many(
entity_ids, session, False
@ -393,14 +393,14 @@ def state_changes_during_period(
) -> dict[str, list[State]]:
"""Return states changes during UTC period start_time - end_time."""
has_last_reported = (
recorder.get_instance(hass).schema_version >= LAST_REPORTED_SCHEMA_VERSION
get_instance(hass).schema_version >= LAST_REPORTED_SCHEMA_VERSION
)
if not entity_id:
raise ValueError("entity_id must be provided")
entity_ids = [entity_id.lower()]
with session_scope(hass=hass, read_only=True) as session:
instance = recorder.get_instance(hass)
instance = get_instance(hass)
if not (
possible_metadata_id := instance.states_meta_manager.get(
entity_id, session, False
@ -507,7 +507,7 @@ def get_last_state_changes(
) -> dict[str, list[State]]:
"""Return the last number_of_states."""
has_last_reported = (
recorder.get_instance(hass).schema_version >= LAST_REPORTED_SCHEMA_VERSION
get_instance(hass).schema_version >= LAST_REPORTED_SCHEMA_VERSION
)
entity_id_lower = entity_id.lower()
entity_ids = [entity_id_lower]
@ -517,7 +517,7 @@ def get_last_state_changes(
# because the metadata_id_last_updated_ts index is in ascending order.
with session_scope(hass=hass, read_only=True) as session:
instance = recorder.get_instance(hass)
instance = get_instance(hass)
if not (
possible_metadata_id := instance.states_meta_manager.get(
entity_id, session, False
@ -604,7 +604,7 @@ def _get_run_start_ts_for_utc_point_in_time(
hass: HomeAssistant, utc_point_in_time: datetime
) -> float | None:
"""Return the start time of a run."""
run = recorder.get_instance(hass).recorder_runs_manager.get(utc_point_in_time)
run = get_instance(hass).recorder_runs_manager.get(utc_point_in_time)
if (
run is not None
and (run_start := process_timestamp(run.start)) < utc_point_in_time