{"record":{"id":"e851af5afc94d590","repo":"HumanSignal/label-studio","slug":"validate-connection-is-not-implemented","errorCode":null,"errorMessage":"validate_connection is not implemented","messagePattern":"validate_connection is not implemented","errorType":"exception","errorClass":"NotImplementedError","httpStatus":null,"severity":"error","filePath":"label_studio/io_storages/base_models.py","lineNumber":318,"sourceCode":"                'and no traceback information is available.\\n'\n                'This typically occurs if job was manually removed '\n                'or workers reloaded unexpectedly.'\n            )\n            self.save(update_fields=['status', 'traceback'])\n            logger.info(f'Storage {self} status moved to `failed` because the job {self.last_sync_job} was not found')\n\n\nclass Storage(StorageInfo):\n    url_scheme = ''\n\n    title = models.CharField(_('title'), null=True, blank=True, max_length=256, help_text='Cloud storage title')\n    description = models.TextField(_('description'), null=True, blank=True, help_text='Cloud storage description')\n    created_at = models.DateTimeField(_('created at'), auto_now_add=True, help_text='Creation time')\n\n    synchronizable = models.BooleanField(_('synchronizable'), default=True, help_text='If storage can be synced')\n\n    def validate_connection(self, client=None):\n        raise NotImplementedError('validate_connection is not implemented')\n\n    class Meta:\n        abstract = True\n\n\nclass ImportStorage(Storage):\n    def iter_objects(self) -> Iterator[Any]:\n        \"\"\"\n        Returns:\n            Iterator[Any]: An iterator for objects in the storage.\n        \"\"\"\n        raise NotImplementedError\n\n    def iter_keys(self) -> Iterator[str]:\n        \"\"\"\n        Returns:\n            Iterator[str]: An iterator of keys for each object in the storage.\n        \"\"\"","sourceCodeStart":300,"sourceCodeEnd":336,"githubUrl":"https://github.com/HumanSignal/label-studio/blob/0b49e9b53917880baf1dd85d574fe5541a9aafb2/label_studio/io_storages/base_models.py#L300-L336","documentation":"BaseStorage (an abstract Django model) declares validate_connection as a required hook that concrete storage backends (S3, GCS, Azure, etc.) must implement to verify credentials/connectivity. The base implementation always raises NotImplementedError; hitting it means the method was called on a storage class that never overrode it.","triggerScenarios":"Calling storage.validate_connection(...) on an instance of a custom or legacy storage subclass that does not override validate_connection; a custom backend added to Label Studio without implementing the connection-validation hook.","commonSituations":"Writing a custom ImportStorage/ExportStorage subclass and forgetting to implement validate_connection while the UI or API invokes it to test the connection; upgrades that newly call validate_connection on backends that predate it.","solutions":["Implement validate_connection in your storage subclass, using the client to perform a cheap read (e.g. list one bucket/container key) and raise on failure.","If using a built-in backend, ensure you are on a version where that backend implements validate_connection (upgrade Label Studio).","Avoid calling validate_connection on the abstract base directly; call it on the concrete storage model instance."],"exampleFix":"// before\nclass MyStorage(ImportStorage):\n    # no validate_connection override\n    ...\n\n// after\nclass MyStorage(ImportStorage):\n    def validate_connection(self, client=None):\n        try:\n            client.list_buckets()\n        except Exception as e:\n            raise ValueError(f'Cannot connect to storage: {e}')","handlingStrategy":"validation","validationCode":"if type(storage).validate_connection is Storage.validate_connection:\n    raise TypeError(f'{type(storage).__name__} must implement validate_connection()')","typeGuard":"def implements_validate_connection(storage_cls) -> bool:\n    return callable(getattr(storage_cls, 'validate_connection', None)) and \\\n           storage_cls.validate_connection.__qualname__ != Storage.validate_connection.__qualname__","tryCatchPattern":"try:\n    storage.validate_connection(client)\nexcept NotImplementedError:\n    logger.warning('%s does not support connection validation; skipping', type(storage).__name__)\nexcept ValueError as e:\n    logger.error('Connection validation failed: %s', e)","preventionTips":["When writing custom storage backends, override all hooks: validate_connection, iter_objects, iter_keys, get_unified_metadata.","Add a unit test that instantiates every registered storage class and calls validate_connection.","Only call validate_connection on concrete backend classes, not the abstract base."],"tags":["not-implemented","storage","subclassing","abstract-method"],"backgroundTag":"notimplemented-abstract-method","analyzedSha":"0b49e9b53917880baf1dd85d574fe5541a9aafb2","analyzedAt":"2026-08-29T00:39:52.578Z","schemaVersion":2},"datasetVersion":"2026-08-29T02:17:18.158Z"}