Coverage for injector/__init__.py: 72%

Shortcuts on this page

r m x p   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

530 statements  

1# encoding: utf-8 

2# 

3# Copyright (C) 2010 Alec Thomas <alec@swapoff.org> 

4# All rights reserved. 

5# 

6# This software is licensed as described in the file COPYING, which 

7# you should have received as part of this distribution. 

8# 

9# Author: Alec Thomas <alec@swapoff.org> 

10 

11"""Injector - Python dependency injection framework, inspired by Guice 

12 

13:copyright: (c) 2012 by Alec Thomas 

14:license: BSD 

15""" 

16 

17import functools 

18import inspect 

19import itertools 

20import logging 

21import sys 

22import threading 

23import types 

24from abc import ABCMeta, abstractmethod 

25from collections import namedtuple 

26from typing import ( 

27 Any, 

28 Callable, 

29 cast, 

30 Dict, 

31 Generic, 

32 Iterable, 

33 List, 

34 Optional, 

35 overload, 

36 Tuple, 

37 Type, 

38 TypeVar, 

39 Union, 

40) 

41 

42from typing_extensions import NoReturn 

43 

44HAVE_ANNOTATED = sys.version_info >= (3, 7, 0) 

45 

46if HAVE_ANNOTATED: 46 ↛ 51line 46 didn't jump to line 51, because the condition on line 46 was never false

47 # Ignoring errors here as typing_extensions stub doesn't know about those things yet 

48 from typing_extensions import _AnnotatedAlias, Annotated, get_type_hints # type: ignore 

49else: 

50 

51 class Annotated: # type: ignore 

52 pass 

53 

54 from typing import get_type_hints as _get_type_hints 

55 

56 def get_type_hints( 

57 obj: Callable[..., Any], 

58 globalns: Optional[Dict[str, Any]] = None, 

59 localns: Optional[Dict[str, Any]] = None, 

60 include_extras: bool = False, 

61 ) -> Dict[str, Any]: 

62 return _get_type_hints(obj, globalns, localns) 

63 

64 

65TYPING353 = hasattr(Union[str, int], '__origin__') 

66 

67 

68__author__ = 'Alec Thomas <alec@swapoff.org>' 

69__version__ = '0.18.3' 

70__version_tag__ = '' 

71 

72log = logging.getLogger('injector') 

73log.addHandler(logging.NullHandler()) 

74 

75if log.level == logging.NOTSET: 75 ↛ 78line 75 didn't jump to line 78, because the condition on line 75 was never false

76 log.setLevel(logging.WARN) 

77 

78T = TypeVar('T') 

79K = TypeVar('K') 

80V = TypeVar('V') 

81 

82 

83def private(something: T) -> T: 

84 something.__private__ = True # type: ignore 

85 return something 

86 

87 

88CallableT = TypeVar('CallableT', bound=Callable) 

89 

90 

91def synchronized(lock: threading.RLock) -> Callable[[CallableT], CallableT]: 

92 def outside_wrapper(function: CallableT) -> CallableT: 

93 @functools.wraps(function) 

94 def wrapper(*args: Any, **kwargs: Any) -> Any: 

95 with lock: 

96 return function(*args, **kwargs) 

97 

98 return cast(CallableT, wrapper) 

99 

100 return outside_wrapper 

101 

102 

103lock = threading.RLock() 

104 

105 

106_inject_marker = object() 

107_noinject_marker = object() 

108 

109if HAVE_ANNOTATED: 109 ↛ 191line 109 didn't jump to line 191, because the condition on line 109 was never false

110 InjectT = TypeVar('InjectT') 

111 Inject = Annotated[InjectT, _inject_marker] 

112 """An experimental way to declare injectable dependencies utilizing a `PEP 593`_ implementation 

113 in `typing_extensions`. 

114 

115 Those two declarations are equivalent:: 

116 

117 @inject 

118 def fun(t: SomeType) -> None: 

119 pass 

120 

121 def fun(t: Inject[SomeType]) -> None: 

122 pass 

123 

124 The advantage over using :func:`inject` is that if you have some noninjectable parameters 

125 it may be easier to spot what are they. Those two are equivalent:: 

126 

127 @inject 

128 @noninjectable('s') 

129 def fun(t: SomeType, s: SomeOtherType) -> None: 

130 pass 

131 

132 def fun(t: Inject[SomeType], s: SomeOtherType) -> None: 

133 pass 

134 

135 .. seealso:: 

136 

137 Function :func:`get_bindings` 

138 A way to inspect how various injection declarations interact with each other. 

139 

140 .. versionadded:: 0.18.0 

141 .. note:: Requires Python 3.7+. 

142 .. note:: 

143 

144 If you're using mypy you need the version 0.750 or newer to fully type-check code using this 

145 construct. 

146 

147 .. _PEP 593: https://www.python.org/dev/peps/pep-0593/ 

148 .. _typing_extensions: https://pypi.org/project/typing-extensions/ 

149 """ 

150 

151 NoInject = Annotated[InjectT, _noinject_marker] 

152 """An experimental way to declare noninjectable dependencies utilizing a `PEP 593`_ implementation 

153 in `typing_extensions`. 

154 

155 Since :func:`inject` declares all function's parameters to be injectable there needs to be a way 

156 to opt out of it. This has been provided by :func:`noninjectable` but `noninjectable` suffers from 

157 two issues: 

158 

159 * You need to repeat the parameter name 

160 * The declaration may be relatively distance in space from the actual parameter declaration, thus 

161 hindering readability 

162 

163 `NoInject` solves both of those concerns, for example (those two declarations are equivalent):: 

164 

165 @inject 

166 @noninjectable('b') 

167 def fun(a: TypeA, b: TypeB) -> None: 

168 pass 

169 

170 @inject 

171 def fun(a: TypeA, b: NoInject[TypeB]) -> None: 

172 pass 

173 

174 .. seealso:: 

175 

176 Function :func:`get_bindings` 

177 A way to inspect how various injection declarations interact with each other. 

178 

179 .. versionadded:: 0.18.0 

180 .. note:: Requires Python 3.7+. 

181 .. note:: 

182 

183 If you're using mypy you need the version 0.750 or newer to fully type-check code using this 

184 construct. 

185 

186 .. _PEP 593: https://www.python.org/dev/peps/pep-0593/ 

187 .. _typing_extensions: https://pypi.org/project/typing-extensions/ 

188 """ 

189 

190 

191def reraise(original: Exception, exception: Exception, maximum_frames: int = 1) -> NoReturn: 

192 prev_cls, prev, tb = sys.exc_info() 

193 frames = inspect.getinnerframes(cast(types.TracebackType, tb)) 

194 if len(frames) > maximum_frames: 

195 exception = original 

196 raise exception.with_traceback(tb) 

197 

198 

199class Error(Exception): 

200 """Base exception.""" 

201 

202 

203class UnsatisfiedRequirement(Error): 

204 """Requirement could not be satisfied.""" 

205 

206 def __init__(self, owner: Optional[object], interface: type) -> None: 

207 super().__init__(owner, interface) 

208 self.owner = owner 

209 self.interface = interface 

210 

211 def __str__(self) -> str: 

212 on = '%s has an ' % _describe(self.owner) if self.owner else '' 

213 return '%sunsatisfied requirement on %s' % (on, _describe(self.interface)) 

214 

215 

216class CallError(Error): 

217 """Call to callable object fails.""" 

