Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
131 changes: 18 additions & 113 deletions edb/common/checked.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,15 +28,13 @@
MutableSequence,
MutableSet,
Sequence,
cast,
overload,
)

import collections.abc
import itertools
import types

from edb.common import debug
from edb.common import parametric


Expand Down Expand Up @@ -78,16 +76,6 @@ class AbstractCheckedList[T]:
type: type
_container: list[T]

@classmethod
def _check_type(cls, value: Any) -> T:
"""Ensure `value` is of type T and return it."""
if not isinstance(value, cls.type):
raise ValueError(
f"{cls!r} accepts only values of type {cls.type!r}, "
f"got {type(value)!r}"
)
return cast(T, value)

def __init__(self, iterable: Iterable[T] = ()) -> None:
pass

Expand Down Expand Up @@ -129,7 +117,7 @@ class FrozenCheckedList[T](
):
def __init__(self, iterable: Iterable[T] = ()) -> None:
super().__init__()
self._container = [self._check_type(element) for element in iterable]
self._container = [element for element in iterable]
self._hash_cache = -1

def __hash__(self) -> int:
Expand Down Expand Up @@ -180,7 +168,7 @@ class CheckedList[T](
):
def __init__(self, iterable: Iterable[T] = ()) -> None:
super().__init__()
self._container = [self._check_type(element) for element in iterable]
self._container = [element for element in iterable]

#
# Sequence
Expand Down Expand Up @@ -210,11 +198,11 @@ def __setitem__(self, index: slice, value: Iterable[T]) -> None: ...

def __setitem__(self, index: int | slice, value: Any) -> None:
if isinstance(index, int):
self._container[index] = self._check_type(value)
self._container[index] = value
return

_slice = index
self._container[_slice] = filter(self._check_type, value)
self._container[_slice] = value

@overload
def __delitem__(self, index: int) -> None: ...
Expand All @@ -226,7 +214,7 @@ def __delitem__(self, index: int | slice) -> None:
del self._container[index]

def insert(self, index: int, value: T) -> None:
self._container.insert(index, self._check_type(value))
self._container.insert(index, value)

def __len__(self) -> int:
return len(self._container)
Expand All @@ -242,7 +230,7 @@ def __radd__(self, other: Iterable[T]) -> CheckedList[T]:
return self.__class__(itertools.chain(other, self))

def __iadd__(self, other: Iterable[T]) -> CheckedList[T]:
self._container.extend(filter(self._check_type, other))
self._container.extend(other)
return self

def __mul__(self, n: int) -> CheckedList[T]:
Expand All @@ -265,16 +253,6 @@ class AbstractCheckedSet[T](AbstractSet[T]):
def __init__(self, iterable: Iterable[T] = ()) -> None:
pass

@classmethod
def _check_type(cls, value: Any) -> T:
"""Ensure `value` is of type T and return it."""
if not isinstance(value, cls.type):
raise ValueError(
f"{cls!r} accepts only values of type {cls.type!r}, "
f"got {type(value)!r}"
)
return cast(T, value)

def _cast(self, other: Any) -> AbstractSet[T]:
if isinstance(other, (FrozenCheckedSet, CheckedSet)):
return other._container
Expand Down Expand Up @@ -326,7 +304,7 @@ class FrozenCheckedSet[T](
):
def __init__(self, iterable: Iterable[T] = ()) -> None:
super().__init__()
self._container = {self._check_type(element) for element in iterable}
self._container = {element for element in iterable}
self._hash_cache = -1

def __hash__(self) -> int:
Expand All @@ -347,10 +325,6 @@ def __hash__(self) -> int:

def __and__(self, other: AbstractSet[T]) -> FrozenCheckedSet[T]:
other_set = self._cast(other)
for elem in other_set:
# We need the explicit type check to reject nonsensical
# & operations that must always result in an empty new set.
self._check_type(elem)
return self.__class__(other_set & self._container)

