Remove LocalStorageDriver import from stored_file.py and storage.py

Author: jowilf

sqlalchemy_file/helpers.py

@@ -7,7 +7,7 @@
 from typing import Any, Dict, Union
 
 INMEMORY_FILESIZE = 1024 * 1024
-
+LOCAL_STORAGE_DRIVER_NAME = "Local Storage"
 
 def get_metadata_file_obj(metadata: Dict[str, Any]) -> "SpooledTemporaryFile[bytes]":
     f = SpooledTemporaryFile(INMEMORY_FILESIZE)

sqlalchemy_file/storage.py

@@ -1,9 +1,8 @@
 from typing import Any, Dict, Iterator, Optional
 
 from libcloud.storage.base import Container
-from libcloud.storage.drivers.local import LocalStorageDriver
 from libcloud.storage.types import ObjectDoesNotExistError
-from sqlalchemy_file.helpers import get_metadata_file_obj
+from sqlalchemy_file.helpers import get_metadata_file_obj, LOCAL_STORAGE_DRIVER_NAME
 from sqlalchemy_file.stored_file import StoredFile
 
 
@@ -75,7 +74,7 @@ def save_file(
     ) -> StoredFile:
         """Save file into provided `upload_storage`"""
         container = cls.get(upload_storage)
-        if isinstance(container.driver, LocalStorageDriver):
+        if container.driver.name == LOCAL_STORAGE_DRIVER_NAME:
             obj = container.upload_object_via_stream(iterator=content, object_name=name)
             if metadata is not None:
                 """
@@ -115,7 +114,7 @@ def delete_file(cls, path: str) -> bool:
         """
         upload_storage, file_id = path.split("/")
         obj = StorageManager.get(upload_storage).get_object(file_id)
-        if isinstance(obj.driver, LocalStorageDriver):
+        if obj.driver.name == LOCAL_STORAGE_DRIVER_NAME:
             """Try deleting associated metadata file"""
             try:
                 obj.container.get_object(f"{obj.name}.metadata.json").delete()

sqlalchemy_file/stored_file.py

@@ -3,13 +3,14 @@
 from typing import Optional
 
 from libcloud.storage.base import Object
-from libcloud.storage.drivers.local import LocalStorageDriver
 from libcloud.storage.types import ObjectDoesNotExistError
 
+from sqlalchemy_file.helpers import LOCAL_STORAGE_DRIVER_NAME
+
 
 class StoredFile:
     def __init__(self, obj: Object) -> None:
-        if isinstance(obj.driver, LocalStorageDriver):
+        if obj.driver.name == LOCAL_STORAGE_DRIVER_NAME:
             """Retrieve metadata from associated metadata file"""
             try:
                 metadata_obj = obj.container.get_object(f"{obj.name}.metadata.json")
@@ -31,7 +32,7 @@ def get_cdn_url(self) -> Optional[str]:
             return None
 
     def read(self, n: int = -1) -> bytes:
-        if isinstance(self.object.driver, LocalStorageDriver):
+        if self.object.driver == LOCAL_STORAGE_DRIVER_NAME:
             return open(self.object.get_cdn_url(), "rb").read(n)
         _file = tempfile.NamedTemporaryFile()
         self.object.download(_file.name, overwrite_existing=True)