218 

219 def __str__(self) -> str: 

220 if len(self.args) == 1: 

221 return self.args[0] 

222 

223 instance, method, args, kwargs, original_error, stack = self.args 

224 if hasattr(method, 'im_class'): 

225 instance = method.__self__ 

226 method_name = method.__func__.__name__ 

227 else: 

228 method_name = method.__name__ 

229 

230 cls = instance.__class__.__name__ if instance is not None else '' 

231 

232 full_method = '.'.join((cls, method_name)).strip('.') 

233 

234 parameters = ', '.join( 

235 itertools.chain( 

236 (repr(arg) for arg in args), ('%s=%r' % (key, value) for (key, value) in kwargs.items()) 

237 ) 

238 ) 

239 return 'Call to %s(%s) failed: %s (injection stack: %r)' % ( 

240 full_method, 

241 parameters, 

242 original_error, 

243 [level[0] for level in stack], 

244 ) 

245 

246 

247class CircularDependency(Error): 

248 """Circular dependency detected.""" 

249 

250 

251class UnknownProvider(Error): 

252 """Tried to bind to a type whose provider couldn't be determined.""" 

253 

254 

255class UnknownArgument(Error): 

256 """Tried to mark an unknown argument as noninjectable.""" 

257 

258 

259class Provider(Generic[T]): 

260 """Provides class instances.""" 

261 

262 __metaclass__ = ABCMeta 

263 

264 @abstractmethod 

265 def get(self, injector: 'Injector') -> T: 

266 raise NotImplementedError # pragma: no cover 

267 

268 

269class ClassProvider(Provider): 

270 """Provides instances from a given class, created using an Injector.""" 

271 

272 def __init__(self, cls: Type[T]) -> None: 

273 self._cls = cls 

274 

275 def get(self, injector: 'Injector') -> T: 

276 return injector.create_object(self._cls) 

277 

278 

279class CallableProvider(Provider): 

280 """Provides something using a callable. 

281 

282 The callable is called every time new value is requested from the provider. 

283 

284 There's no need to explicitly use :func:`inject` or :data:`Inject` with the callable as it's 

285 assumed that, if the callable has annotated parameters, they're meant to be provided 

286 automatically. It wouldn't make sense any other way, as there's no mechanism to provide 

287 parameters to the callable at a later time, so either they'll be injected or there'll be 

288 a `CallError`. 

289 

290 :: 

291 

292 >>> class MyClass: 

293 ... def __init__(self, value: int) -> None: 

294 ... self.value = value 

295 ... 

296 >>> def factory(): 

297 ... print('providing') 

298 ... return MyClass(42) 

299 ... 

300 >>> def configure(binder): 

301 ... binder.bind(MyClass, to=CallableProvider(factory)) 

302 ... 

303 >>> injector = Injector(configure) 

304 >>> injector.get(MyClass) is injector.get(MyClass) 

305 providing 

306 providing 

307 False 

308 """ 

309 

310 def __init__(self, callable: Callable[..., T]): 

311 self._callable = callable 

312 

313 def get(self, injector: 'Injector') -> T: 

314 return injector.call_with_injection(self._callable) 

315 

316 def __repr__(self) -> str: 

317 return '%s(%r)' % (type(self).__name__, self._callable) 

318 

319 

320class InstanceProvider(Provider): 

321 """Provide a specific instance. 

322 

323 :: 

324 

325 >>> class MyType: 

326 ... def __init__(self): 

327 ... self.contents = [] 

328 >>> def configure(binder): 

329 ... binder.bind(MyType, to=InstanceProvider(MyType())) 

330 ... 

331 >>> injector = Injector(configure) 

332 >>> injector.get(MyType) is injector.get(MyType) 

333 True 

334 >>> injector.get(MyType).contents.append('x') 

335 >>> injector.get(MyType).contents 

336 ['x'] 

337 """ 

338 

339 def __init__(self, instance: T) -> None: 

340 self._instance = instance 

341 

342 def get(self, injector: 'Injector') -> T: 

343 return self._instance 

344 

345 def __repr__(self) -> str: 

346 return '%s(%r)' % (type(self).__name__, self._instance) 

347 

348 

349@private 

350class ListOfProviders(Provider, Generic[T]): 

351 """Provide a list of instances via other Providers.""" 

352 

353 def __init__(self) -> None: 

354 self._providers = [] # type: List[Provider[T]] 

355 

356 def append(self, provider: Provider[T]) -> None: 

357 self._providers.append(provider) 

358 

359 def __repr__(self) -> str: 

360 return '%s(%r)' % (type(self).__name__, self._providers) 

361 

362 

363class MultiBindProvider(ListOfProviders[List[T]]): 

364 """Used by :meth:`Binder.multibind` to flatten results of providers that 

365 return sequences.""" 

366 

367 def get(self, injector: 'Injector') -> List[T]: 

368 return [i for provider in self._providers for i in provider.get(injector)] 

369 

370 

371class MapBindProvider(ListOfProviders[Dict[str, T]]): 

372 """A provider for map bindings.""" 

373 

374 def get(self, injector: 'Injector') -> Dict[str, T]: 

375 map = {} # type: Dict[str, T] 

376 for provider in self._providers: 

377 map.update(provider.get(injector)) 

378 return map 

379 

380 

381_BindingBase = namedtuple('_BindingBase', 'interface provider scope') 

382 

383 

384@private 

385class Binding(_BindingBase): 

386 """A binding from an (interface,) to a provider in a scope.""" 

387 

388 def is_multibinding(self) -> bool: 

389 return _get_origin(_punch_through_alias(self.interface)) in {dict, list} 

390 

391 

392_InstallableModuleType = Union[Callable[['Binder'], None], 'Module', Type['Module']] 

393 

394 

395class Binder: 

396 """Bind interfaces to implementations. 

397 

398 .. note:: This class is instantiated internally for you and there's no need 

399 to instantiate it on your own. 

400 """ 

401 

402 @private 

403 def __init__(self, injector: 'Injector', auto_bind: bool = True, parent: 'Binder' = None) -> None: 

404 """Create a new Binder. 

405 

406 :param injector: Injector we are binding for. 

407 :param auto_bind: Whether to automatically bind missing types. 

408 :param parent: Parent binder. 

409 """ 

410 self.injector = injector 

411 self._auto_bind = auto_bind 

412 self._bindings = {} # type: Dict[type, Binding] 

413 self.parent = parent 

414 

415 def bind( 

416 self, 

417 interface: Type[T], 

418 to: Union[None, T, Callable[..., T], Provider[T]] = None, 

419 scope: Union[None, Type['Scope'], 'ScopeDecorator'] = None, 

420 ) -> None: 

421 """Bind an interface to an implementation. 

422 

423 Binding `T` to an instance of `T` like 

424 

425 :: 

426 

427 binder.bind(A, to=A('some', 'thing')) 

428 

429 is, for convenience, a shortcut for 

430 

431 :: 

432 

433 binder.bind(A, to=InstanceProvider(A('some', 'thing'))). 

434 

435 Likewise, binding to a callable like 

436 

437 :: 

438 

439 binder.bind(A, to=some_callable) 

440 

441 is a shortcut for 

442 

443 :: 

444 

445 binder.bind(A, to=CallableProvider(some_callable)) 

446 

447 and, as such, if `some_callable` there has any annotated parameters they'll be provided 

448 automatically without having to use :func:`inject` or :data:`Inject` with the callable. 

449 

450 `typing.List` and `typing.Dict` instances are reserved for multibindings and trying to bind them 

451 here will result in an error (use :meth:`multibind` instead):: 

452 

453 binder.bind(List[str], to=['hello', 'there']) # Error 

454 

455 :param interface: Type to bind. 

456 :param to: Instance or class to bind to, or an instance of 

457 :class:`Provider` subclass. 

458 :param scope: Optional :class:`Scope` in which to bind. 

459 """ 

