(
self,
uri: str,
access_key: str | None = None,
secret_key: str | None = None,
acl: str | None = None,
endpoint_url: str | None = None,
*,
feed_options: dict[str, Any] | None = None,
session_token: str | None = None,
region_name: str | None = None,
)
| 186 | |
| 187 | class S3FeedStorage(BlockingFeedStorage): |
| 188 | def __init__( |
| 189 | self, |
| 190 | uri: str, |
| 191 | access_key: str | None = None, |
| 192 | secret_key: str | None = None, |
| 193 | acl: str | None = None, |
| 194 | endpoint_url: str | None = None, |
| 195 | *, |
| 196 | feed_options: dict[str, Any] | None = None, |
| 197 | session_token: str | None = None, |
| 198 | region_name: str | None = None, |
| 199 | ): |
| 200 | try: |
| 201 | import boto3.session # noqa: PLC0415 |
| 202 | except ImportError: |
| 203 | raise NotConfigured("missing boto3 library") from None |
| 204 | u = urlparse(uri) |
| 205 | assert u.hostname |
| 206 | self.bucketname: str = u.hostname |
| 207 | self.access_key: str | None = u.username or access_key |
| 208 | self.secret_key: str | None = u.password or secret_key |
| 209 | self.session_token: str | None = session_token |
| 210 | self.keyname: str = u.path[1:] # remove first "/" |
| 211 | self.acl: str | None = acl |
| 212 | self.endpoint_url: str | None = endpoint_url |
| 213 | self.region_name: str | None = region_name |
| 214 | |
| 215 | boto3_session = boto3.session.Session() |
| 216 | self.s3_client = boto3_session.client( |
| 217 | "s3", |
| 218 | aws_access_key_id=self.access_key, |
| 219 | aws_secret_access_key=self.secret_key, |
| 220 | aws_session_token=self.session_token, |
| 221 | endpoint_url=self.endpoint_url, |
| 222 | region_name=self.region_name, |
| 223 | ) |
| 224 | |
| 225 | if feed_options and feed_options.get("overwrite", True) is False: |
| 226 | logger.warning( |
| 227 | "S3 does not support appending to files. To " |
| 228 | "suppress this warning, remove the overwrite " |
| 229 | "option from your FEEDS setting or set it to True." |
| 230 | ) |
| 231 | |
| 232 | @classmethod |
| 233 | def from_crawler( |
nothing calls this directly
no test coverage detected