I'm creating a subclass of s3fs.S3FileSystem
that connects to either AWS S3 or MinIO based on environment variables. The connection should fail immediately during initialization if credentials are invalid.
The parent class accepts invalid credentials during initialization, only failing later during actual operations (like ls()
or file access).
When I try to validate the connection by calling self.ls('')
inside __init__
(via a _test_connection()
helper), I get a NotImplementedError
. However, if I call the same method after instantiation, it works correctly.
class S3MinioConnector(s3fs.S3FileSystem):
def __init__(self) -> None:
# ... (setup code)
super().__init__(**params) # Initialize parent
self._test_connection() # Fails here with NotImplementedError
def _test_connection(self):
self.ls('') # Inherited method
How can I properly validate the connection during initialization without:
Getting
NotImplementedError
?Using obscure solutions like
time.sleep(2)
?