460 if _get_origin(_punch_through_alias(interface)) in {dict, list}: 460 ↛ 461line 460 didn't jump to line 461, because the condition on line 460 was never true

461 raise Error( 

462 'Type %s is reserved for multibindings. Use multibind instead of bind.' % (interface,) 

463 ) 

464 self._bindings[interface] = self.create_binding(interface, to, scope) 

465 

466 @overload 

467 def multibind( 

468 self, 

469 interface: Type[List[T]], 

470 to: Union[List[T], Callable[..., List[T]], Provider[List[T]]], 

471 scope: Union[Type['Scope'], 'ScopeDecorator'] = None, 

472 ) -> None: # pragma: no cover 

473 pass 

474 

475 @overload 

476 def multibind( 

477 self, 

478 interface: Type[Dict[K, V]], 

479 to: Union[Dict[K, V], Callable[..., Dict[K, V]], Provider[Dict[K, V]]], 

480 scope: Union[Type['Scope'], 'ScopeDecorator'] = None, 

481 ) -> None: # pragma: no cover 

482 pass 

483 

484 def multibind( 

485 self, interface: type, to: Any, scope: Union['ScopeDecorator', Type['Scope']] = None 

486 ) -> None: 

487 """Creates or extends a multi-binding. 

488 

489 A multi-binding contributes values to a list or to a dictionary. For example:: 

490 

491 binder.multibind(List[str], to=['some', 'strings']) 

492 binder.multibind(List[str], to=['other', 'strings']) 

493 injector.get(List[str]) # ['some', 'strings', 'other', 'strings'] 

494 

495 binder.multibind(Dict[str, int], to={'key': 11}) 

496 binder.multibind(Dict[str, int], to={'other_key': 33}) 

497 injector.get(Dict[str, int]) # {'key': 11, 'other_key': 33} 

498 

499 .. versionchanged:: 0.17.0 

500 Added support for using `typing.Dict` and `typing.List` instances as interfaces. 

501 Deprecated support for `MappingKey`, `SequenceKey` and single-item lists and 

502 dictionaries as interfaces. 

503 

504 :param interface: typing.Dict or typing.List instance to bind to. 

505 :param to: Instance, class to bind to, or an explicit :class:`Provider` 

506 subclass. Must provide a list or a dictionary, depending on the interface. 

507 :param scope: Optional Scope in which to bind. 

508 """ 

509 if interface not in self._bindings: 

510 if ( 

511 isinstance(interface, dict) 

512 or isinstance(interface, type) 

513 and issubclass(interface, dict) 

514 or _get_origin(_punch_through_alias(interface)) is dict 

515 ): 

516 provider = MapBindProvider() # type: ListOfProviders 

517 else: 

518 provider = MultiBindProvider() 

519 binding = self.create_binding(interface, provider, scope) 

520 self._bindings[interface] = binding 

521 else: 

522 binding = self._bindings[interface] 

523 provider = binding.provider 

524 assert isinstance(provider, ListOfProviders) 

525 provider.append(self.provider_for(interface, to)) 

526 

527 def install(self, module: _InstallableModuleType) -> None: 

528 """Install a module into this binder. 

529 

530 In this context the module is one of the following: 

531 

532 * function taking the :class:`Binder` as it's only parameter 

533 

534 :: 

535 

536 def configure(binder): 

537 bind(str, to='s') 

538 

539 binder.install(configure) 

540 

541 * instance of :class:`Module` (instance of it's subclass counts) 

542 

543 :: 

544 

545 class MyModule(Module): 

546 def configure(self, binder): 

547 binder.bind(str, to='s') 

548 

549 binder.install(MyModule()) 

550 

551 * subclass of :class:`Module` - the subclass needs to be instantiable so if it 

552 expects any parameters they need to be injected 

553 

554 :: 

555 

556 binder.install(MyModule) 

557 """ 

558 if type(module) is type and issubclass(cast(type, module), Module): 

559 instance = cast(type, module)() 

560 else: 

561 instance = module 

562 instance(self) 

563 

564 def create_binding( 

565 self, interface: type, to: Any = None, scope: Union['ScopeDecorator', Type['Scope']] = None 

566 ) -> Binding: 

567 provider = self.provider_for(interface, to) 

568 scope = scope or getattr(to or interface, '__scope__', NoScope) 

569 if isinstance(scope, ScopeDecorator): 569 ↛ 570line 569 didn't jump to line 570, because the condition on line 569 was never true

570 scope = scope.scope 

571 return Binding(interface, provider, scope) 

572 

573 def provider_for(self, interface: Any, to: Any = None) -> Provider: 

574 base_type = _punch_through_alias(interface) 

575 origin = _get_origin(base_type) 

576 

577 if interface is Any: 577 ↛ 578line 577 didn't jump to line 578, because the condition on line 577 was never true

578 raise TypeError('Injecting Any is not supported') 

579 elif _is_specialization(interface, ProviderOf): 

580 (target,) = interface.__args__ 

581 if to is not None: 581 ↛ 582line 581 didn't jump to line 582, because the condition on line 581 was never true

582 raise Exception('ProviderOf cannot be bound to anything') 

583 return InstanceProvider(ProviderOf(self.injector, target)) 

584 elif isinstance(to, Provider): 

585 return to 

586 elif isinstance( 

587 to, 

588 ( 

589 types.FunctionType, 

590 types.LambdaType, 

591 types.MethodType, 

592 types.BuiltinFunctionType, 

593 types.BuiltinMethodType, 

594 ), 

595 ): 

596 return CallableProvider(to) 

597 elif issubclass(type(to), type): 597 ↛ 598line 597 didn't jump to line 598, because the condition on line 597 was never true

598 return ClassProvider(cast(type, to)) 

599 elif isinstance(interface, BoundKey): 

600 

601 def proxy(injector: Injector) -> Any: 

602 binder = injector.binder 

603 kwarg_providers = { 

604 name: binder.provider_for(None, provider) for (name, provider) in interface.kwargs.items() 

605 } 

606 kwargs = {name: provider.get(injector) for (name, provider) in kwarg_providers.items()} 

607 return interface.interface(**kwargs) 

608 

609 return CallableProvider(inject(proxy)) 

610 elif _is_specialization(interface, AssistedBuilder): 610 ↛ 611line 610 didn't jump to line 611, because the condition on line 610 was never true

611 (target,) = interface.__args__ 

612 builder = interface(self.injector, target) 

613 return InstanceProvider(builder) 

614 elif ( 

615 origin is None 

616 and isinstance(base_type, (tuple, type)) 

617 and interface is not Any 

618 and isinstance(to, base_type) 

619 or origin in {dict, list} 

620 and isinstance(to, origin) 

621 ): 

622 return InstanceProvider(to) 

623 elif issubclass(type(base_type), type) or isinstance(base_type, (tuple, list)): 623 ↛ 629line 623 didn't jump to line 629, because the condition on line 623 was never false

