(self, uri)
| 64 | OSS_CONNECT_TIMEOUT = None |
| 65 | |
| 66 | def __init__(self, uri): |
| 67 | import oss2 |
| 68 | assert uri.startswith('oss://') |
| 69 | self.OSS_BUCKET_NAME, self.prefix = uri[6:].split('/', 1) |
| 70 | self.bucket = oss2.Bucket(oss2.Auth(self.OSS_ACCESS_KEY_ID, self.OSS_ACCESS_KEY_SECRET), |
| 71 | self.OSS_ENDPOINT, |
| 72 | self.OSS_BUCKET_NAME, |
| 73 | connect_timeout=self.OSS_CONNECT_TIMEOUT) |
| 74 | |
| 75 | def stat_file(self, path, info): |
| 76 | key_name = os.path.join(self.prefix, path) |
nothing calls this directly
no outgoing calls
no test coverage detected