meadsteve /
lagom
| @@ 135-156 (lines=22) @@ | ||
| 132 | ||
| 133 | return _with_context |
|
| 134 | ||
| 135 | def magic_partial( |
|
| 136 | self, |
|
| 137 | func: Callable[..., X], |
|
| 138 | shared: Optional[List[Type]] = None, |
|
| 139 | keys_to_skip: Optional[List[str]] = None, |
|
| 140 | skip_pos_up_to: int = 0, |
|
| 141 | container_updater: Optional[CallTimeContainerUpdate] = None, |
|
| 142 | ) -> Callable[..., X]: |
|
| 143 | if not inspect.iscoroutinefunction(func): |
|
| 144 | raise MissingFeature( |
|
| 145 | "AsyncContextManager currently can only deal with async functions" |
|
| 146 | ) |
|
| 147 | ||
| 148 | async def _with_context(*args, **kwargs): |
|
| 149 | async with self as c: |
|
| 150 | # TODO: Try and move this partial outside the function as this is expensive |
|
| 151 | base_partial = super(AsyncContextContainer, c).magic_partial( |
|
| 152 | func, shared, keys_to_skip, skip_pos_up_to, container_updater |
|
| 153 | ) |
|
| 154 | return await base_partial(*args, **kwargs) # type: ignore |
|
| 155 | ||
| 156 | return _with_context |
|
| 157 | ||
| 158 | def _context_type_def(self, dep_type: Type): |
|
| 159 | type_def = self.get_definition(ContextManager[dep_type]) or self.get_definition(Iterator[dep_type]) or self.get_definition(Generator[dep_type, None, None]) or self.get_definition(AsyncGenerator[dep_type, None]) or self.get_definition(AsyncContextManager[dep_type]) # type: ignore |
|
| @@ 115-131 (lines=17) @@ | ||
| 112 | ||
| 113 | return _with_context |
|
| 114 | ||
| 115 | def magic_partial( |
|
| 116 | self, |
|
| 117 | func: Callable[..., X], |
|
| 118 | shared: Optional[List[Type]] = None, |
|
| 119 | keys_to_skip: Optional[List[str]] = None, |
|
| 120 | skip_pos_up_to: int = 0, |
|
| 121 | container_updater: Optional[CallTimeContainerUpdate] = None, |
|
| 122 | ) -> Callable[..., X]: |
|
| 123 | def _with_context(*args, **kwargs): |
|
| 124 | with self as c: |
|
| 125 | # TODO: Try and move this partial outside the function as this is expensive |
|
| 126 | base_partial = super(ContextContainer, c).magic_partial( |
|
| 127 | func, shared, keys_to_skip, skip_pos_up_to, container_updater |
|
| 128 | ) |
|
| 129 | return base_partial(*args, **kwargs) |
|
| 130 | ||
| 131 | return _with_context |
|
| 132 | ||
| 133 | def _context_type_def(self, dep_type: Type): |
|
| 134 | type_def = self.get_definition(ContextManager[dep_type]) or self.get_definition(Iterator[dep_type]) or self.get_definition(Generator[dep_type, None, None]) # type: ignore |
|
