fix: ruff "unsafe" fixes
Signed-off-by: Akhil Narang <me@akhilnarang.dev>
This commit is contained in:
parent
8723a2b6ee
commit
f7feeea0a0
8 changed files with 27 additions and 36 deletions
|
|
@ -88,17 +88,17 @@ if _dev_server:
|
|||
|
||||
|
||||
# local-globals
|
||||
ConfType: TypeAlias = _dict[str, Any] # type: ignore[no-any-explicit]
|
||||
type ConfType = _dict[str, Any] # type: ignore[no-any-explicit]
|
||||
# TODO: make session a dataclass instead of undtyped _dict
|
||||
SessionType: TypeAlias = _dict[str, Any] # type: ignore[no-any-explicit]
|
||||
type SessionType = _dict[str, Any] # type: ignore[no-any-explicit]
|
||||
# TODO: implement dataclass
|
||||
LogMessageType: TypeAlias = _dict[str, Any] # type: ignore[no-any-explicit]
|
||||
type LogMessageType = _dict[str, Any] # type: ignore[no-any-explicit]
|
||||
# TODO: implement dataclass
|
||||
# holds job metadata if the code is run in a background job context
|
||||
JobMetaType: TypeAlias = _dict[str, Any] # type: ignore[no-any-explicit]
|
||||
ResponseDict: TypeAlias = _dict[str, Any] # type: ignore[no-any-explicit]
|
||||
FlagsDict: TypeAlias = _dict[str, Any] # type: ignore[no-any-explicit]
|
||||
FormDict: TypeAlias = _dict[str, str]
|
||||
type JobMetaType = _dict[str, Any] # type: ignore[no-any-explicit]
|
||||
type ResponseDict = _dict[str, Any] # type: ignore[no-any-explicit]
|
||||
type FlagsDict = _dict[str, Any] # type: ignore[no-any-explicit]
|
||||
type FormDict = _dict[str, str]
|
||||
|
||||
db: LocalProxy["PyMariaDBDatabase" | "MariaDBDatabase" | "PostgresDatabase" | "SQLiteDatabase"] = local("db")
|
||||
qb: LocalProxy["MariaDB" | "Postgres" | "SQLite"] = local("qb")
|
||||
|
|
|
|||
|
|
@ -180,8 +180,7 @@ class TestRQJob(IntegrationTestCase):
|
|||
# Observed higher usage on 3.14. Temporarily raising the limit
|
||||
from sys import version_info
|
||||
|
||||
if version_info >= (3, 14):
|
||||
LAST_MEASURED_USAGE += 5
|
||||
LAST_MEASURED_USAGE += 5
|
||||
|
||||
self.assertLessEqual(rss, LAST_MEASURED_USAGE * 1.05, msg)
|
||||
|
||||
|
|
|
|||
|
|
@ -285,7 +285,7 @@ def get_events(
|
|||
start: date, end: date, user: str | None = None, for_reminder: bool = False, filters=None
|
||||
) -> list[frappe._dict]:
|
||||
user = user or frappe.session.user
|
||||
EventLikeDict: TypeAlias = Event | frappe._dict
|
||||
type EventLikeDict = Event | frappe._dict
|
||||
resolved_events: list[EventLikeDict] = []
|
||||
|
||||
if isinstance(filters, str):
|
||||
|
|
|
|||
|
|
@ -42,8 +42,8 @@ DOCUMENT_LOCK_EXPIRY = 3 * 60 * 60 # All locks expire in 3 hours automatically
|
|||
DOCUMENT_LOCK_SOFT_EXPIRY = 30 * 60 # Let users force-unlock after 30 minutes
|
||||
|
||||
|
||||
_SingleDocument: TypeAlias = "Document"
|
||||
_NewDocument: TypeAlias = "Document"
|
||||
type _SingleDocument = "Document"
|
||||
type _NewDocument = "Document"
|
||||
|
||||
|
||||
@overload
|
||||
|
|
|
|||
|
|
@ -8,20 +8,20 @@ from typing import Any, NamedTuple, Self, TypeAlias, TypeGuard, TypeVar, cast, o
|
|||
|
||||
from pypika import Column
|
||||
|
||||
Doct: TypeAlias = str
|
||||
Fld: TypeAlias = str
|
||||
Op: TypeAlias = str
|
||||
DateTime: TypeAlias = datetime | date
|
||||
_Value: TypeAlias = str | int | float | None | DateTime | Column
|
||||
_InputValue: TypeAlias = _Value | bool
|
||||
Value: TypeAlias = _Value | Sequence[_Value]
|
||||
InputValue: TypeAlias = _InputValue | Sequence[_InputValue]
|
||||
type Doct = str
|
||||
type Fld = str
|
||||
type Op = str
|
||||
type DateTime = datetime | date
|
||||
type _Value = str | int | float | None | DateTime | Column
|
||||
type _InputValue = _Value | bool
|
||||
type Value = _Value | Sequence[_Value]
|
||||
type InputValue = _InputValue | Sequence[_InputValue]
|
||||
|
||||
|
||||
FilterTupleSpec: TypeAlias = (
|
||||
type FilterTupleSpec = (
|
||||
tuple[Fld, InputValue] | tuple[Fld, Op, InputValue] | tuple[Doct, Fld, Op, InputValue]
|
||||
)
|
||||
FilterMappingSpec: TypeAlias = Mapping[Fld, _InputValue | tuple[Op, InputValue]]
|
||||
type FilterMappingSpec = Mapping[Fld, _InputValue | tuple[Op, InputValue]]
|
||||
|
||||
|
||||
class Sentinel:
|
||||
|
|
@ -38,7 +38,7 @@ UNSPECIFIED = Sentinel()
|
|||
T = TypeVar("T")
|
||||
|
||||
|
||||
def is_unspecified(value: T | Sentinel) -> TypeGuard[Sentinel]:
|
||||
def is_unspecified[T](value: T | Sentinel) -> TypeGuard[Sentinel]:
|
||||
return value is UNSPECIFIED
|
||||
|
||||
|
||||
|
|
@ -280,4 +280,4 @@ class Filters(list[FilterTuple]):
|
|||
return f"Filters(\n{filters_str}\n)"
|
||||
|
||||
|
||||
FilterSignature: TypeAlias = Filters | FilterTuple | FilterMappingSpec | FilterTupleSpec
|
||||
type FilterSignature = Filters | FilterTuple | FilterMappingSpec | FilterTupleSpec
|
||||
|
|
|
|||
|
|
@ -46,16 +46,9 @@ EMAIL_MATCH_PATTERN = re.compile(
|
|||
|
||||
UNSET = object()
|
||||
|
||||
PropertyType: TypeAlias = property | functools.cached_property
|
||||
type PropertyType = property | functools.cached_property
|
||||
|
||||
|
||||
if sys.version_info < (3, 11):
|
||||
|
||||
def exception():
|
||||
_exc_type, exc_value, _exc_traceback = sys.exc_info()
|
||||
return exc_value
|
||||
|
||||
sys.exception = exception
|
||||
|
||||
|
||||
def get_fullname(user=None):
|
||||
|
|
|
|||
|
|
@ -451,10 +451,9 @@ def start_worker_pool(
|
|||
if sbool(os.environ.get("FRAPPE_BACKGROUND_WORKERS_NOFORK", False)):
|
||||
worker_klass = FrappeWorkerNoFork
|
||||
else:
|
||||
if sys.version_info >= (3, 14):
|
||||
import multiprocessing
|
||||
import multiprocessing
|
||||
|
||||
multiprocessing.set_start_method("fork", force=True)
|
||||
multiprocessing.set_start_method("fork", force=True)
|
||||
worker_klass = FrappeWorker
|
||||
|
||||
pool = WorkerPool(
|
||||
|
|
|
|||
|
|
@ -59,7 +59,7 @@ class Local:
|
|||
return lp
|
||||
|
||||
|
||||
class LocalProxy(WerkzeugLocalProxy, Generic[T]):
|
||||
class LocalProxy[T](WerkzeugLocalProxy):
|
||||
__slots__ = ()
|
||||
|
||||
def __getattr__(self, name: str) -> Any:
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue