(self, uri: str)
| 167 | } |
| 168 | |
| 169 | def __init__(self, uri: str): |
| 170 | if not is_botocore_available(): |
| 171 | raise NotConfigured("missing botocore library") |
| 172 | import botocore.session # noqa: PLC0415 |
| 173 | |
| 174 | session = botocore.session.get_session() |
| 175 | self.s3_client = session.create_client( |
| 176 | "s3", |
| 177 | aws_access_key_id=self.AWS_ACCESS_KEY_ID, |
| 178 | aws_secret_access_key=self.AWS_SECRET_ACCESS_KEY, |
| 179 | aws_session_token=self.AWS_SESSION_TOKEN, |
| 180 | endpoint_url=self.AWS_ENDPOINT_URL, |
| 181 | region_name=self.AWS_REGION_NAME, |
| 182 | use_ssl=self.AWS_USE_SSL, |
| 183 | verify=self.AWS_VERIFY, |
| 184 | ) |
| 185 | if not uri.startswith("s3://"): |
| 186 | raise ValueError(f"Incorrect URI scheme in {uri}, expected 's3'") |
| 187 | self.bucket, self.prefix = uri[5:].split("/", 1) |
| 188 | |
| 189 | @staticmethod |
| 190 | def _onsuccess(boto_key: dict[str, Any]) -> StatInfo: |
nothing calls this directly
no test coverage detected