624 if to is not None: 624 ↛ 625line 624 didn't jump to line 625, because the condition on line 624 was never true

625 return InstanceProvider(to) 

626 return ClassProvider(base_type) 

627 

628 else: 

629 raise UnknownProvider('couldn\'t determine provider for %r to %r' % (interface, to)) 

630 

631 def _get_binding(self, key: type, *, only_this_binder: bool = False) -> Tuple[Binding, 'Binder']: 

632 binding = self._bindings.get(key) 

633 if binding: 

634 return binding, self 

635 if self.parent and not only_this_binder: 635 ↛ 636line 635 didn't jump to line 636, because the condition on line 635 was never true

636 return self.parent._get_binding(key) 

637 

638 raise KeyError 

639 

640 def get_binding(self, interface: type) -> Tuple[Binding, 'Binder']: 

641 is_scope = isinstance(interface, type) and issubclass(interface, Scope) 

642 try: 

643 return self._get_binding(interface, only_this_binder=is_scope) 

644 except (KeyError, UnsatisfiedRequirement): 

645 if is_scope: 

646 scope = interface 

647 self.bind(scope, to=scope(self.injector)) 

648 return self._get_binding(interface) 

649 # The special interface is added here so that requesting a special 

650 # interface with auto_bind disabled works 

651 if self._auto_bind or self._is_special_interface(interface): 651 ↛ 656line 651 didn't jump to line 656, because the condition on line 651 was never false

652 binding = self.create_binding(interface) 

653 self._bindings[interface] = binding 

654 return binding, self 

655 

656 raise UnsatisfiedRequirement(None, interface) 

657 

658 def _is_special_interface(self, interface: type) -> bool: 

659 # "Special" interfaces are ones that you cannot bind yourself but 

660 # you can request them (for example you cannot bind ProviderOf(SomeClass) 

661 # to anything but you can inject ProviderOf(SomeClass) just fine 

662 return any(_is_specialization(interface, cls) for cls in [AssistedBuilder, ProviderOf]) 

663 

664 

665if TYPING353: 665 ↛ 694line 665 didn't jump to line 694, because the condition on line 665 was never false

666 

667 def _is_specialization(cls: type, generic_class: Any) -> bool: 

668 # Starting with typing 3.5.3/Python 3.6 it is no longer necessarily true that 

669 # issubclass(SomeGeneric[X], SomeGeneric) so we need some other way to 

670 # determine whether a particular object is a generic class with type parameters 

671 # provided. Fortunately there seems to be __origin__ attribute that's useful here. 

672 

673 # We need to special-case Annotated as its __origin__ behaves differently than 

674 # other typing generic classes. See https://github.com/python/typing/pull/635 

675 # for some details. 

676 if HAVE_ANNOTATED and generic_class is Annotated and isinstance(cls, _AnnotatedAlias): 

677 return True 

678 

679 if not hasattr(cls, '__origin__'): 

680 return False 

681 origin = cast(Any, cls).__origin__ 

682 if not inspect.isclass(generic_class): 682 ↛ 683line 682 didn't jump to line 683, because the condition on line 682 was never true

683 generic_class = type(generic_class) 

684 if not inspect.isclass(origin): 684 ↛ 685line 684 didn't jump to line 685, because the condition on line 684 was never true

685 origin = type(origin) 

686 # __origin__ is generic_class is a special case to handle Union as 

687 # Union cannot be used in issubclass() check (it raises an exception 

688 # by design). 

689 return origin is generic_class or issubclass(origin, generic_class) 

690 

691 

692else: 

693 # To maintain compatibility we fall back to an issubclass check. 

694 def _is_specialization(cls: type, generic_class: Any) -> bool: 

695 return isinstance(cls, type) and cls is not Any and issubclass(cls, generic_class) 

696 

697 

698def _punch_through_alias(type_: Any) -> type: 

699 if getattr(type_, '__qualname__', '') == 'NewType.<locals>.new_type': 699 ↛ 700line 699 didn't jump to line 700, because the condition on line 699 was never true

700 return type_.__supertype__ 

701 else: 

702 return type_ 

703 

704 

705def _get_origin(type_: type) -> type: 

706 origin = getattr(type_, '__origin__', None) 

707 # Older typing behaves differently there and stores Dict and List as origin, we need to be flexible. 

708 if origin is List: 708 ↛ 709line 708 didn't jump to line 709, because the condition on line 708 was never true

709 return list 

710 elif origin is Dict: 710 ↛ 711line 710 didn't jump to line 711, because the condition on line 710 was never true

711 return dict 

712 return origin 

713 

714 

715class Scope: 

716 """A Scope looks up the Provider for a binding. 

717 

718 By default (ie. :class:`NoScope` ) this simply returns the default 

719 :class:`Provider` . 

720 """ 

721 

722 __metaclass__ = ABCMeta 

723 

724 def __init__(self, injector: 'Injector') -> None: 

725 self.injector = injector 

726 self.configure() 

727 

728 def configure(self) -> None: 

729 """Configure the scope.""" 

730 

731 @abstractmethod 

732 def get(self, key: Type[T], provider: Provider[T]) -> Provider[T]: 

733 """Get a :class:`Provider` for a key. 

734 

735 :param key: The key to return a provider for. 

736 :param provider: The default Provider associated with the key. 

737 :returns: A Provider instance that can provide an instance of key. 

738 """ 

739 raise NotImplementedError # pragma: no cover 

740 

741 

742class ScopeDecorator: 

743 def __init__(self, scope: Type[Scope]) -> None: 

744 self.scope = scope 

745 

746 def __call__(self, cls: T) -> T: 

747 cast(Any, cls).__scope__ = self.scope 

748 binding = getattr(cls, '__binding__', None) 

749 if binding: 

750 new_binding = Binding(interface=binding.interface, provider=binding.provider, scope=self.scope) 

751 setattr(cls, '__binding__', new_binding) 

752 return cls 

753 

754 def __repr__(self) -> str: 

755 return 'ScopeDecorator(%s)' % self.scope.__name__ 

756 

757 

758class NoScope(Scope): 

759 """An unscoped provider.""" 

760 

761 def get(self, unused_key: Type[T], provider: Provider[T]) -> Provider[T]: 

762 return provider 

763 

764 

765noscope = ScopeDecorator(NoScope) 

766 

767 

768class SingletonScope(Scope): 

769 """A :class:`Scope` that returns a per-Injector instance for a key. 

770 

771 :data:`singleton` can be used as a convenience class decorator. 

772 

773 >>> class A: pass 

774 >>> injector = Injector() 

775 >>> provider = ClassProvider(A) 

776 >>> singleton = SingletonScope(injector) 

777 >>> a = singleton.get(A, provider) 

778 >>> b = singleton.get(A, provider) 

779 >>> a is b 

780 True 

781 """ 

782 

783 def configure(self) -> None: 

784 self._context = {} # type: Dict[type, Provider] 

785 

786 @synchronized(lock) 

787 def get(self, key: Type[T], provider: Provider[T]) -> Provider[T]: 

788 try: 

789 return self._context[key] 

790 except KeyError: 

791 provider = InstanceProvider(provider.get(self.injector)) 

792 self._context[key] = provider 

793 return provider 

794 

795 

796singleton = ScopeDecorator(SingletonScope) 

797 

798 

799class ThreadLocalScope(Scope): 

