RomanNum3ral
/
scrapy
mirror of https://github.com/scrapy/scrapy.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Make media pipeline storages configurable components

This commit is contained in:
Adrian Chaves 2026-08-07 03:01:19 +02:00
parent 1bd839b57d
commit c99a7b695d
7 changed files with 303 additions and 133 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
docs/topics/media-pipeline.rst
View File

@ -315,6 +315,50 @@ For more information, see `Predefined ACLs`_ in the Google Cloud Platform Develo
.. _Predefined ACLs: https://docs.cloud.google.com/storage/docs/access-control/lists#predefined-acl
.. _media-pipeline-custom-storage:
Custom storage
--------------
.. setting:: MEDIA_STORAGES
.. versionadded:: VERSION
The ``MEDIA_STORAGES`` setting maps URL schemes to storage classes, so that you
can support additional schemes or replace a built-in storage:
.. code-block:: python
MEDIA_STORAGES = {"myscheme": "myproject.storages.MyStorage"}
FILES_STORE = "myscheme://example.com/files/"
It is merged into ``MEDIA_STORAGES_BASE``, which holds the built-in storages.
A storage class receives the :setting:`FILES_STORE` or :setting:`IMAGES_STORE`
value, and must define ``stat_file`` and ``persist_file``:
.. code-block:: python
class MyStorage:
def __init__(self, uri, *, acl=None):
self.uri = uri
self.acl = acl
@classmethod
def from_crawler(cls, crawler, uri, *, resolve):
return cls(uri, acl=crawler.settings[resolve("FILES_STORE_ACL")])
def stat_file(self, path, info): ...
def persist_file(self, path, buf, info, meta=None, headers=None): ...
*resolve* maps a setting name to the name that applies to the pipeline being
built, so that a single storage class can be configured separately for files and
for images: ``resolve("FILES_STORE_ACL")`` returns ``"FILES_STORE_ACL"`` for
:class:`~scrapy.pipelines.files.FilesPipeline` and ``"IMAGES_STORE_ACL"`` for
:class:`~scrapy.pipelines.images.ImagesPipeline`, and takes :ref:`per-class
setting names <media-pipeline-class-settings>` into account.
Usage example
=============
@ -369,6 +413,8 @@ For the Images Pipeline, set :setting:`IMAGES_URLS_FIELD` and/or
If you need something more complex and want to override the custom pipeline
behaviour, see :ref:`topics-media-pipeline-override`.
.. _media-pipeline-class-settings:
If you have multiple image pipelines inheriting from :class:`ImagesPipeline`
and you want to have different settings in different pipelines you can set
setting keys preceded with uppercase name of your pipeline class. E.g. if your

136
scrapy/pipelines/files.py
View File

