chores: updates-and-linters (#1868)

* switch to ruff

* add ruff

* run ruff --fix

* update ruff

* resolve ruff errors

* drop isort from CI

* fix decorator order
This commit is contained in:
Hayden
2022-11-30 20:20:28 -09:00
committed by GitHub
parent fd0e02a5c6
commit 82dc586bac
62 changed files with 362 additions and 536 deletions

View File

@@ -6,22 +6,22 @@ import logging
from asyncio import ensure_future
from functools import wraps
from traceback import format_exception
from typing import Any, Callable, Coroutine, Optional, Union
from typing import Any, Callable, Coroutine
from starlette.concurrency import run_in_threadpool
NoArgsNoReturnFuncT = Callable[[], None]
NoArgsNoReturnAsyncFuncT = Callable[[], Coroutine[Any, Any, None]]
NoArgsNoReturnDecorator = Callable[[Union[NoArgsNoReturnFuncT, NoArgsNoReturnAsyncFuncT]], NoArgsNoReturnAsyncFuncT]
NoArgsNoReturnDecorator = Callable[[NoArgsNoReturnFuncT | NoArgsNoReturnAsyncFuncT], NoArgsNoReturnAsyncFuncT]
def repeat_every(
*,
minutes: float,
wait_first: bool = False,
logger: Optional[logging.Logger] = None,
logger: logging.Logger | None = None,
raise_exceptions: bool = False,
max_repetitions: Optional[int] = None,
max_repetitions: int | None = None,
) -> NoArgsNoReturnDecorator:
"""
This function returns a decorator that modifies a function so it is periodically re-executed after its first call.
@@ -46,7 +46,7 @@ def repeat_every(
The maximum number of times to call the repeated function. If `None`, the function is repeated forever.
"""
def decorator(func: Union[NoArgsNoReturnAsyncFuncT, NoArgsNoReturnFuncT]) -> NoArgsNoReturnAsyncFuncT:
def decorator(func: NoArgsNoReturnAsyncFuncT | NoArgsNoReturnFuncT) -> NoArgsNoReturnAsyncFuncT:
"""
Converts the decorated function into a repeated, periodically-called version of itself.
"""

View File

@@ -1,5 +1,4 @@
from datetime import datetime, timezone
from typing import Optional
from pydantic import UUID4
@@ -18,7 +17,7 @@ from mealie.services.event_bus_service.event_types import (
last_ran = datetime.now(timezone.utc)
def post_group_webhooks(start_dt: Optional[datetime] = None, group_id: Optional[UUID4] = None) -> None:
def post_group_webhooks(start_dt: datetime | None = None, group_id: UUID4 | None = None) -> None:
"""Post webhook events to specified group, or all groups"""
global last_ran