800 """A :class:`Scope` that returns a per-thread instance for a key.""" 

801 

802 def configure(self) -> None: 

803 self._locals = threading.local() 

804 

805 def get(self, key: Type[T], provider: Provider[T]) -> Provider[T]: 

806 try: 

807 return getattr(self._locals, repr(key)) 

808 except AttributeError: 

809 provider = InstanceProvider(provider.get(self.injector)) 

810 setattr(self._locals, repr(key), provider) 

811 return provider 

812 

813 

814threadlocal = ScopeDecorator(ThreadLocalScope) 

815 

816 

817class Module: 

818 """Configures injector and providers.""" 

819 

820 def __call__(self, binder: Binder) -> None: 

821 """Configure the binder.""" 

822 self.__injector__ = binder.injector 

823 for unused_name, function in inspect.getmembers(self, inspect.ismethod): 

824 binding = None 

825 if hasattr(function, '__binding__'): 

826 binding = function.__binding__ 

827 if binding.interface == '__deferred__': 827 ↛ 830line 827 didn't jump to line 830, because the condition on line 827 was never true

828 # We could not evaluate a forward reference at @provider-decoration time, we need to 

829 # try again now. 

830 try: 

831 annotations = get_type_hints(function) 

832 except NameError as e: 

833 raise NameError( 

834 'Cannot avaluate forward reference annotation(s) in method %r belonging to %r: %s' 

835 % (function.__name__, type(self), e) 

836 ) from e 

837 return_type = annotations['return'] 

838 binding = function.__func__.__binding__ = Binding( 

839 interface=return_type, provider=binding.provider, scope=binding.scope 

840 ) 

841 bind_method = binder.multibind if binding.is_multibinding() else binder.bind 

842 bind_method( # type: ignore 

843 binding.interface, to=types.MethodType(binding.provider, self), scope=binding.scope 

844 ) 

845 self.configure(binder) 

846 

847 def configure(self, binder: Binder) -> None: 

848 """Override to configure bindings.""" 

849 

850 

851class Injector: 

852 """ 

853 :param modules: Optional - a configuration module or iterable of configuration modules. 

854 Each module will be installed in current :class:`Binder` using :meth:`Binder.install`. 

855 

856 Consult :meth:`Binder.install` documentation for the details. 

857 

858 :param auto_bind: Whether to automatically bind missing types. 

859 :param parent: Parent injector. 

860 

861 .. versionadded:: 0.7.5 

862 ``use_annotations`` parameter 

863 

864 .. versionchanged:: 0.13.0 

865 ``use_annotations`` parameter is removed 

866 """ 

867 

868 def __init__( 

869 self, 

870 modules: Union[_InstallableModuleType, Iterable[_InstallableModuleType]] = None, 

871 auto_bind: bool = True, 

872 parent: 'Injector' = None, 

873 ) -> None: 

874 # Stack of keys currently being injected. Used to detect circular 

875 # dependencies. 

876 self._stack = () # type: Tuple[Tuple[object, Callable, Tuple[Tuple[str, type], ...]], ...] 

877 

878 self.parent = parent 

879 

880 # Binder 

881 self.binder = Binder( 

882 self, auto_bind=auto_bind, parent=parent.binder if parent is not None else None 

883 ) # type: Binder 

884 

885 if not modules: 

886 modules = [] 

887 elif not hasattr(modules, '__iter__'): 887 ↛ 890line 887 didn't jump to line 890, because the condition on line 887 was never false

888 modules = [cast(_InstallableModuleType, modules)] 

889 # This line is needed to pelase mypy. We know we have Iteable of modules here. 

890 modules = cast(Iterable[_InstallableModuleType], modules) 

891 

892 # Bind some useful types 

893 self.binder.bind(Injector, to=self) 

894 self.binder.bind(Binder, to=self.binder) 

895 

896 # Initialise modules 

897 for module in modules: 

898 self.binder.install(module) 

899 

900 @property 

901 def _log_prefix(self) -> str: 

902 return '>' * (len(self._stack) + 1) + ' ' 

903 

904 def get(self, interface: Type[T], scope: Union[ScopeDecorator, Type[Scope]] = None) -> T: 

905 """Get an instance of the given interface. 

906 

907 .. note:: 

908 

909 Although this method is part of :class:`Injector`'s public interface 

910 it's meant to be used in limited set of circumstances. 

911 

912 For example, to create some kind of root object (application object) 

913 of your application (note that only one `get` call is needed, 

914 inside the `Application` class and any of its dependencies 

915 :func:`inject` can and should be used): 

916 

917 .. code-block:: python 

918 

919 class Application: 

920 

921 @inject 

922 def __init__(self, dep1: Dep1, dep2: Dep2): 

923 self.dep1 = dep1 

924 self.dep2 = dep2 

925 

926 def run(self): 

927 self.dep1.something() 

928 

929 injector = Injector(configuration) 

930 application = injector.get(Application) 

931 application.run() 

932 

933 :param interface: Interface whose implementation we want. 

934 :param scope: Class of the Scope in which to resolve. 

935 :returns: An implementation of interface. 

936 """ 

937 binding, binder = self.binder.get_binding(interface) 

938 scope = scope or binding.scope 

939 if isinstance(scope, ScopeDecorator): 939 ↛ 940line 939 didn't jump to line 940, because the condition on line 939 was never true

940 scope = scope.scope 

941 # Fetch the corresponding Scope instance from the Binder. 

942 scope_binding, _ = binder.get_binding(scope) 

943 scope_instance = scope_binding.provider.get(self) 

944 

945 log.debug( 

946 '%sInjector.get(%r, scope=%r) using %r', self._log_prefix, interface, scope, binding.provider 

947 ) 

948 result = scope_instance.get(interface, binding.provider).get(self) 

949 log.debug('%s -> %r', self._log_prefix, result) 

950 return result 

951 

952 def create_child_injector(self, *args: Any, **kwargs: Any) -> 'Injector': 

953 kwargs['parent'] = self 

954 return Injector(*args, **kwargs) 

955 

956 def create_object(self, cls: Type[T], additional_kwargs: Any = None) -> T: 

957 """Create a new instance, satisfying any dependencies on cls.""" 

958 additional_kwargs = additional_kwargs or {} 

959 log.debug('%sCreating %r object with %r', self._log_prefix, cls, additional_kwargs) 

960 

961 try: 

962 instance = cls.__new__(cls) 

963 except TypeError as e: 

964 reraise( 

965 e, 

966 CallError(cls, getattr(cls.__new__, '__func__', cls.__new__), (), {}, e, self._stack), 

967 maximum_frames=2, 

968 ) 

969 try: 

970 # On Python 3.5.3 calling call_with_injection() with object.__init__ would fail further 

971 # down the line as get_type_hints(object.__init__) raises an exception until Python 3.5.4. 

972 # And since object.__init__ doesn't do anything useful and can't have any injectable 

973 # arguments we can skip calling it altogether. See GH-135 for more information. 

974 if cls.__init__ is not object.__init__: 

975 self.call_with_injection(cls.__init__, self_=instance, kwargs=additional_kwargs) 

976 except TypeError as e: 

977 reraise(e, CallError(instance, instance.__init__.__func__, (), additional_kwargs, e, self._stack)) 

978 return instance 

979 

980 def call_with_injection( 

981 self, callable: Callable[..., T], self_: Any = None, args: Any = (), kwargs: Any = {} 

982 ) -> T: 

