Coverage for human_requests / base.py: 91%
46 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-03-07 17:38 +0000
« prev ^ index » next coverage.py v7.13.4, created at 2026-03-07 17:38 +0000
1from __future__ import annotations
3from dataclasses import Field, field
4from inspect import signature
5from typing import Any, Callable, Generic, TypeVar, cast
7ParentT = TypeVar("ParentT")
8FactoryParentT = TypeVar("FactoryParentT")
9FactoryChildT = TypeVar("FactoryChildT")
11_API_CHILD_FACTORY_META = "human_requests_api_child_factory"
12_UNSET = object()
15class ApiChild(Generic[ParentT]):
16 """
17 Base class for API child services that keeps a typed parent reference.
18 """
20 _parent: ParentT
22 def __init__(self, parent: ParentT) -> None:
23 self._parent = parent
25 @property
26 def parent(self) -> ParentT:
27 return self._parent
30class ApiParent:
31 """
32 Dataclass mixin that initializes fields declared with `api_child_field`.
33 """
35 def __post_init__(self) -> None:
36 dataclass_fields = getattr(self, "__dataclass_fields__", None)
37 if not isinstance(dataclass_fields, dict):
38 raise TypeError("ApiParent can only be used with dataclasses")
40 for dataclass_field in cast(dict[str, Field[Any]], dataclass_fields).values():
41 child_factory = cast(
42 Callable[..., Any] | None,
43 dataclass_field.metadata.get(_API_CHILD_FACTORY_META),
44 )
45 if child_factory is None:
46 continue
48 # Keep initialization idempotent if parent __post_init__ is called twice.
49 if getattr(self, dataclass_field.name, _UNSET) is not _UNSET:
50 continue
52 setattr(self, dataclass_field.name, _create_child(child_factory, self))
55def api_child_field(
56 child_factory: Callable[..., FactoryChildT],
57 *,
58 repr: bool = False,
59 compare: bool = False,
60) -> FactoryChildT:
61 """
62 Dataclass field helper for child API services initialized in `ApiParent.__post_init__`.
63 """
65 return cast(
66 FactoryChildT,
67 field(
68 init=False,
69 repr=repr,
70 compare=compare,
71 metadata={_API_CHILD_FACTORY_META: child_factory},
72 ),
73 )
76def _create_child(child_factory: Callable[..., Any], parent: Any) -> Any:
77 try:
78 call_signature = signature(child_factory)
79 accepts_parent = _can_bind_single_positional(call_signature, parent)
80 except (TypeError, ValueError):
81 # Fallback for callables without inspectable signatures.
82 accepts_parent = True
84 child = child_factory(parent) if accepts_parent else child_factory()
85 if not accepts_parent and isinstance(child, ApiChild):
86 child._parent = parent
87 return child
90def _can_bind_single_positional(call_signature: Any, value: Any) -> bool:
91 try:
92 call_signature.bind(value)
93 return True
94 except TypeError:
95 return False