| FazBrowse GitHub Viewer | Trending | | Home |
| Tools: [Download Repo ZIP] [Original HTTPS Page] |
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| Expand Up | @@ -3,7 +3,8 @@ | |
| from __future__ import annotations | ||
|
|
||
| import logging | ||
| from collections.abc import Awaitable, Callable, Coroutine | ||
| from collections.abc import Callable, Coroutine | ||
| from functools import wraps | ||
| from typing import TYPE_CHECKING, Any, Concatenate, ParamSpec, TypeVar | ||
|
|
||
| from ..exceptions import DeviceError, KasaException, SmartErrorCode | ||
| Expand All | @@ -20,15 +21,16 @@ | |
|
|
||
|
|
||
| def allow_update_after( | ||
| func: Callable[Concatenate[_T, _P], Awaitable[dict]], | ||
| ) -> Callable[Concatenate[_T, _P], Coroutine[Any, Any, dict]]: | ||
| func: Callable[Concatenate[_T, _P], Coroutine[Any, Any, _R]], | ||
| ) -> Callable[Concatenate[_T, _P], Coroutine[Any, Any, _R]]: | ||
| """Define a wrapper to set _last_update_time to None. | ||
| This will ensure that a module is updated in the next update cycle after | ||
| a value has been changed. | ||
| """ | ||
|
|
||
| async def _async_wrap(self: _T, *args: _P.args, **kwargs: _P.kwargs) -> dict: | ||
| @wraps(func) | ||
|
Comment thread
Copy link
Copy Markdown
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Choose a reason Spam Abuse Off Topic Outdated Duplicate Resolved Low QualityThis was previously not working when using a decorator which the test cases highlighted. i.e. if using a decorator the type hints were not copied to the decorator function so get_feature would fail.
Sorry, something went wrong.
rytilahti reacted with thumbs up emoji
All reactions
|
||
| async def _async_wrap(self: _T, *args: _P.args, **kwargs: _P.kwargs) -> _R: | ||
| try: | ||
| return await func(self, *args, **kwargs) | ||
| finally: | ||
| Expand All | @@ -40,6 +42,7 @@ async def _async_wrap(self: _T, *args: _P.args, **kwargs: _P.kwargs) -> dict: | |
| def raise_if_update_error(func: Callable[[_T], _R]) -> Callable[[_T], _R]: | ||
| """Define a wrapper to raise an error if the last module update was an error.""" | ||
|
|
||
| @wraps(func) | ||
| def _wrap(self: _T) -> _R: | ||
| if err := self._last_update_error: | ||
| raise err | ||
| Expand Down | ||
| Back | FazBrowse Home | New Git URL |
Uh oh!
There was an error while loading. Please reload this page.