983 """Call a callable and provide it's dependencies if needed. 

984 

985 :param self_: Instance of a class callable belongs to if it's a method, 

986 None otherwise. 

987 :param args: Arguments to pass to callable. 

988 :param kwargs: Keyword arguments to pass to callable. 

989 :type callable: callable 

990 :type args: tuple of objects 

991 :type kwargs: dict of string -> object 

992 :return: Value returned by callable. 

993 """ 

994 

995 bindings = get_bindings(callable) 

996 signature = inspect.signature(callable) 

997 full_args = args 

998 if self_ is not None: 

999 full_args = (self_,) + full_args 

1000 bound_arguments = signature.bind_partial(*full_args) 

1001 

1002 needed = dict( 

1003 (k, v) for (k, v) in bindings.items() if k not in kwargs and k not in bound_arguments.arguments 

1004 ) 

1005 

1006 dependencies = self.args_to_inject( 

1007 function=callable, 

1008 bindings=needed, 

1009 owner_key=self_.__class__ if self_ is not None else callable.__module__, 

1010 ) 

1011 

1012 dependencies.update(kwargs) 

1013 

1014 try: 

1015 return callable(*full_args, **dependencies) 

1016 except TypeError as e: 

1017 reraise(e, CallError(self_, callable, args, dependencies, e, self._stack)) 

1018 # Needed because of a mypy-related issue (https://github.com/python/mypy/issues/8129). 

1019 assert False, "unreachable" # pragma: no cover 

1020 

1021 @private 

1022 @synchronized(lock) 

1023 def args_to_inject( 

1024 self, function: Callable, bindings: Dict[str, type], owner_key: object 

1025 ) -> Dict[str, Any]: 

1026 """Inject arguments into a function. 

1027 

1028 :param function: The function. 

1029 :param bindings: Map of argument name to binding key to inject. 

1030 :param owner_key: A key uniquely identifying the *scope* of this function. 

1031 For a method this will be the owning class. 

1032 :returns: Dictionary of resolved arguments. 

1033 """ 

1034 dependencies = {} 

1035 

1036 key = (owner_key, function, tuple(sorted(bindings.items()))) 

1037 

1038 def repr_key(k: Tuple[object, Callable, Tuple[Tuple[str, type], ...]]) -> str: 

1039 owner_key, function, bindings = k 

1040 return '%s.%s(injecting %s)' % (tuple(map(_describe, k[:2])) + (dict(k[2]),)) 

1041 

1042 log.debug('%sProviding %r for %r', self._log_prefix, bindings, function) 

1043 

1044 if key in self._stack: 1044 ↛ 1045line 1044 didn't jump to line 1045, because the condition on line 1044 was never true

1045 raise CircularDependency( 

1046 'circular dependency detected: %s -> %s' 

1047 % (' -> '.join(map(repr_key, self._stack)), repr_key(key)) 

1048 ) 

1049 

1050 self._stack += (key,) 

1051 try: 

1052 for arg, interface in bindings.items(): 

1053 try: 

1054 instance = self.get(interface) # type: Any 

1055 except UnsatisfiedRequirement as e: 

1056 if not e.owner: 

1057 e = UnsatisfiedRequirement(owner_key, e.interface) 

1058 raise e 

1059 dependencies[arg] = instance 

1060 finally: 

1061 self._stack = tuple(self._stack[:-1]) 1061 ↛ exitline 1061 didn't except from function 'args_to_inject', because the raise on line 1058 wasn't executed

1062 

1063 return dependencies 

1064 

1065 

1066def get_bindings(callable: Callable) -> Dict[str, type]: 

1067 """Get bindings of injectable parameters from a callable. 

1068 

1069 If the callable is not decorated with :func:`inject` and does not have any of its 

1070 parameters declared as injectable using :data:`Inject` an empty dictionary will 

1071 be returned. Otherwise the returned dictionary will contain a mapping 

1072 between parameter names and their types with the exception of parameters 

1073 excluded from dependency injection (either with :func:`noninjectable`, :data:`NoInject` 

1074 or only explicit injection with :data:`Inject` being used). For example:: 

1075 

1076 >>> def function1(a: int) -> None: 

1077 ... pass 

1078 ... 

1079 >>> get_bindings(function1) 

1080 {} 

1081 

1082 >>> @inject 

1083 ... def function2(a: int) -> None: 

1084 ... pass 

1085 ... 

1086 >>> get_bindings(function2) 

1087 {'a': <class 'int'>} 

1088 

1089 >>> @inject 

1090 ... @noninjectable('b') 

1091 ... def function3(a: int, b: str) -> None: 

1092 ... pass 

1093 ... 

1094 >>> get_bindings(function3) 

1095 {'a': <class 'int'>} 

1096 

1097 >>> import sys, pytest 

1098 >>> if sys.version_info < (3, 7, 0): 

1099 ... pytest.skip('Python 3.7.0 required for sufficient Annotated support') 

1100 

1101 >>> # The simple case of no @inject but injection requested with Inject[...] 

1102 >>> def function4(a: Inject[int], b: str) -> None: 

1103 ... pass 

1104 ... 

1105 >>> get_bindings(function4) 

1106 {'a': <class 'int'>} 

1107 

1108 >>> # Using @inject with Inject is redundant but it should not break anything 

1109 >>> @inject 

1110 ... def function5(a: Inject[int], b: str) -> None: 

1111 ... pass 

1112 ... 

1113 >>> get_bindings(function5) 

1114 {'a': <class 'int'>, 'b': <class 'str'>} 

1115 

1116 >>> # We need to be able to exclude a parameter from injection with NoInject 

1117 >>> @inject 

1118 ... def function6(a: int, b: NoInject[str]) -> None: 

1119 ... pass 

1120 ... 

1121 >>> get_bindings(function6) 

1122 {'a': <class 'int'>} 

1123 

1124 >>> # The presence of NoInject should not trigger anything on its own 

1125 >>> def function7(a: int, b: NoInject[str]) -> None: 

1126 ... pass 

1127 ... 

1128 >>> get_bindings(function7) 

1129 {} 

1130 

1131 This function is used internally so by calling it you can learn what exactly 

1132 Injector is going to try to provide to a callable. 

1133 """ 

1134 look_for_explicit_bindings = False 

1135 if not hasattr(callable, '__bindings__'): 

1136 type_hints = get_type_hints(callable, include_extras=True) 

1137 has_injectable_parameters = any( 

1138 _is_specialization(v, Annotated) and _inject_marker in v.__metadata__ for v in type_hints.values() 

1139 ) 

1140 

1141 if not has_injectable_parameters: 

1142 return {} 

1143 else: 

1144 look_for_explicit_bindings = True 

1145 

1146 if look_for_explicit_bindings or cast(Any, callable).__bindings__ == 'deferred': 

1147 read_and_store_bindings( 

1148 callable, _infer_injected_bindings(callable, only_explicit_bindings=look_for_explicit_bindings) 

1149 ) 

1150 noninjectables = getattr(callable, '__noninjectables__', set()) 

1151 return {k: v for k, v in cast(Any, callable).__bindings__.items() if k not in noninjectables} 

1152 

1153 

1154class _BindingNotYetAvailable(Exception): 

1155 pass 

1156 

1157 

1158def _infer_injected_bindings(callable: Callable, only_explicit_bindings: bool) -> Dict[str, type]: 

