Coverage for src/lexigram/graphql/resolvers/adapter.py: 0%

26 statements  

« prev     ^ index     » next       coverage.py v7.15.4, created at 2026-08-25 04:37 +0800

1"""Helper utilities for working with GraphQL resolvers. 

2 

3This module contains the :class:`ResolverAdapter` implementation and a 

4convenience decorator. They were formerly defined in the package 

5``__init__`` but have been moved here to keep package root clean. 

6""" 

7 

8from __future__ import annotations 

9 

10import inspect 

11from typing import TYPE_CHECKING, Any 

12 

13if TYPE_CHECKING: 

14 from lexigram.contracts.graphql import ResolverProtocol 

15 

16 

17class ResolverAdapter: 

18 """Adapts a plain callable to the :class:`~lexigram.contracts.graphql.ResolverProtocol` protocol. 

19 

20 Accepts any sync or async callable with the signature 

21 ``(parent, args, context, info) -> Any`` and exposes it as a ``ResolverProtocol`` 

22 via a ``resolve`` method. Sync callables are called directly; their 

23 return value is awaited only when it happens to be a coroutine. 

24 

25 Args: 

26 func: A sync or async callable that implements the resolver logic. 

27 The callable should accept up to four positional arguments: 

28 ``(parent, args, context, info)``. Arguments beyond what the 

29 callable declares are silently dropped so that short signatures 

30 (e.g. ``lambda parent, args: parent.id``) work transparently. 

31 """ 

32 

33 __slots__ = ("_func", "_nparams") 

34 

35 def __init__(self, func: Any) -> None: 

36 self._func = func 

37 # Introspect arity once so we can trim args at call time. 

38 try: 

39 sig = inspect.signature(func) 

40 self._nparams: int | None = len(sig.parameters) 

41 except (ValueError, TypeError): 

42 self._nparams = None # unknown — pass all 4 args 

43 

44 async def resolve( 

45 self, 

46 parent: Any, 

47 args: dict[str, Any], 

48 context: Any, 

49 info: Any, 

50 ) -> Any: 

51 """Invoke the wrapped callable with the resolver arguments. 

52 

53 Args: 

54 parent: Parent (root) object for the field. 

55 args: Parsed field arguments from the GraphQL query. 

56 context: Shared execution context. 

57 info: GraphQL resolution info object. 

58 

59 Returns: 

60 Resolved field value. 

61 """ 

62 all_args = (parent, args, context, info) 

63 call_args = all_args[: self._nparams] if self._nparams is not None else all_args 

64 

65 result = self._func(*call_args) 

66 if inspect.isawaitable(result): 

67 return await result 

68 return result 

69 

70 def __repr__(self) -> str: 

71 name = getattr(self._func, "__name__", repr(self._func)) 

72 return f"ResolverAdapter({name!r})" 

73 

74 

75def resolver(func: Any) -> ResolverAdapter: 

76 """Decorator that wraps a plain callable as a :class:`ResolverAdapter`. 

77 

78 Example:: 

79 

80 @resolver 

81 async def user_resolver(parent, args, context, info): 

82 return await context.users.get(args["id"]) 

83 

84 Args: 

85 func: Callable to wrap. 

86 

87 Returns: 

88 :class:`ResolverAdapter` wrapping *func*. 

89 """ 

90 return ResolverAdapter(func) 

91 

92 

93# Type narrowing — confirm ResolverAdapter satisfies the protocol at import time. 

94_: ResolverProtocol = ResolverAdapter(lambda: None) 

95del _