@ -43,11 +43,12 @@ from scrapy.utils.defer import deferred_from_coro, ensure_awaitable
from scrapy.utils.ftp import ftp_store_file
from scrapy.utils.httpobj import urlparse_cached
from scrapy.utils.log import failure_to_exc_info
from scrapy.utils.python import to_bytes
from scrapy.utils.misc import load_object
from scrapy.utils.python import to_bytes, without_none_values
from scrapy.utils.request import referer_str
if TYPE_CHECKING:
from collections.abc import Awaitable
from collections.abc import Awaitable, Callable
from os import PathLike
from twisted.python.failure import Failure
@ -56,7 +57,6 @@ if TYPE_CHECKING:
from typing_extensions import Self
from scrapy.crawler import Crawler
from scrapy.settings import BaseSettings
logger = logging.getLogger(__name__)
@ -89,7 +89,7 @@ class StatInfo(TypedDict, total=False):
class FilesStoreProtocol(Protocol):
def __init__(self, basedir: str): ...
def __init__(self, uri: str): ...
def persist_file(
self,
@ -164,22 +164,40 @@ class S3FilesStore:
AWS_REGION_NAME = None
AWS_USE_SSL = None
AWS_VERIFY = None
# Overridden from settings.AWS_MAX_POOL_CONNECTIONS in
# FilesPipeline.from_crawler(); None means the botocore default
# None means the botocore default
AWS_MAX_POOL_CONNECTIONS: int | None = None
POLICY = "private" # Overridden from settings.FILES_STORE_S3_ACL in FilesPipeline.from_crawler()
POLICY = "private"
HEADERS: ClassVar[dict[str, str]] = {
"Cache-Control": "max-age=172800",
}
def __init__(self, uri: str):
@classmethod
def from_crawler(
cls, crawler: Crawler, uri: str, *, resolve: Callable[[str], str]
) -> Self:
settings = crawler.settings
return cls(
uri,
AWS_ACCESS_KEY_ID=settings["AWS_ACCESS_KEY_ID"],
AWS_SECRET_ACCESS_KEY=settings["AWS_SECRET_ACCESS_KEY"],
AWS_SESSION_TOKEN=settings["AWS_SESSION_TOKEN"],
AWS_ENDPOINT_URL=settings["AWS_ENDPOINT_URL"],
AWS_REGION_NAME=settings["AWS_REGION_NAME"],
AWS_USE_SSL=settings["AWS_USE_SSL"],
AWS_VERIFY=settings["AWS_VERIFY"],
AWS_MAX_POOL_CONNECTIONS=_get_max_pool_connections(settings),
POLICY=settings[resolve("FILES_STORE_S3_ACL")],
)
def __init__(self, uri: str, **config: Any):
self.__dict__.update(config)
if not is_botocore_available():
raise NotConfigured("missing botocore library")
import botocore.session # noqa: PLC0415
from botocore.config import Config # noqa: PLC0415
config = (
botocore_config = (
Config(max_pool_connections=self.AWS_MAX_POOL_CONNECTIONS)
if self.AWS_MAX_POOL_CONNECTIONS is not None
else None
@ -194,7 +212,7 @@ class S3FilesStore:
region_name=self.AWS_REGION_NAME,
use_ssl=self.AWS_USE_SSL,
verify=self.AWS_VERIFY,
config=config,
config=botocore_config,
)
if not uri.startswith("s3://"):
raise ValueError(f"Incorrect URI scheme in {uri}, expected 's3'")
@ -299,10 +317,21 @@ class GCSFilesStore:
CACHE_CONTROL = "max-age=172800"
# The bucket's default object ACL will be applied to the object.
# Overridden from settings.FILES_STORE_GCS_ACL in FilesPipeline.from_crawler().
POLICY = None
def __init__(self, uri: str):
@classmethod
def from_crawler(
cls, crawler: Crawler, uri: str, *, resolve: Callable[[str], str]
) -> Self:
settings = crawler.settings
return cls(
uri,
GCS_PROJECT_ID=settings["GCS_PROJECT_ID"],
POLICY=settings[resolve("FILES_STORE_GCS_ACL")] or None,
)
def __init__(self, uri: str, **config: Any):
self.__dict__.update(config)
from google.cloud import storage # noqa: PLC0415
client = storage.Client(project=self.GCS_PROJECT_ID)
@ -377,7 +406,20 @@ class FTPFilesStore:
FTP_PASSWORD: str | None = None
USE_ACTIVE_MODE: bool | None = None
def __init__(self, uri: str):
@classmethod
def from_crawler(
cls, crawler: Crawler, uri: str, *, resolve: Callable[[str], str]
) -> Self:
settings = crawler.settings
return cls(
uri,
FTP_USERNAME=settings["FTP_USER"],
FTP_PASSWORD=settings["FTP_PASSWORD"],
USE_ACTIVE_MODE=settings.getbool("FEED_STORAGE_FTP_ACTIVE"),
)
def __init__(self, uri: str, **config: Any):
self.__dict__.update(config)
if not uri.startswith("ftp://"):
raise ValueError(f"Incorrect URI scheme in {uri}, expected 'ftp'")
u = urlparse(uri)
@ -457,13 +499,7 @@ class FilesPipeline(MediaPipeline):
MEDIA_NAME: str = "file"
EXPIRES: int = 90
STORE_SCHEMES: ClassVar[dict[str, type[FilesStoreProtocol]]] = {
"": FSFilesStore,
"file": FSFilesStore,
"s3": S3FilesStore,
"gs": GCSFilesStore,
"ftp": FTPFilesStore,
}
STORE_SCHEMES: ClassVar[dict[str, type[FilesStoreProtocol]]] = {}
DEFAULT_FILES_URLS_FIELD: str = "file_urls"
DEFAULT_FILES_RESULT_FIELD: str = "files"
@ -493,8 +529,21 @@ class FilesPipeline(MediaPipeline):
f"to enable {self.__class__.__name__}."
)
super().__init__(crawler=crawler)
settings = crawler.settings
cls_name = "FilesPipeline"
self._storages: dict[str, Any] = without_none_values(
settings.getwithbase("MEDIA_STORAGES")
)
if self.STORE_SCHEMES:
warnings.warn(
f"{type(self).__name__} defines STORE_SCHEMES, which is"
" deprecated. Use the MEDIA_STORAGES setting instead.",
category=ScrapyDeprecationWarning,
stacklevel=2,
)
self._storages.update(self.STORE_SCHEMES)
self.store: FilesStoreProtocol = self._get_store(store_uri)
resolve = functools.partial(
self._key_for_pipe, base_class_name=cls_name, settings=settings
@ -511,48 +560,25 @@ class FilesPipeline(MediaPipeline):
resolve("FILES_RESULT_FIELD"), self.FILES_RESULT_FIELD
)
super().__init__(crawler=crawler)
@classmethod
def from_crawler(cls, crawler: Crawler) -> Self:
settings = crawler.settings
cls._update_stores(settings)
store_uri = settings["FILES_STORE"]
return cls(store_uri, crawler=crawler)
return cls(crawler.settings["FILES_STORE"], crawler=crawler)
@classmethod
def _update_stores(cls, settings: BaseSettings) -> None:
s3store: type[S3FilesStore] = cast(
"type[S3FilesStore]", cls.STORE_SCHEMES["s3"]
)
s3store.AWS_ACCESS_KEY_ID = settings["AWS_ACCESS_KEY_ID"]
s3store.AWS_SECRET_ACCESS_KEY = settings["AWS_SECRET_ACCESS_KEY"]
s3store.AWS_SESSION_TOKEN = settings["AWS_SESSION_TOKEN"]
s3store.AWS_ENDPOINT_URL = settings["AWS_ENDPOINT_URL"]
s3store.AWS_REGION_NAME = settings["AWS_REGION_NAME"]
s3store.AWS_USE_SSL = settings["AWS_USE_SSL"]
s3store.AWS_VERIFY = settings["AWS_VERIFY"]
s3store.AWS_MAX_POOL_CONNECTIONS = _get_max_pool_connections(settings)
s3store.POLICY = settings["FILES_STORE_S3_ACL"]
gcs_store: type[GCSFilesStore] = cast(
"type[GCSFilesStore]", cls.STORE_SCHEMES["gs"]
)
gcs_store.GCS_PROJECT_ID = settings["GCS_PROJECT_ID"]
gcs_store.POLICY = settings["FILES_STORE_GCS_ACL"] or None
ftp_store: type[FTPFilesStore] = cast(
"type[FTPFilesStore]", cls.STORE_SCHEMES["ftp"]
)
ftp_store.FTP_USERNAME = settings["FTP_USER"]
ftp_store.FTP_PASSWORD = settings["FTP_PASSWORD"]
ftp_store.USE_ACTIVE_MODE = settings.getbool("FEED_STORAGE_FTP_ACTIVE")
def _resolve_store_setting(self, setting: str) -> str:
return self._key_for_pipe(setting, "FilesPipeline", self.crawler.settings)
def _get_store(self, uri: str) -> FilesStoreProtocol:
# to support win32 paths like: C:\\some\dir
scheme = "file" if Path(uri).is_absolute() else urlparse(uri).scheme
store_cls = self.STORE_SCHEMES[scheme]
return store_cls(uri)
store_cls = load_object(self._storages[scheme])
if hasattr(store_cls, "from_crawler"):
return cast(
"FilesStoreProtocol",
store_cls.from_crawler(
self.crawler, uri, resolve=self._resolve_store_setting
),
)
return cast("FilesStoreProtocol", store_cls(uri))
def _onsuccess(
self,

37
scrapy/pipelines/images.py
View File

@ -11,20 +11,14 @@ import hashlib
import warnings
from contextlib import suppress
from io import BytesIO
from typing import TYPE_CHECKING, Any, ClassVar, cast
from typing import TYPE_CHECKING, Any, ClassVar
from itemadapter import ItemAdapter
from scrapy.exceptions import NotConfigured, ScrapyDeprecationWarning
from scrapy.http import Request, Response
from scrapy.http.request import NO_CALLBACK
from scrapy.pipelines.files import (
FileException,
FilesPipeline,
GCSFilesStore,
S3FilesStore,
_md5sum,
)
from scrapy.pipelines.files import FileException, FilesPipeline, _md5sum
from scrapy.utils.defer import ensure_awaitable
from scrapy.utils.python import to_bytes
@ -39,7 +33,6 @@ if TYPE_CHECKING:
from scrapy.crawler import Crawler
from scrapy.pipelines.media import FileInfoOrError, MediaPipeline
from scrapy.settings import BaseSettings
class ImageException(FileException):
@ -118,10 +111,14 @@ class ImagesPipeline(FilesPipeline):
@classmethod
def from_crawler(cls, crawler: Crawler) -> Self:
settings = crawler.settings
cls._update_stores(settings)
store_uri = settings["IMAGES_STORE"]
return cls(store_uri, crawler=crawler)
return cls(crawler.settings["IMAGES_STORE"], crawler=crawler)
def _resolve_store_setting(self, setting: str) -> str:
return self._key_for_pipe(
setting.replace("FILES_", "IMAGES_", 1),
"ImagesPipeline",
self.crawler.settings,
)
async def file_downloaded(
self,
@ -133,20 +130,6 @@ class ImagesPipeline(FilesPipeline):
) -> str:
return await self.image_downloaded(response, request, info, item=item)
@classmethod
def _update_stores(cls, settings: BaseSettings) -> None:
super()._update_stores(settings)
s3store: type[S3FilesStore] = cast(
"type[S3FilesStore]", cls.STORE_SCHEMES["s3"]
)
s3store.POLICY = settings["IMAGES_STORE_S3_ACL"]
gcs_store: type[GCSFilesStore] = cast(
"type[GCSFilesStore]", cls.STORE_SCHEMES["gs"]
)
gcs_store.POLICY = settings["IMAGES_STORE_GCS_ACL"] or None
async def image_downloaded(
self,
response: Response,

11
scrapy/settings/default_settings.py
View File

@ -154,6 +154,8 @@ __all__ = [
"MAIL_SSL",
"MAIL_TLS",
"MAIL_USER",
"MEDIA_STORAGES",
"MEDIA_STORAGES_BASE",
"MEMDEBUG_ENABLED",
"MEMUSAGE_CHECK_INTERVAL_SECONDS",
"MEMUSAGE_ENABLED",
@ -470,6 +472,15 @@ MAIL_PASS = None
MAIL_SSL = False
MAIL_TLS = False
MEDIA_STORAGES = {}
MEDIA_STORAGES_BASE = {
"": "scrapy.pipelines.files.FSFilesStore",
"file": "scrapy.pipelines.files.FSFilesStore",
"ftp": "scrapy.pipelines.files.FTPFilesStore",
"gs": "scrapy.pipelines.files.GCSFilesStore",
"s3": "scrapy.pipelines.files.S3FilesStore",
}
MEMDEBUG_ENABLED = False # enable memory debugging
MEMUSAGE_ENABLED = True

113
tests/test_pipeline_files.py
View File

@ -5,6 +5,7 @@ import random
import re
import time
from abc import ABC, abstractmethod
from collections.abc import Callable
from datetime import datetime
from ftplib import FTP
from io import BytesIO
@ -22,7 +23,8 @@ from itemadapter import ItemAdapter
from twisted.internet.defer import Deferred
from twisted.python.failure import Failure
from scrapy.exceptions import IgnoreRequest, NotConfigured
from scrapy.crawler import Crawler
from scrapy.exceptions import IgnoreRequest, NotConfigured, ScrapyDeprecationWarning
from scrapy.http import Request, Response
from scrapy.item import Field, Item
from scrapy.pipelines.files import (
@ -749,6 +751,73 @@ class TestFilesPipelineCustomSettings:
assert pipeline.file_path(request) == Path("subdir/image01.jpg")
class SimpleStore:
def __init__(self, uri: str) -> None:
self.uri = uri
class CrawlerAwareStore:
def __init__(self, uri: str, crawler: Crawler, acl_setting: str) -> None:
self.uri = uri
self.crawler = crawler
self.acl_setting = acl_setting
@classmethod
def from_crawler(
cls, crawler: Crawler, uri: str, *, resolve: Callable[[str], str]
) -> "CrawlerAwareStore":
return cls(uri, crawler, resolve("FILES_STORE_S3_ACL"))
class TestMediaStorages:
def test_custom_scheme(self, tmp_path: Path) -> None:
crawler = get_crawler(
None,
{
"FILES_STORE": "mystore://example.com/",
"MEDIA_STORAGES": {
"mystore": "tests.test_pipeline_files.CrawlerAwareStore"
},
},
)
store = FilesPipeline.from_crawler(crawler).store
assert isinstance(store, CrawlerAwareStore)
assert store.uri == "mystore://example.com/"
assert store.crawler is crawler
assert store.acl_setting == "FILES_STORE_S3_ACL"
def test_custom_scheme_without_from_crawler(self, tmp_path: Path) -> None:
crawler = get_crawler(
None,
{
"FILES_STORE": "mystore://example.com/",
"MEDIA_STORAGES": {"mystore": "tests.test_pipeline_files.SimpleStore"},
},
)
store = FilesPipeline.from_crawler(crawler).store
assert isinstance(store, SimpleStore)
assert store.uri == "mystore://example.com/"
def test_override_builtin_scheme(self, tmp_path: Path) -> None:
crawler = get_crawler(
None,
{
"FILES_STORE": str(tmp_path),
"MEDIA_STORAGES": {"file": "tests.test_pipeline_files.SimpleStore"},
},
)
assert isinstance(FilesPipeline.from_crawler(crawler).store, SimpleStore)
def test_store_schemes_deprecated(self, tmp_path: Path) -> None:
class DeprecatedPipeline(FilesPipeline):
STORE_SCHEMES = {**FilesPipeline.STORE_SCHEMES, "file": SimpleStore} # type: ignore[dict-item]
crawler = get_crawler(None, {"FILES_STORE": str(tmp_path)})
with pytest.warns(ScrapyDeprecationWarning, match="STORE_SCHEMES"):
store = DeprecatedPipeline.from_crawler(crawler).store
assert isinstance(store, SimpleStore)
class TestFSFilesStore:
def test_constructor_with_pathlike_object(self, tmp_path: Path) -> None:
assert FSFilesStore(tmp_path).basedir == str(tmp_path)
@ -910,10 +979,8 @@ class TestS3FilesStore:
],
)
def test_max_pool_connections(
self, monkeypatch: pytest.MonkeyPatch, settings: dict[str, Any], expected: int
self, settings: dict[str, Any], expected: int
) -> None:
# restores the value that FilesPipeline.from_crawler() sets on the class
monkeypatch.setattr(S3FilesStore, "AWS_MAX_POOL_CONNECTIONS", None)
crawler = get_crawler(
settings_dict={"FILES_STORE": "s3://mybucket/prefix/", **settings}
)
@ -966,20 +1033,36 @@ class TestGCSFilesStore:
in caplog.text
)
def test_update_stores(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(GCSFilesStore, "GCS_PROJECT_ID", None)
monkeypatch.setattr(GCSFilesStore, "POLICY", None)
@pytest.mark.parametrize(
("acl", "policy"),
[
("publicRead", "publicRead"),
# An empty FILES_STORE_GCS_ACL is normalised to None.
("", None),
],
)
def test_from_crawler(self, acl: str, policy: str | None) -> None:
pytest.importorskip("google.cloud.storage")
settings = Settings(
{"GCS_PROJECT_ID": "my-project", "FILES_STORE_GCS_ACL": "publicRead"}
client_mock, bucket_mock, _ = mock_google_cloud_storage()
bucket_mock.test_iam_permissions.return_value = [
"storage.objects.get",
"storage.objects.create",
]
crawler = get_crawler(
settings_dict={
"GCS_PROJECT_ID": "my-project",
"FILES_STORE_GCS_ACL": acl,
}
)
FilesPipeline._update_stores(settings)
assert GCSFilesStore.GCS_PROJECT_ID == "my-project"
assert GCSFilesStore.POLICY == "publicRead"
with mock.patch("google.cloud.storage.Client", return_value=client_mock):
store = GCSFilesStore.from_crawler(
crawler, "gs://my_bucket/my_prefix/", resolve=lambda setting: setting
)
# An empty FILES_STORE_GCS_ACL is normalised to None.
settings = Settings({"GCS_PROJECT_ID": "my-project", "FILES_STORE_GCS_ACL": ""})
FilesPipeline._update_stores(settings)
assert store.GCS_PROJECT_ID == "my-project"
assert store.POLICY == policy
assert GCSFilesStore.GCS_PROJECT_ID is None
assert GCSFilesStore.POLICY is None
@coroutine_test

92
tests/test_pipeline_images.py
View File

@ -10,6 +10,7 @@ from shutil import rmtree
from tempfile import mkdtemp
from types import SimpleNamespace
from typing import Any
from unittest import mock
import attr
import pytest
@ -18,9 +19,11 @@ from itemadapter import ItemAdapter
from scrapy.exceptions import NotConfigured
from scrapy.http import Request, Response
from scrapy.item import Field, Item
from scrapy.pipelines.files import GCSFilesStore, S3FilesStore, _md5sum
from scrapy.pipelines.files import FilesPipeline, GCSFilesStore, S3FilesStore, _md5sum
from scrapy.pipelines.images import ImageException, ImagesPipeline
from scrapy.utils.test import get_crawler
from tests.test_pipeline_files import CrawlerAwareStore
from tests.utils.cloud import mock_google_cloud_storage
from tests.utils.decorators import coroutine_test
from tests.utils.media_pipelines import DUMMY_SPIDER_INFO
@ -590,43 +593,60 @@ class TestImagesPipelineCustomSettings:
expected_value = settings.get(settings_attr)
assert getattr(pipeline_cls, pipe_attr.lower()) == expected_value
def test_images_store_s3_acl_setting_used(self, tmp_path):
old_policy = S3FilesStore.POLICY
try:
crawler = get_crawler(
None,
{
"IMAGES_STORE": tmp_path,
"IMAGES_STORE_S3_ACL": "public-read",
"FILES_STORE_S3_ACL": "private",
def test_media_storages_resolve_setting(self):
crawler = get_crawler(
None,
{
"IMAGES_STORE": "mystore://example.com/",
"MEDIA_STORAGES": {
"mystore": "tests.test_pipeline_files.CrawlerAwareStore"
},
},
)
store = ImagesPipeline.from_crawler(crawler).store
assert isinstance(store, CrawlerAwareStore)
assert store.acl_setting == "IMAGES_STORE_S3_ACL"
@pytest.mark.requires_botocore
def test_images_store_s3_acl_setting_used(self):
crawler = get_crawler(
None,
{
"FILES_STORE": "s3://bucket/files/",
"IMAGES_STORE": "s3://bucket/images/",
"FILES_STORE_S3_ACL": "private",
"IMAGES_STORE_S3_ACL": "public-read",
},
)
assert FilesPipeline.from_crawler(crawler).store.POLICY == "private"
assert ImagesPipeline.from_crawler(crawler).store.POLICY == "public-read"
assert S3FilesStore.POLICY == "private"
def test_images_store_gcs_acl_setting_used(self):
pytest.importorskip("google.cloud.storage")
client_mock, bucket_mock, _ = mock_google_cloud_storage()
bucket_mock.test_iam_permissions.return_value = [
"storage.objects.get",
"storage.objects.create",
]
crawler = get_crawler(
None,
{
"FILES_STORE": "gs://bucket/files/",
"IMAGES_STORE": "gs://bucket/images/",
"FILES_STORE_GCS_ACL": "",
"IMAGES_STORE_GCS_ACL": "authenticatedRead",
},
)
with mock.patch("google.cloud.storage.Client", return_value=client_mock):
assert FilesPipeline.from_crawler(crawler).store.POLICY is None
assert (
ImagesPipeline.from_crawler(crawler).store.POLICY == "authenticatedRead"
)
ImagesPipeline.from_crawler(crawler)
assert S3FilesStore.POLICY == "public-read"
finally:
S3FilesStore.POLICY = old_policy
def test_images_store_gcs_acl_setting_used(self, tmp_path):
old_policy = GCSFilesStore.POLICY
try:
crawler = get_crawler(
None,
{
"IMAGES_STORE": tmp_path,
"IMAGES_STORE_GCS_ACL": "authenticatedRead",
"FILES_STORE_GCS_ACL": "",
},
)
ImagesPipeline.from_crawler(crawler)
assert GCSFilesStore.POLICY == "authenticatedRead"
finally:
GCSFilesStore.POLICY = old_policy
assert GCSFilesStore.POLICY is None
def _create_image(format_: str, *a: Any, **kw: Any) -> tuple[Image.Image, io.BytesIO]:

1
tests/test_settings/__init__.py
View File

@ -25,6 +25,7 @@ NON_COMPONENT_PRIORITY_DICT_BASE_SETTING_NAMES = {
"DOWNLOAD_HANDLERS_BASE",
"FEED_EXPORTERS_BASE",
"FEED_STORAGES_BASE",
"MEDIA_STORAGES_BASE",
}