1159 spec = inspect.getfullargspec(callable) 

1160 try: 

1161 bindings = get_type_hints(callable, include_extras=True) 

1162 except NameError as e: 

1163 raise _BindingNotYetAvailable(e) 

1164 

1165 # We don't care about the return value annotation as it doesn't matter 

1166 # injection-wise. 

1167 bindings.pop('return', None) 

1168 

1169 # If we're dealing with a bound method get_type_hints will still return `self` annotation even though 

1170 # it's already provided and we're not really interested in its type. So – drop it. 

1171 if isinstance(callable, types.MethodType): 1171 ↛ 1172line 1171 didn't jump to line 1172, because the condition on line 1171 was never true

1172 self_name = spec.args[0] 

1173 bindings.pop(self_name, None) 

1174 

1175 # variadic arguments aren't supported at the moment (this may change 

1176 # in the future if someone has a good idea how to utilize them) 

1177 if spec.varargs: 1177 ↛ 1178line 1177 didn't jump to line 1178, because the condition on line 1177 was never true

1178 bindings.pop(spec.varargs, None) 

1179 if spec.varkw: 1179 ↛ 1180line 1179 didn't jump to line 1180, because the condition on line 1179 was never true

1180 bindings.pop(spec.varkw, None) 

1181 

1182 for k, v in list(bindings.items()): 

1183 if _is_specialization(v, Annotated): 

1184 v, metadata = v.__origin__, v.__metadata__ 

1185 bindings[k] = v 

1186 else: 

1187 metadata = tuple() 

1188 

1189 if only_explicit_bindings and _inject_marker not in metadata or _noinject_marker in metadata: 

1190 del bindings[k] 

1191 break 

1192 

1193 if _is_specialization(v, Union): 1193 ↛ 1195line 1193 didn't jump to line 1195, because the condition on line 1193 was never true

1194 # We don't treat Optional parameters in any special way at the moment. 

1195 if TYPING353: 

1196 union_members = v.__args__ 

1197 else: 

1198 union_members = v.__union_params__ 

1199 new_members = tuple(set(union_members) - {type(None)}) 

1200 # mypy stared complaining about this line for some reason: 

1201 # error: Variable "new_members" is not valid as a type 

1202 new_union = Union[new_members] # type: ignore 

1203 # mypy complains about this construct: 

1204 # error: The type alias is invalid in runtime context 

1205 # See: https://github.com/python/mypy/issues/5354 

1206 bindings[k] = new_union # type: ignore 

1207 

1208 return bindings 

1209 

1210 

1211def provider(function: CallableT) -> CallableT: 

1212 """Decorator for :class:`Module` methods, registering a provider of a type. 

1213 

1214 >>> class MyModule(Module): 

1215 ... @provider 

1216 ... def provide_name(self) -> str: 

1217 ... return 'Bob' 

1218 

1219 @provider-decoration implies @inject so you can omit it and things will 

1220 work just the same: 

1221 

1222 >>> class MyModule2(Module): 

1223 ... def configure(self, binder): 

1224 ... binder.bind(int, to=654) 

1225 ... 

1226 ... @provider 

1227 ... def provide_str(self, i: int) -> str: 

1228 ... return str(i) 

1229 ... 

1230 >>> injector = Injector(MyModule2) 

1231 >>> injector.get(str) 

1232 '654' 

1233 """ 

1234 _mark_provider_function(function, allow_multi=False) 

1235 return function 

1236 

1237 

1238def multiprovider(function: CallableT) -> CallableT: 

1239 """Like :func:`provider`, but for multibindings. Example usage:: 

1240 

1241 class MyModule(Module): 

1242 @multiprovider 

1243 def provide_strs(self) -> List[str]: 

1244 return ['str1'] 

1245 

1246 class OtherModule(Module): 

1247 @multiprovider 

1248 def provide_strs_also(self) -> List[str]: 

1249 return ['str2'] 

1250 

1251 Injector([MyModule, OtherModule]).get(List[str]) # ['str1', 'str2'] 

1252 

1253 See also: :meth:`Binder.multibind`.""" 

1254 _mark_provider_function(function, allow_multi=True) 

1255 return function 

1256 

1257 

1258def _mark_provider_function(function: Callable, *, allow_multi: bool) -> None: 

1259 scope_ = getattr(function, '__scope__', None) 

1260 try: 

1261 annotations = get_type_hints(function) 

1262 except NameError: 

1263 return_type = '__deferred__' 

1264 else: 

1265 return_type = annotations['return'] 

1266 _validate_provider_return_type(function, cast(type, return_type), allow_multi) 

1267 function.__binding__ = Binding(return_type, inject(function), scope_) # type: ignore 

1268 

1269 

1270def _validate_provider_return_type(function: Callable, return_type: type, allow_multi: bool) -> None: 

1271 origin = _get_origin(_punch_through_alias(return_type)) 

1272 if origin in {dict, list} and not allow_multi: 1272 ↛ 1273line 1272 didn't jump to line 1273, because the condition on line 1272 was never true

1273 raise Error( 

1274 'Function %s needs to be decorated with multiprovider instead of provider if it is to ' 

1275 'provide values to a multibinding of type %s' % (function.__name__, return_type) 

1276 ) 

1277 

1278 

1279ConstructorOrClassT = TypeVar('ConstructorOrClassT', bound=Union[Callable, Type]) 

1280 

1281 

1282@overload 

1283def inject(constructor_or_class: CallableT) -> CallableT: # pragma: no cover 

1284 pass 

1285 

1286 

1287@overload 

1288def inject(constructor_or_class: Type[T]) -> Type[T]: # pragma: no cover 

1289 pass 

1290 

1291 

1292def inject(constructor_or_class: ConstructorOrClassT) -> ConstructorOrClassT: 

1293 """Decorator declaring parameters to be injected. 

1294 

1295 eg. 

1296 

1297 >>> class A: 

1298 ... @inject 

1299 ... def __init__(self, number: int, name: str): 

1300 ... print([number, name]) 

1301 ... 

1302 >>> def configure(binder): 

1303 ... binder.bind(A) 

1304 ... binder.bind(int, to=123) 

1305 ... binder.bind(str, to='Bob') 

1306 

1307 Use the Injector to get a new instance of A: 

1308 

1309 >>> a = Injector(configure).get(A) 

1310 [123, 'Bob'] 

1311 

1312 As a convenience one can decorate a class itself:: 

1313 

1314 @inject 

1315 class B: 

1316 def __init__(self, dependency: Dependency): 

1317 self.dependency = dependency 

1318 

1319 This is equivalent to decorating its constructor. In particular this provides integration with 

1320 `dataclasses <https://docs.python.org/3/library/dataclasses.html>`_ (the order of decorator 

1321 application is important here):: 

1322 

1323 @inject 

1324 @dataclass 

1325 class C: 

1326 dependency: Dependency 

1327 

1328 .. note:: 

1329 

1330 This decorator is to be used on class constructors (or, as a convenience, on classes). 

1331 Using it on non-constructor methods worked in the past but it was an implementation 

1332 detail rather than a design decision. 

1333 

1334 Third party libraries may, however, provide support for injecting dependencies 

1335 into non-constructor methods or free functions in one form or another. 

1336 

1337 .. seealso:: 

1338 

1339 Generic type :data:`Inject` 

1340 A more explicit way to declare parameters as injectable. 

1341 

1342 Function :func:`get_bindings` 

1343 A way to inspect how various injection declarations interact with each other. 

1344 

1345 .. versionchanged:: 0.16.2 

1346 

1347 (Re)added support for decorating classes with @inject. 

1348 """ 

