From bce59e43562e50ee561a5d699001e6b829111861 Mon Sep 17 00:00:00 2001 From: Abram Booth Date: Fri, 3 May 2024 11:08:18 -0400 Subject: [PATCH] tighten addon_toolkit types --- addon_imps/storage/box_dot_com.py | 2 +- addon_service/common/network.py | 2 +- addon_toolkit/constrained_network/http.py | 22 ++++++++-------- addon_toolkit/credentials.py | 7 +++-- addon_toolkit/cursor.py | 24 +++++++++++------- addon_toolkit/declarator.py | 31 +++++++++++++---------- addon_toolkit/imp.py | 25 ++++++++---------- addon_toolkit/iri_utils.py | 6 ++--- addon_toolkit/json_arguments.py | 28 ++++++++++++++------ addon_toolkit/operation.py | 4 +-- addon_toolkit/tests/_doctest.py | 29 ++++++++++++++++++--- addon_toolkit/typing.py | 8 ++++++ mypy.ini | 6 ++--- 13 files changed, 120 insertions(+), 74 deletions(-) create mode 100644 addon_toolkit/typing.py diff --git a/addon_imps/storage/box_dot_com.py b/addon_imps/storage/box_dot_com.py index abf788d6..0b5a9e8b 100644 --- a/addon_imps/storage/box_dot_com.py +++ b/addon_imps/storage/box_dot_com.py @@ -61,7 +61,7 @@ def _params_from_cursor(self, cursor: str = "") -> dict[str, str]: # https://developer.box.com/guides/api-calls/pagination/offset-based/ try: _cursor = OffsetCursor.from_str(cursor) - return {"offset": _cursor.offset, "limit": _cursor.limit} + return {"offset": str(_cursor.offset), "limit": str(_cursor.limit)} except ValueError: return {} diff --git a/addon_service/common/network.py b/addon_service/common/network.py index a804eeed..4160d03c 100644 --- a/addon_service/common/network.py +++ b/addon_service/common/network.py @@ -66,7 +66,7 @@ def __init__( # abstract method from HttpRequestor: @contextlib.asynccontextmanager - async def do_send(self, request: HttpRequestInfo): + async def send_request(self, request: HttpRequestInfo): try: async with self._try_send(request) as _response: yield _response diff --git a/addon_toolkit/constrained_network/http.py b/addon_toolkit/constrained_network/http.py index 1ad34ffd..282d404d 100644 --- a/addon_toolkit/constrained_network/http.py +++ b/addon_toolkit/constrained_network/http.py @@ -61,25 +61,25 @@ class HttpRequestor(typing.Protocol): def response_info_cls(self) -> type[HttpResponseInfo]: ... # abstract method for subclasses - def do_send( + def send_request( self, request: HttpRequestInfo ) -> contextlib.AbstractAsyncContextManager[HttpResponseInfo]: ... @contextlib.asynccontextmanager - async def request( + async def _request( self, http_method: HTTPMethod, uri_path: str, query: Multidict | KeyValuePairs | None = None, headers: Multidict | KeyValuePairs | None = None, - ): + ) -> typing.Any: # loose type; method-specific methods below are more accurate _request_info = HttpRequestInfo( http_method=http_method, uri_path=uri_path, query=(query if isinstance(query, Multidict) else Multidict(query)), headers=(headers if isinstance(headers, Multidict) else Multidict(headers)), ) - async with self.do_send(_request_info) as _response: + async with self.send_request(_request_info) as _response: yield _response # TODO: streaming send/receive (only if/when needed) @@ -88,10 +88,10 @@ async def request( # convenience methods for http methods # (same call signature as self.request, minus `http_method`) - OPTIONS: _MethodRequestMethod = partialmethod(request, HTTPMethod.OPTIONS) - HEAD: _MethodRequestMethod = partialmethod(request, HTTPMethod.HEAD) - GET: _MethodRequestMethod = partialmethod(request, HTTPMethod.GET) - PATCH: _MethodRequestMethod = partialmethod(request, HTTPMethod.PATCH) - POST: _MethodRequestMethod = partialmethod(request, HTTPMethod.POST) - PUT: _MethodRequestMethod = partialmethod(request, HTTPMethod.PUT) - DELETE: _MethodRequestMethod = partialmethod(request, HTTPMethod.DELETE) + OPTIONS: _MethodRequestMethod = partialmethod(_request, HTTPMethod.OPTIONS) + HEAD: _MethodRequestMethod = partialmethod(_request, HTTPMethod.HEAD) + GET: _MethodRequestMethod = partialmethod(_request, HTTPMethod.GET) + PATCH: _MethodRequestMethod = partialmethod(_request, HTTPMethod.PATCH) + POST: _MethodRequestMethod = partialmethod(_request, HTTPMethod.POST) + PUT: _MethodRequestMethod = partialmethod(_request, HTTPMethod.PUT) + DELETE: _MethodRequestMethod = partialmethod(_request, HTTPMethod.DELETE) diff --git a/addon_toolkit/credentials.py b/addon_toolkit/credentials.py index 913d804c..bce71036 100644 --- a/addon_toolkit/credentials.py +++ b/addon_toolkit/credentials.py @@ -4,19 +4,18 @@ @dataclasses.dataclass(frozen=True) class Credentials(typing.Protocol): - def asdict(self): + def asdict(self) -> dict[str, typing.Any]: return dataclasses.asdict(self) def iter_headers(self) -> typing.Iterator[tuple[str, str]]: - return - yield + yield from () # no headers unless implemented by subclass @dataclasses.dataclass(frozen=True, kw_only=True) class AccessTokenCredentials(Credentials): access_token: str - def iter_headers(self): + def iter_headers(self) -> typing.Iterator[tuple[str, str]]: yield ("Authorization", f"Bearer {self.access_token}") diff --git a/addon_toolkit/cursor.py b/addon_toolkit/cursor.py index d92c09e9..6bc3f7a0 100644 --- a/addon_toolkit/cursor.py +++ b/addon_toolkit/cursor.py @@ -1,26 +1,32 @@ import base64 import dataclasses import json -from typing import ( - ClassVar, - Protocol, -) +import typing -def encode_cursor_dataclass(dataclass_instance) -> str: +class DataclassInstance(typing.Protocol): + __dataclass_fields__: typing.ClassVar[dict[str, typing.Any]] + + +SomeDataclassInstance = typing.TypeVar("SomeDataclassInstance", bound=DataclassInstance) + + +def encode_cursor_dataclass(dataclass_instance: DataclassInstance) -> str: _as_json = json.dumps(dataclasses.astuple(dataclass_instance)) _cursor_bytes = base64.b64encode(_as_json.encode()) return _cursor_bytes.decode() -def decode_cursor_dataclass(cursor: str, dataclass_class): +def decode_cursor_dataclass( + cursor: str, dataclass_class: type[SomeDataclassInstance] +) -> SomeDataclassInstance: _as_list = json.loads(base64.b64decode(cursor)) return dataclass_class(*_as_list) -class Cursor(Protocol): +class Cursor(DataclassInstance, typing.Protocol): @classmethod - def from_str(cls, cursor: str): + def from_str(cls, cursor: str) -> typing.Self: return decode_cursor_dataclass(cursor, cls) @property @@ -52,7 +58,7 @@ class OffsetCursor(Cursor): limit: int total_count: int # use -1 to mean "many more" - MAX_INDEX: ClassVar[int] = 9999 + MAX_INDEX: typing.ClassVar[int] = 9999 @property def next_cursor_str(self) -> str | None: diff --git a/addon_toolkit/declarator.py b/addon_toolkit/declarator.py index d3eca8a2..c6929e6d 100644 --- a/addon_toolkit/declarator.py +++ b/addon_toolkit/declarator.py @@ -7,13 +7,14 @@ TypeVar, ) +from addon_toolkit.typing import DataclassInstance + DecoratorTarget = TypeVar("DecoratorTarget") -DeclarationDataclass = TypeVar("DeclarationDataclass") @dataclasses.dataclass -class Declarator(Generic[DeclarationDataclass]): +class Declarator(Generic[DataclassInstance]): """Declarator: add declarative metadata in python using decorators and dataclasses define a dataclass with fields you want declared in your decorator, plus a field @@ -48,15 +49,15 @@ class Declarator(Generic[DeclarationDataclass]): TwoPartGreetingDeclaration(a='kia', b='ora', on=) """ - declaration_dataclass: type[DeclarationDataclass] + declaration_dataclass: type[DataclassInstance] field_for_target: str static_kwargs: dict[str, Any] | None = None # private storage linking a decorated class or function to data gleaned from its decorator - __declarations_by_target: weakref.WeakKeyDictionary[ - object, DeclarationDataclass - ] = dataclasses.field( - default_factory=weakref.WeakKeyDictionary, + __declarations_by_target: weakref.WeakKeyDictionary[object, DataclassInstance] = ( + dataclasses.field( + default_factory=weakref.WeakKeyDictionary, + ) ) def __post_init__(self) -> None: @@ -69,7 +70,7 @@ def __post_init__(self) -> None: ), f'expected field "{self.field_for_target}" on dataclass "{self.declaration_dataclass}"' def __call__( - self, **declaration_dataclass_kwargs + self, **declaration_dataclass_kwargs: Any ) -> Callable[[DecoratorTarget], DecoratorTarget]: """for using a Declarator as a decorator""" @@ -79,13 +80,13 @@ def _decorator(decorator_target: DecoratorTarget) -> DecoratorTarget: return _decorator - def with_kwargs(self, **static_kwargs) -> "Declarator[DeclarationDataclass]": + def with_kwargs(self, **static_kwargs: Any) -> "Declarator[DataclassInstance]": """convenience for decorators that differ only by static field values""" # note: shared __declarations_by_target return dataclasses.replace(self, static_kwargs=static_kwargs) def set_declaration( - self, declaration_target: DecoratorTarget, **declaration_dataclass_kwargs + self, declaration_target: DecoratorTarget, **declaration_dataclass_kwargs: Any ) -> None: """create a declaration associated with the target @@ -98,14 +99,14 @@ def set_declaration( **{self.field_for_target: declaration_target}, ) - def get_declaration(self, target) -> DeclarationDataclass: + def get_declaration(self, target: DecoratorTarget) -> DataclassInstance: try: return self.__declarations_by_target[target] except KeyError: raise ValueError(f"no declaration found for {target}") -class ClassDeclarator(Declarator[DeclarationDataclass]): +class ClassDeclarator(Declarator[DataclassInstance]): """add declarative metadata to python classes using decorators (same as Declarator but with additional methods that only make @@ -157,13 +158,15 @@ class ClassDeclarator(Declarator[DeclarationDataclass]): SemanticVersionDeclaration(major=4, minor=2, patch=9, subj=) """ - def get_declaration_for_class_or_instance(self, type_or_object: type | object): + def get_declaration_for_class_or_instance( + self, type_or_object: type | object + ) -> DataclassInstance: _cls = ( type_or_object if isinstance(type_or_object, type) else type(type_or_object) ) return self.get_declaration_for_class(_cls) - def get_declaration_for_class(self, cls: type): + def get_declaration_for_class(self, cls: type) -> DataclassInstance: for _cls in cls.__mro__: try: return self.get_declaration(_cls) diff --git a/addon_toolkit/imp.py b/addon_toolkit/imp.py index c2023dd2..ea4f47eb 100644 --- a/addon_toolkit/imp.py +++ b/addon_toolkit/imp.py @@ -1,10 +1,7 @@ import dataclasses import enum import inspect -from typing import ( - Iterable, - Iterator, -) +import typing from asgiref.sync import ( async_to_sync, @@ -37,7 +34,7 @@ class AddonImp: imp_number: int addon_protocol: AddonProtocolDeclaration = dataclasses.field(init=False) - def __post_init__(self, addon_protocol_cls): + def __post_init__(self, addon_protocol_cls: type) -> None: object.__setattr__( # using __setattr__ to bypass dataclass frozenness self, "addon_protocol", @@ -45,8 +42,8 @@ def __post_init__(self, addon_protocol_cls): ) def get_operation_imps( - self, *, capabilities: Iterable[enum.Enum] = () - ) -> Iterator["AddonOperationImp"]: + self, *, capabilities: typing.Iterable[enum.Enum] = () + ) -> typing.Iterator["AddonOperationImp"]: for _declaration in self.addon_protocol.get_operation_declarations( capabilities=capabilities ): @@ -74,13 +71,13 @@ class AddonOperationImp: addon_imp: AddonImp declaration: AddonOperationDeclaration - def __post_init__(self): + def __post_init__(self) -> None: _protocol_fn = getattr( self.addon_imp.addon_protocol.protocol_cls, self.declaration.name ) try: _imp_fn = self.imp_function - except AttributeError: + except Exception: _imp_fn = _protocol_fn if _imp_fn is _protocol_fn: raise NotImplementedError( # TODO: helpful exception type @@ -88,12 +85,12 @@ def __post_init__(self): ) @property - def imp_function(self): + def imp_function(self) -> typing.Any: # TODO: less typing.Any return getattr(self.addon_imp.imp_cls, self.declaration.name) async def invoke_thru_addon( self, addon_instance: object, json_kwargs: JsonableDict - ): + ) -> typing.Any: # TODO: less typing.Any _method = self._get_instance_method(addon_instance) _kwargs = kwargs_from_json(self.declaration.call_signature, json_kwargs) if not inspect.iscoroutinefunction(_method): @@ -104,7 +101,7 @@ async def invoke_thru_addon( invoke_thru_addon__blocking = async_to_sync(invoke_thru_addon) - def _get_instance_method(self, addon_instance: object): + def _get_instance_method( + self, addon_instance: object + ) -> typing.Any: # TODO: less typing.Any return getattr(addon_instance, self.declaration.name) - - # TODO: async def async_call_with_json_kwargs(self, addon_instance: object, json_kwargs: dict): diff --git a/addon_toolkit/iri_utils.py b/addon_toolkit/iri_utils.py index c8b9230f..2250a3a3 100644 --- a/addon_toolkit/iri_utils.py +++ b/addon_toolkit/iri_utils.py @@ -63,14 +63,14 @@ def __init__(self, key_value_pairs: KeyValuePairs | None = None): _headerslist = list(key_value_pairs) super().__init__(_headerslist) - def add(self, key: str, value: str, **mediatype_params): + def add(self, key: str, value: str) -> None: """add a key-value pair (allowing other values to exist) alias of `wsgiref.headers.Headers.add_header` """ - super().add_header(key, value, **mediatype_params) + super().add_header(key, value) - def add_many(self, pairs: Iterable[tuple[str, str]]): + def add_many(self, pairs: Iterable[tuple[str, str]]) -> None: for _key, _value in pairs: self.add(_key, _value) diff --git a/addon_toolkit/json_arguments.py b/addon_toolkit/json_arguments.py index c1ccb122..4c79c781 100644 --- a/addon_toolkit/json_arguments.py +++ b/addon_toolkit/json_arguments.py @@ -4,6 +4,11 @@ import types import typing +from addon_toolkit.typing import ( + AnyDataclassInstance, + DataclassInstance, +) + __all__ = ( "kwargs_from_json", @@ -13,7 +18,7 @@ "jsonschema_for_signature_params", ) -JsonableScalar = str | int | float | None | bool | enum.Enum +JsonableScalar = str | int | float | None | bool | enum.Enum | AnyDataclassInstance JsonableList = typing.Iterable["JsonableValue"] JsonableDict = typing.Mapping[str, "JsonableValue"] JsonableValue = JsonableScalar | JsonableList | JsonableDict @@ -45,7 +50,7 @@ def jsonschema_for_signature_params(signature: inspect.Signature) -> JsonableDic } -def jsonschema_for_dataclass(dataclass: type) -> JsonableDict: +def jsonschema_for_dataclass(dataclass: type[DataclassInstance]) -> JsonableDict: """build jsonschema corresponding to the constructor signature of a dataclass""" assert dataclasses.is_dataclass(dataclass) and isinstance(dataclass, type) return jsonschema_for_signature_params(inspect.signature(dataclass)) @@ -67,7 +72,9 @@ def jsonschema_for_annotation(annotation: type) -> JsonableDict: # TODO generic type: def json_for_typed_value[_ValueType: object](type_annotation: type[_ValueType], value: _ValueType): -def json_for_typed_value(type_annotation: typing.Any, value: typing.Any): +def json_for_typed_value( + type_annotation: typing.Any, value: typing.Any +) -> JsonableValue: """return json-serializable representation of field value >>> json_for_typed_value(int, 13) @@ -87,11 +94,12 @@ def json_for_typed_value(type_annotation: typing.Any, value: typing.Any): raise ValueError(f"expected instance of {_type}, got {value}") return json_for_dataclass(value) if issubclass(_type, enum.Enum): + assert isinstance(value, enum.Enum) if value not in _type: raise ValueError(f"expected member of enum {_type}, got {value}") return value.name if _type in (str, int, float): # check str before Iterable - return _type(value) + return _type(value) # type: ignore[no-any-return] if isinstance(_type, types.GenericAlias): # parameterized generic like `list[int]` if issubclass(_type.__origin__, typing.Iterable): @@ -147,7 +155,7 @@ def arg_value_from_json( raise NotImplementedError(f"what do with `{json_arg_value}` (value for {param})") -def json_for_dataclass(dataclass_instance) -> JsonableDict: +def json_for_dataclass(dataclass_instance: DataclassInstance) -> JsonableDict: """return json-serializable representation of the dataclass instance""" _field_value_pairs = ( (_field, getattr(dataclass_instance, _field.name)) @@ -160,7 +168,10 @@ def json_for_dataclass(dataclass_instance) -> JsonableDict: } -def dataclass_from_json(dataclass: type, dataclass_json: JsonableDict): +def dataclass_from_json( + dataclass: type[DataclassInstance], + dataclass_json: JsonableDict, +) -> DataclassInstance: return dataclass( **{ _field.name: field_value_from_json(_field, dataclass_json) @@ -170,8 +181,9 @@ def dataclass_from_json(dataclass: type, dataclass_json: JsonableDict): def field_value_from_json( - field: dataclasses.Field[JsonableValue], dataclass_json: JsonableDict -): + field: dataclasses.Field[JsonableValue], + dataclass_json: JsonableDict, +) -> JsonableValue: _json_value = dataclass_json.get(field.name) if _json_value is None: return None # TODO: check optional diff --git a/addon_toolkit/operation.py b/addon_toolkit/operation.py index 6ce30096..de7f6a4c 100644 --- a/addon_toolkit/operation.py +++ b/addon_toolkit/operation.py @@ -44,7 +44,7 @@ def for_function( ) -> "AddonOperationDeclaration": return addon_operation.get_declaration(fn) - def __post_init__(self): + def __post_init__(self) -> None: if self.required_return_type and not issubclass( self.return_type, self.required_return_type ): @@ -64,7 +64,7 @@ def return_type(self) -> type: return _return_type @property - def name(self): + def name(self) -> str: # TODO: language tag (kwarg for tagged string?) return self.operation_fn.__name__ diff --git a/addon_toolkit/tests/_doctest.py b/addon_toolkit/tests/_doctest.py index 60239610..8d96f7dd 100644 --- a/addon_toolkit/tests/_doctest.py +++ b/addon_toolkit/tests/_doctest.py @@ -1,7 +1,26 @@ import doctest +import typing +import unittest +from types import ModuleType -def load_doctests(*modules): +class LoadTestsFunction(typing.Protocol): + """structural type for the function expected by the "load_tests protocol" + https://docs.python.org/3/library/unittest.html#load-tests-protocol + """ + + def __call__( + _, # implicit, nonexistent "self" + loader: unittest.TestLoader, + tests: unittest.TestSuite, + pattern: str | None, + ) -> unittest.TestSuite: ... + + +def load_doctests( + *modules: ModuleType, + doctestflags: int = doctest.ELLIPSIS | doctest.NORMALIZE_WHITESPACE, +) -> LoadTestsFunction: """shorthand for unittests from doctests meant for implementing the "load_tests protocol" @@ -18,12 +37,16 @@ def load_doctests(*modules): (if there's a need, could support pass-thru kwargs to DocTestSuite) """ - def _load_tests(loader, tests, pattern): + def _load_tests( + loader: unittest.TestLoader, + tests: unittest.TestSuite, + pattern: str | None, + ) -> unittest.TestSuite: for _module in modules: tests.addTests( doctest.DocTestSuite( _module, - optionflags=doctest.ELLIPSIS | doctest.NORMALIZE_WHITESPACE, + optionflags=doctestflags, ) ) return tests diff --git a/addon_toolkit/typing.py b/addon_toolkit/typing.py new file mode 100644 index 00000000..baa68e30 --- /dev/null +++ b/addon_toolkit/typing.py @@ -0,0 +1,8 @@ +import typing + + +class AnyDataclassInstance(typing.Protocol): + __dataclass_fields__: typing.ClassVar[dict[str, typing.Any]] + + +DataclassInstance = typing.TypeVar("DataclassInstance", bound=AnyDataclassInstance) diff --git a/mypy.ini b/mypy.ini index 6d83679c..1789d58a 100644 --- a/mypy.ini +++ b/mypy.ini @@ -28,10 +28,7 @@ warn_unreachable = True # module-specific config [mypy-addon_toolkit.*] -## loosen strict: -disallow_untyped_calls = False -disallow_incomplete_defs = False -disallow_untyped_defs = False +# addon_toolkit fully typed! [mypy-addon_service.*,app.*,manage] # got untyped dependencies -- this is fine @@ -47,5 +44,6 @@ disallow_untyped_defs = False warn_return_any = False [mypy-addon_service.migrations.*] +# django migrations are whatever strict = False disable_error_code = var-annotated