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

Simplify generic decorators in recorder (#125301)

Browse source 
* Simplify generic decorators in recorder

* Remove additional case
This commit is contained in:
Erik Montnemery 2024-09-05 10:16:44 +02:00 • committed by GitHub
parent f778033bd8
commit 984eba809c
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 6 additions and 6 deletions
Download patch file Download diff file Expand all files Collapse all files

12
homeassistant/components/recorder/util.py
View file

@ -644,7 +644,7 @@ def _is_retryable_error(instance: Recorder, err: OperationalError) -> bool:
)
type _FuncType[_T, **_P, _R] = Callable[Concatenate[_T, _P], _R]
type _FuncType[**P, R] = Callable[Concatenate[Recorder, P], R]
type _FuncOrMethType[**_P, _R] = Callable[_P, _R]
@ -683,9 +683,9 @@ def retryable_database_job[**_P](
return decorator
def database_job_retry_wrapper[_RecorderT: Recorder, **_P](
def database_job_retry_wrapper[**_P](
description: str, attempts: int = 5
) -> Callable[[_FuncType[_RecorderT, _P, None]], _FuncType[_RecorderT, _P, None]]:
) -> Callable[[_FuncType[_P, None]], _FuncType[_P, None]]:
"""Try to execute a database job multiple times.
This wrapper handles InnoDB deadlocks and lock timeouts.
@ -695,10 +695,10 @@ def database_job_retry_wrapper[_RecorderT: Recorder, **_P](
"""
def decorator(
job: _FuncType[_RecorderT, _P, None],
) -> _FuncType[_RecorderT, _P, None]:
job: _FuncType[_P, None],
) -> _FuncType[_P, None]:
@functools.wraps(job)
def wrapper(instance: _RecorderT, *args: _P.args, **kwargs: _P.kwargs) -> None:
def wrapper(instance: Recorder, *args: _P.args, **kwargs: _P.kwargs) -> None:
for attempt in range(attempts):
try:
job(instance, *args, **kwargs)