1349 if isinstance(constructor_or_class, type) and hasattr(constructor_or_class, '__init__'): 1349 ↛ 1350line 1349 didn't jump to line 1350, because the condition on line 1349 was never true

1350 inject(cast(Any, constructor_or_class).__init__) 

1351 else: 

1352 function = constructor_or_class 

1353 try: 

1354 bindings = _infer_injected_bindings(function, only_explicit_bindings=False) 

1355 read_and_store_bindings(function, bindings) 

1356 except _BindingNotYetAvailable: 

1357 cast(Any, function).__bindings__ = 'deferred' 

1358 return constructor_or_class 

1359 

1360 

1361def noninjectable(*args: str) -> Callable[[CallableT], CallableT]: 

1362 """Mark some parameters as not injectable. 

1363 

1364 This serves as documentation for people reading the code and will prevent 

1365 Injector from ever attempting to provide the parameters. 

1366 

1367 For example: 

1368 

1369 >>> class Service: 

1370 ... pass 

1371 ... 

1372 >>> class SomeClass: 

1373 ... @inject 

1374 ... @noninjectable('user_id') 

1375 ... def __init__(self, service: Service, user_id: int): 

1376 ... # ... 

1377 ... pass 

1378 

1379 :func:`noninjectable` decorations can be stacked on top of 

1380 each other and the order in which a function is decorated with 

1381 :func:`inject` and :func:`noninjectable` 

1382 doesn't matter. 

1383 

1384 .. seealso:: 

1385 

1386 Generic type :data:`NoInject` 

1387 A nicer way to declare parameters as noninjectable. 

1388 

1389 Function :func:`get_bindings` 

1390 A way to inspect how various injection declarations interact with each other. 

1391 

1392 """ 

1393 

1394 def decorator(function: CallableT) -> CallableT: 

1395 argspec = inspect.getfullargspec(inspect.unwrap(function)) 

1396 for arg in args: 

1397 if arg not in argspec.args and arg not in argspec.kwonlyargs: 1397 ↛ 1398line 1397 didn't jump to line 1398, because the condition on line 1397 was never true

1398 raise UnknownArgument('Unable to mark unknown argument %s ' 'as non-injectable.' % arg) 

1399 

1400 existing = getattr(function, '__noninjectables__', set()) 

1401 merged = existing | set(args) 

1402 cast(Any, function).__noninjectables__ = merged 

1403 return function 

1404 

1405 return decorator 

1406 

1407 

1408@private 

1409def read_and_store_bindings(f: Callable, bindings: Dict[str, type]) -> None: 

1410 function_bindings = getattr(f, '__bindings__', None) or {} 

1411 if function_bindings == 'deferred': 1411 ↛ 1412line 1411 didn't jump to line 1412, because the condition on line 1411 was never true

1412 function_bindings = {} 

1413 merged_bindings = dict(function_bindings, **bindings) 

1414 

1415 if hasattr(f, '__func__'): 1415 ↛ 1416line 1415 didn't jump to line 1416, because the condition on line 1415 was never true

1416 f = cast(Any, f).__func__ 

1417 cast(Any, f).__bindings__ = merged_bindings 

1418 

1419 

1420class BoundKey(tuple): 

1421 """A BoundKey provides a key to a type with pre-injected arguments. 

1422 

1423 >>> class A: 

1424 ... def __init__(self, a, b): 

1425 ... self.a = a 

1426 ... self.b = b 

1427 >>> InjectedA = BoundKey(A, a=InstanceProvider(1), b=InstanceProvider(2)) 

1428 >>> injector = Injector() 

1429 >>> a = injector.get(InjectedA) 

1430 >>> a.a, a.b 

1431 (1, 2) 

1432 """ 

1433 

1434 def __new__(cls, interface: Type[T], **kwargs: Any) -> 'BoundKey': 

1435 kwargs_tuple = tuple(sorted(kwargs.items())) 

1436 return super(BoundKey, cls).__new__(cls, (interface, kwargs_tuple)) # type: ignore 

1437 

1438 @property 

1439 def interface(self) -> Type[T]: 

1440 return self[0] 

1441 

1442 @property 

1443 def kwargs(self) -> Dict[str, Any]: 

1444 return dict(self[1]) 

1445 

1446 

1447class AssistedBuilder(Generic[T]): 

1448 def __init__(self, injector: Injector, target: Type[T]) -> None: 

1449 self._injector = injector 

1450 self._target = target 

1451 

1452 def build(self, **kwargs: Any) -> T: 

1453 binder = self._injector.binder 

1454 binding, _ = binder.get_binding(self._target) 

1455 provider = binding.provider 

1456 if not isinstance(provider, ClassProvider): 

1457 raise Error( 

1458 'Assisted interface building works only with ClassProviders, ' 

1459 'got %r for %r' % (provider, binding.interface) 

1460 ) 

1461 

1462 return self._build_class(cast(Type[T], provider._cls), **kwargs) 

1463 

1464 def _build_class(self, cls: Type[T], **kwargs: Any) -> T: 

1465 return self._injector.create_object(cls, additional_kwargs=kwargs) 

1466 

1467 

1468class ClassAssistedBuilder(AssistedBuilder[T]): 

1469 def build(self, **kwargs: Any) -> T: 

1470 return self._build_class(self._target, **kwargs) 

1471 

1472 

1473def _describe(c: Any) -> str: 

1474 if hasattr(c, '__name__'): 

1475 return cast(str, c.__name__) 

1476 if type(c) in (tuple, list): 

1477 return '[%s]' % c[0].__name__ 

1478 return str(c) 

1479 

1480 

1481class ProviderOf(Generic[T]): 

1482 """Can be used to get a provider of an interface, for example: 

1483 

1484 >>> def provide_int(): 

1485 ... print('providing') 

1486 ... return 123 

1487 >>> 

1488 >>> def configure(binder): 

1489 ... binder.bind(int, to=provide_int) 

1490 >>> 

1491 >>> injector = Injector(configure) 

1492 >>> provider = injector.get(ProviderOf[int]) 

1493 >>> value = provider.get() 

1494 providing 

1495 >>> value 

1496 123 

1497 """ 

1498 

1499 def __init__(self, injector: Injector, interface: Type[T]): 

1500 self._injector = injector 

1501 self._interface = interface 

1502 

1503 def __repr__(self) -> str: 

1504 return '%s(%r, %r)' % (type(self).__name__, self._injector, self._interface) 

1505 

1506 def get(self) -> T: 

1507 """Get an implementation for the specified interface.""" 

1508 return self._injector.get(self._interface) 

1509 

1510 

1511def is_decorated_with_inject(function: Callable[..., Any]) -> bool: 

1512 """See if given callable is declared to want some dependencies injected. 

1513 

1514 Example use: 

1515 

1516 >>> def fun(i: int) -> str: 

1517 ... return str(i) 

1518 

1519 >>> is_decorated_with_inject(fun) 

1520 False 

1521 >>> 

1522 >>> @inject 

1523 ... def fun2(i: int) -> str: 

1524 ... return str(i) 

1525 

1526 >>> is_decorated_with_inject(fun2) 

1527 True 

1528 """ 

1529 return hasattr(function, '__bindings__')