Search for credentials in explicit environment variables.
(self)
| 1237 | return var_mapping |
| 1238 | |
| 1239 | def load(self): |
| 1240 | """ |
| 1241 | Search for credentials in explicit environment variables. |
| 1242 | """ |
| 1243 | |
| 1244 | access_key = self.environ.get(self._mapping['access_key'], '') |
| 1245 | |
| 1246 | if access_key: |
| 1247 | logger.info('Found credentials in environment variables.') |
| 1248 | fetcher = self._create_credentials_fetcher() |
| 1249 | credentials = fetcher(require_expiry=False) |
| 1250 | register_feature_id('CREDENTIALS_ENV_VARS') |
| 1251 | |
| 1252 | expiry_time = credentials['expiry_time'] |
| 1253 | if expiry_time is not None: |
| 1254 | expiry_time = parse(expiry_time) |
| 1255 | return RefreshableCredentials( |
| 1256 | credentials['access_key'], |
| 1257 | credentials['secret_key'], |
| 1258 | credentials['token'], |
| 1259 | expiry_time, |
| 1260 | refresh_using=fetcher, |
| 1261 | method=self.METHOD, |
| 1262 | account_id=credentials['account_id'], |
| 1263 | ) |
| 1264 | |
| 1265 | return Credentials( |
| 1266 | credentials['access_key'], |
| 1267 | credentials['secret_key'], |
| 1268 | credentials['token'], |
| 1269 | method=self.METHOD, |
| 1270 | account_id=credentials['account_id'], |
| 1271 | ) |
| 1272 | else: |
| 1273 | return None |
| 1274 | |
| 1275 | def _create_credentials_fetcher(self): |
| 1276 | mapping = self._mapping |