(
self,
request_or_url: Request | str,
spider: Spider | None = None,
redirect: bool = True,
**kwargs: Any,
)
| 218 | self.spider = spider |
| 219 | |
| 220 | def fetch( |
| 221 | self, |
| 222 | request_or_url: Request | str, |
| 223 | spider: Spider | None = None, |
| 224 | redirect: bool = True, |
| 225 | **kwargs: Any, |
| 226 | ) -> None: |
| 227 | if isinstance(request_or_url, Request): |
| 228 | request = request_or_url |
| 229 | else: |
| 230 | url = any_to_uri(request_or_url) |
| 231 | request = Request(url, dont_filter=True, **kwargs) |
| 232 | if redirect: |
| 233 | request.meta["handle_httpstatus_list"] = SequenceExclude( |
| 234 | range(300, 400) |
| 235 | ) |
| 236 | else: |
| 237 | request.meta["handle_httpstatus_all"] = True |
| 238 | response: Response | None = None |
| 239 | if self._use_reactor: |
| 240 | from twisted.internet import reactor |
| 241 | |
| 242 | with contextlib.suppress(IgnoreRequest): |
| 243 | response = threads.blockingCallFromThread( |
| 244 | reactor, |
| 245 | deferred_f_from_coro_f(self._schedule), # type: ignore[arg-type] |
| 246 | request, |
| 247 | spider, |
| 248 | ) |
| 249 | else: |
| 250 | assert self._loop |
| 251 | with contextlib.suppress(IgnoreRequest): |
| 252 | future = asyncio.run_coroutine_threadsafe( |
| 253 | self._schedule(request, spider), self._loop |
| 254 | ) |
| 255 | response = future.result() |
| 256 | self.populate_vars(response, request, self.spider) |
| 257 | |
| 258 | def populate_vars( |
| 259 | self, |
no test coverage detected