(
self,
store_uri: str | PathLike[str],
download_func: None = None,
*,
crawler: Crawler,
)
| 454 | DEFAULT_FILES_RESULT_FIELD: str = "files" |
| 455 | |
| 456 | def __init__( |
| 457 | self, |
| 458 | store_uri: str | PathLike[str], |
| 459 | download_func: None = None, |
| 460 | *, |
| 461 | crawler: Crawler, |
| 462 | ): |
| 463 | if download_func is not None: # pragma: no cover |
| 464 | warnings.warn( |
| 465 | "The download_func argument of FilesPipeline.__init__() is ignored" |
| 466 | " and will be removed in a future Scrapy version.", |
| 467 | category=ScrapyDeprecationWarning, |
| 468 | stacklevel=2, |
| 469 | ) |
| 470 | |
| 471 | if not (store_uri and (store_uri := _to_string(store_uri))): |
| 472 | from scrapy.pipelines.images import ImagesPipeline # noqa: PLC0415 |
| 473 | |
| 474 | setting_name = ( |
| 475 | "IMAGES_STORE" if isinstance(self, ImagesPipeline) else "FILES_STORE" |
| 476 | ) |
| 477 | raise NotConfigured( |
| 478 | f"{setting_name} setting must be set to a valid path (not empty) " |
| 479 | f"to enable {self.__class__.__name__}." |
| 480 | ) |
| 481 | |
| 482 | settings = crawler.settings |
| 483 | cls_name = "FilesPipeline" |
| 484 | self.store: FilesStoreProtocol = self._get_store(store_uri) |
| 485 | resolve = functools.partial( |
| 486 | self._key_for_pipe, base_class_name=cls_name, settings=settings |
| 487 | ) |
| 488 | self.expires: int = settings.getint(resolve("FILES_EXPIRES"), self.EXPIRES) |
| 489 | if not hasattr(self, "FILES_URLS_FIELD"): |
| 490 | self.FILES_URLS_FIELD = self.DEFAULT_FILES_URLS_FIELD |
| 491 | if not hasattr(self, "FILES_RESULT_FIELD"): |
| 492 | self.FILES_RESULT_FIELD = self.DEFAULT_FILES_RESULT_FIELD |
| 493 | self.files_urls_field: str = settings.get( |
| 494 | resolve("FILES_URLS_FIELD"), self.FILES_URLS_FIELD |
| 495 | ) |
| 496 | self.files_result_field: str = settings.get( |
| 497 | resolve("FILES_RESULT_FIELD"), self.FILES_RESULT_FIELD |
| 498 | ) |
| 499 | |
| 500 | super().__init__(crawler=crawler) |
| 501 | |
| 502 | @classmethod |
| 503 | def from_crawler(cls, crawler: Crawler) -> Self: |
nothing calls this directly
no test coverage detected