diff --git a/tests/unit/test__extras.py b/tests/unit/test__extras.py index e15319c..27f7227 100644 --- a/tests/unit/test__extras.py +++ b/tests/unit/test__extras.py @@ -22,6 +22,7 @@ def test_load_entry_points(mocker): # Verify func.assert_called_with("plugin-value") + iter_entry_points.assert_called_with(group="entry-point") def test_install(mocker): diff --git a/tests/unit/test_arguments.py b/tests/unit/test_arguments.py index 0330111..65e3c99 100644 --- a/tests/unit/test_arguments.py +++ b/tests/unit/test_arguments.py @@ -1,3 +1,5 @@ +from typing import Annotated + import pytest # Local imports @@ -133,6 +135,22 @@ def dummy(): builder.add_annotation(type, "arg1") assert builder.remaining_args_count == 1 + def test_add_annotation_with_annotated_marker(self): + builder = arguments.ArgumentAnnotationHandlerBuilder(None, ["arg1"], False) + annotation = builder.add_annotation(Annotated[float, arguments.Query], "arg1") + assert isinstance(annotation, arguments.Query) + assert annotation.type is float + assert annotation.name == "arg1" + assert "arg1" not in builder.missing_arguments + + def test_add_annotation_with_annotated_non_marker(self): + def dummy(): + pass + + builder = arguments.ArgumentAnnotationHandlerBuilder(dummy, ["arg1"], False) + builder.add_annotation(Annotated[int, "not-a-marker"], "arg1") + assert builder.remaining_args_count == 1 + @inject_args def test_set_annotations(self, mocker, argument_mock, args): builder = arguments.ArgumentAnnotationHandlerBuilder(None, args, False) diff --git a/tests/unit/test_utils.py b/tests/unit/test_utils.py index ab897f9..2980246 100644 --- a/tests/unit/test_utils.py +++ b/tests/unit/test_utils.py @@ -15,6 +15,21 @@ def func(pos1, *args: 2, **kwargs: 3) -> 4: assert signature.return_annotation == 4 +def test_get_arg_spec_with_deferred_annotations(): + namespace = {} + exec( + "from __future__ import annotations\n" + "def func(pos1, *args: int, **kwargs: str) -> bool:\n" + " pass\n", + namespace, + ) + func = namespace["func"] + + signature = utils.get_arg_spec(func) + assert signature.annotations == {"args": int, "kwargs": str} + assert signature.return_annotation is bool + + def test_call_args(): def func(pos1, *args, **kwargs): pass diff --git a/uplink/_extras.py b/uplink/_extras.py index 5d9d9ea..cf8748c 100644 --- a/uplink/_extras.py +++ b/uplink/_extras.py @@ -36,7 +36,7 @@ def load_entry_points( for name in _entry_points: plugins = { entry_point.name: entry_point.load() - for entry_point in _iter_entry_points(name=name) + for entry_point in _iter_entry_points(group=name) } func = _entry_points[name] for value in plugins.values(): diff --git a/uplink/arguments.py b/uplink/arguments.py index 4fb8650..d3f16f2 100644 --- a/uplink/arguments.py +++ b/uplink/arguments.py @@ -7,6 +7,7 @@ import collections import functools import inspect +import typing # Local imports from uplink import exceptions, hooks, interfaces, utils @@ -87,8 +88,19 @@ def _is_annotation(annotation): cls = interfaces.Annotation return utils.is_subclass(annotation, cls) or isinstance(annotation, cls) + @classmethod + def _unwrap_annotated(cls, annotation): + if typing.get_origin(annotation) is not typing.Annotated: + return annotation, None + arg_type, *metadata = typing.get_args(annotation) + marker = next((m for m in metadata if cls._is_annotation(m)), None) + return (marker, arg_type) if marker is not None else (annotation, None) + def add_annotation(self, annotation, name=None, *args, **kwargs): + annotation, arg_type = self._unwrap_annotated(annotation) if self._is_annotation(annotation): + if arg_type is not None: + self._argument_types[name] = arg_type return self._add_annotation(annotation, name) self._argument_types[name] = annotation return None diff --git a/uplink/utils.py b/uplink/utils.py index cf19bd4..3f1b28d 100644 --- a/uplink/utils.py +++ b/uplink/utils.py @@ -46,7 +46,8 @@ def get_call_args(f, *args, **kwargs): return collections.OrderedDict(new_arguments) def get_arg_spec(f): - sig = signature(f) + # resolve string annotations (e.g. from `__future__.annotations`): + sig = signature(f, eval_str=True) parameters = sig.parameters args = [] annotations = collections.OrderedDict()