__rand__ = __and__
Expand All @@ -365,10 +339,6 @@ def __or__( # type: ignore

def __sub__(self, other: AbstractSet[T]) -> FrozenCheckedSet[T]:
other_set = self._cast(other)
for elem in other_set:
# We need the explicit type check to reject nonsensical
# - operations that always return the original checked set.
self._check_type(elem)
return self.__class__(self._container - other_set)

def __rsub__(self, other: AbstractSet[T]) -> FrozenCheckedSet[T]:
Expand Down Expand Up @@ -403,7 +373,7 @@ class CheckedSet[T](

def __init__(self, iterable: Iterable[T] = ()) -> None:
super().__init__()
self._container = {self._check_type(element) for element in iterable}
self._container = {element for element in iterable}

#
# Replaced mixins of collections.abc.Set
Expand All @@ -418,10 +388,6 @@ def __init__(self, iterable: Iterable[T] = ()) -> None:

def __and__(self, other: AbstractSet[T]) -> CheckedSet[T]:
other_set = self._cast(other)
for elem in other_set:
# We need the explicit type check to reject nonsensical
# & operations that must always result in an empty new set.
self._check_type(elem)
return self.__class__(other_set & self._container)

__rand__ = __and__
Expand All @@ -434,10 +400,6 @@ def __or__(self, other: AbstractSet[T]) -> CheckedSet[T]: # type: ignore

def __sub__(self, other: AbstractSet[T]) -> CheckedSet[T]:
other_set = self._cast(other)
for elem in other_set:
# We need the explicit type check to reject nonsensical
# - operations that always return the original checked set.
self._check_type(elem)
return self.__class__(self._container - other_set)

def __rsub__(self, other: AbstractSet[T]) -> CheckedSet[T]:
Expand All @@ -455,33 +417,29 @@ def __xor__(self, other: AbstractSet[T]) -> CheckedSet[T]: # type: ignore
#

def add(self, value: T) -> None:
self._container.add(self._check_type(value))
self._container.add(value)

def discard(self, value: T) -> None:
self._container.discard(self._check_type(value))
self._container.discard(value)

#
# Replaced mixins of collections.abc.MutableSet
#

def __ior__(self, other: AbstractSet[T]) -> CheckedSet[T]: # type: ignore
self._container |= set(filter(self._check_type, other))
self._container |= set(other)
return self

def __iand__(self, other: AbstractSet[T]) -> CheckedSet[T]:
# We do the type check here to reject nonsensical
# & operations that always clear the checked set.
self._container &= set(filter(self._check_type, other))
self._container &= set(other)
return self

def __ixor__(self, other: AbstractSet[T]) -> CheckedSet[T]: # type: ignore
self._container ^= set(filter(self._check_type, other))
self._container ^= set(other)
return self

def __isub__(self, other: AbstractSet[T]) -> CheckedSet[T]:
# We do the type check here to reject nonsensical
# - operations that could never affect the checked set.
self._container -= set(filter(self._check_type, other))
self._container -= set(other)
return self

#
Expand Down Expand Up @@ -518,26 +476,6 @@ class AbstractCheckedDict[K, V]:
valuetype: type
_container: dict[K, V]

@classmethod
def _check_key_type(cls, key: Any) -> K:
"""Ensure `key` is of type K and return it."""
if not isinstance(key, cls.keytype):
raise KeyError(
f"{cls!r} accepts only keys of type {cls.keytype!r}, "
f"got {type(key)!r}"
)
return cast(K, key)

@classmethod
def _check_value_type(cls, value: Any) -> V:
"""Ensure `value` is of type V and return it."""
if not isinstance(value, cls.valuetype):
raise ValueError(
f"{cls!r} accepts only values of type "
"{cls.valuetype!r}, got {type(value)!r}"
)
return cast(V, value)

def __eq__(self, other: object) -> bool:
if isinstance(other, CheckedDict):
other = other._container
Expand Down Expand Up @@ -590,8 +528,7 @@ def __len__(self) -> int:
#

def __setitem__(self, key: K, value: V) -> None:
self._check_key_type(key)
self._container[key] = self._check_value_type(value)
self._container[key] = value

def __delitem__(self, key: K) -> None:
del self._container[key]
Expand All @@ -603,40 +540,8 @@ def __delitem__(self, key: K) -> None:
@classmethod
def fromkeys(
cls, iterable: Iterable[K], value: Optional[V] = None
) -> CheckedDict[K, V]:
new: CheckedDict[K, V] = cls()
) -> CheckedDict[K, V | None]:
new: CheckedDict[K, V | None] = cls() # type: ignore
for key in iterable:
new[cls._check_key_type(key)] = cls._check_value_type(value)
new[key] = value
return new


def _identity[T](cls: type, value: T) -> T:
return value


_type_checking = {
CheckedList: ["_check_type"],
CheckedDict: ["_check_key_type", "_check_value_type"],
CheckedSet: ["_check_type"],
FrozenCheckedList: ["_check_type"],
FrozenCheckedSet: ["_check_type"],
}


def disable_typechecks() -> None:
for type_, methods in _type_checking.items():
for method in methods:
setattr(type_, method, _identity)


def enable_typechecks() -> None:
for type_, methods in _type_checking.items():
for method in methods:
try:
delattr(type_, method)
except AttributeError:
continue


if not debug.flags.typecheck:
disable_typechecks()
1 change: 0 additions & 1 deletion edb/common/ordered.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,6 @@
MutableSet,
)

import collections
import collections.abc


Expand Down
1 change: 0 additions & 1 deletion edb/language_server/definition.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@
from typing import Optional

from lsprotocol import types as lsp_types
import pygls
import pygls.workspace

from edb.common import span as edb_span
Expand Down
3 changes: 1 addition & 2 deletions edb/language_server/schema.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@
import os

from pygls import uris as pygls_uris
import pygls
import pygls.workspace
from lsprotocol import types as lsp_types


Expand All @@ -32,7 +32,6 @@
from edb.schema import schema as s_schema
from edb.schema import std as s_std
from edb.schema import ddl as s_ddl
import pygls.workspace

from . import parsing as ls_parsing
from . import is_schema_file
Expand Down
2 changes: 1 addition & 1 deletion edb/schema/expr.py
Original file line number Diff line number Diff line change
Expand Up @@ -448,7 +448,7 @@ def __repr__(self) -> str:
return f'<ExpressionShell {self.text} refs=({refs})>'


class ExpressionList(checked.FrozenCheckedList[Expression]):
class ExpressionList(list[Expression]):

@staticmethod
def merge_values(
Expand Down
3 changes: 1 addition & 2 deletions edb/schema/types.py
Original file line number Diff line number Diff line change
Expand Up @@ -1262,7 +1262,6 @@ def as_type_delete_if_unused(
)


Dimensions = checked.FrozenCheckedList[int]
Array_T = typing.TypeVar("Array_T", bound="Array")
Array_T_co = typing.TypeVar("Array_T_co", bound="Array", covariant=True)

Expand Down Expand Up @@ -1332,7 +1331,7 @@ class Array(
)

dimensions = so.SchemaField(
Dimensions,
checked.FrozenCheckedList[int],
coerce=True,
# We want a low compcoef so that array types are *never* altered.
compcoef=0,
Expand Down
2 changes: 0 additions & 2 deletions edb/tools/test/runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,6 @@
import json
import multiprocessing
import multiprocessing.reduction
import multiprocessing.util
import os
import pathlib
import random
Expand All @@ -46,7 +45,6 @@
import types
import unittest.case
import unittest.result
import unittest.runner
import unittest.signals
import warnings

Expand Down
Loading