Ported and enhanced tenant aware file system storage

django_pgschemas/contrib/files/__init__.py

@@ -0,0 +1 @@
+from .storage import TenantFileSystemStorage

django_pgschemas/contrib/files/storage.py

@@ -0,0 +1,55 @@
+import os
+
+from django.conf import settings
+from django.core.files.storage import FileSystemStorage
+from django.db import connection
+
+
+class TenantFileSystemStorage(FileSystemStorage):
+    """
+    Tenant aware file system storage. Appends the schema name of the tenant to
+    the base location and base URL.
+    """
+
+    def get_schema_path_identifier(self):
+        if not connection.schema:
+            return ""
+        path_identifier = connection.schema.schema_name
+        if hasattr(connection.schema, "schema_pathname"):
+            path_identifier = connection.schema.schema_pathname()
+        elif hasattr(settings, "PGSCHEMAS_PATHNAME_FUNCTION"):
+            path_identifier = settings.PGSCHEMAS_PATHNAME_FUNCTION(connection.schema)
+        return path_identifier
+
+    @property  # To avoid caching of tenant
+    def base_location(self):
+        """
+        Appends base location with the schema path identifier.
+        """
+        file_folder = self.get_schema_path_identifier()
+        location = os.path.join(super().base_location, file_folder)
+        if not location.endswith("/"):
+            location += "/"
+        return location
+
+    @property  # To avoid caching of tenant
+    def location(self):
+        return super().location
+
+    @property  # To avoid caching of tenant
+    def base_url(self):
+        """
+        Optionally appends base URL with the schema path identifier.
+        If the current schema is already using a folder, no path identifier is
+        appended.
+        """
+        url_folder = self.get_schema_path_identifier()
+        if url_folder and connection.schema and connection.schema.folder:
+            # Since we're already prepending all URLs with schema, there is no
+            # need to make the differentiation here
+            url_folder = ""
+        parent_base_url = super().base_url.strip("/")
+        url = "/".join(["", parent_base_url, url_folder])
+        if not url.endswith("/"):
+            url += "/"
+        return url

docs/contrib.rst

@@ -15,6 +15,52 @@ We're striving to maintain/increase our code coverage, but please, make sure you
 integration is properly tested. Proper tests will always beat meaningless 100%
 coverage.
 
+Tenant aware file system storage
+--------------------------------
+
+We provide a tenant aware file system storage at
+``django_pgschemas.contrib.files.TenantFileSystemStorage``. It subclasses
+``django.core.files.storage.FileSystemStorage`` and behaves like it in every
+aspect, except that it prepends a tenant identifier to the path and URL of all
+files.
+
+By default, the tenant identifier is the schema name of the current tenant. In
+order to override this behavior, it is possible to provide a different
+identifier. The storage will consider these options when looking for an
+identifier:
+
+* A method called ``schema_pathname`` in the current tenant. This method must
+  accept no arguments and return an identifier.
+* A function specified in a setting called ``PGSCHEMAS_PATHNAME_FUNCTION``. This
+  function must accept a schema descriptor and return an identifier.
+* Finally, the identifier will default to the schema name of the current tenant.
+
+In the case of the URL returned from the storage, if the storage detects that
+the current schema has been routed via subfolder, it won't prepend the schema
+identifier, because it considers that the path is properly disambiguated as is.
+This means that instead of something like::
+
+    /tenant1/static/tenant1/path/to/file.txt
+
+It will generate::
+
+    /tenant1/static/path/to/file.txt
+
+This storage class is a convenient way of storing media files in a folder
+structure organized at the top by tenants, as well as providing a perceived
+tenant centric organization in the URLs that are generated. However, this
+storage class does NOT provide any form of security, such as controlling that
+from one tenant, files from another tenant are not accessible. Such security
+requirements have other implications that fall out of the scope of this basic
+utility.
+
+.. tip::
+
+    In a project that requires airtight security, you might want to use and
+    customize `django-private-storage`_.
+
+.. _django-private-storage: https://github.com/edoburu/django-private-storage
+
 Channels (websockets)
 ---------------------
 

dpgs_sandbox/tests/test_file_storage.py

@@ -0,0 +1,110 @@
+import os
+import shutil
+import tempfile
+
+from django.db import connection
+from django.core.files.base import ContentFile
+from django.test import TransactionTestCase, override_settings
+
+from django_pgschemas.contrib.files import TenantFileSystemStorage
+from django_pgschemas.schema import SchemaDescriptor
+from django_pgschemas.utils import get_tenant_model
+
+TenantModel = get_tenant_model()
+
+
+class TenantFileSystemStorageTestCase(TransactionTestCase):
+    """
+    Tests the tenant file system storage.
+    """
+
+    @classmethod
+    def setUpClass(cls):
+        cls.temp_dir = tempfile.mkdtemp()
+        cls.storage = TenantFileSystemStorage(location=cls.temp_dir, base_url="/base-url/")
+
+    @classmethod
+    def tearDownClass(cls):
+        for tenant in TenantModel.objects.all():
+            tenant.delete(force_drop=True)
+        shutil.rmtree(cls.temp_dir)
+
+    def test_path_identifier_basic(self):
+        with SchemaDescriptor.create(schema_name=""):
+            self.assertEquals(self.storage.get_schema_path_identifier(), "")
+        with SchemaDescriptor.create(schema_name="public"):
+            self.assertEquals(self.storage.get_schema_path_identifier(), "public")
+        with SchemaDescriptor.create(schema_name="blog"):
+            self.assertEquals(self.storage.get_schema_path_identifier(), "blog")
+        with TenantModel(schema_name="tenant"):
+            self.assertEquals(self.storage.get_schema_path_identifier(), "tenant")
+
+    def test_path_identifier_method_in_tenant(self):
+        TenantModel.schema_pathname = lambda x: "custom-pathname"
+        with TenantModel(schema_name="tenant"):
+            self.assertEquals(self.storage.get_schema_path_identifier(), "custom-pathname")
+        del TenantModel.schema_pathname
+
+    def test_path_identifier_function_in_settings(self):
+        with override_settings(PGSCHEMAS_PATHNAME_FUNCTION=lambda tenant: tenant.schema_name + "-custom-pathname"):
+            with TenantModel(schema_name="tenant"):
+                self.assertEquals(self.storage.get_schema_path_identifier(), "tenant-custom-pathname")
+
+    def test_base_location(self):
+        with SchemaDescriptor.create(schema_name=""):
+            self.assertEquals(self.storage.base_location, self.temp_dir + "/")
+        with SchemaDescriptor.create(schema_name="public"):
+            self.assertEquals(self.storage.base_location, self.temp_dir + "/public/")
+        with SchemaDescriptor.create(schema_name="blog"):
+            self.assertEquals(self.storage.base_location, self.temp_dir + "/blog/")
+        with SchemaDescriptor.create(schema_name="tenant", folder="folder"):
+            self.assertEquals(self.storage.base_location, self.temp_dir + "/tenant/")
+
+    def test_base_url(self):
+        with SchemaDescriptor.create(schema_name=""):
+            self.assertEquals(self.storage.base_url, "/base-url/")
+        with SchemaDescriptor.create(schema_name="public"):
+            self.assertEquals(self.storage.base_url, "/base-url/public/")
+        with SchemaDescriptor.create(schema_name="blog"):
+            self.assertEquals(self.storage.base_url, "/base-url/blog/")
+        with SchemaDescriptor.create(schema_name="tenant", folder="folder"):
+            self.assertEquals(self.storage.base_url, "/base-url/")
+
+    def test_file_path(self):
+        self.assertFalse(self.storage.exists("test.file"))
+        with SchemaDescriptor.create(schema_name="tenant1"):
+            f = ContentFile("random content")
+            f_name = self.storage.save("test.file", f)
+            self.assertEqual(os.path.join(self.temp_dir, "tenant1", f_name), self.storage.path(f_name))
+            self.storage.delete(f_name)
+        self.assertFalse(self.storage.exists("test.file"))
+
+    def test_file_save_with_path(self):
+        self.assertFalse(self.storage.exists("path/to"))
+        with SchemaDescriptor.create(schema_name="tenant1"):
+            self.storage.save("path/to/test.file", ContentFile("file saved with path"))
+            self.assertTrue(self.storage.exists("path/to"))
+            with self.storage.open("path/to/test.file") as f:
+                self.assertEqual(f.read(), b"file saved with path")
+            self.assertTrue(os.path.exists(os.path.join(self.temp_dir, "tenant1", "path", "to", "test.file")))
+            self.storage.delete("path/to/test.file")
+            self.assertFalse(self.storage.exists("test.file"))
+
+    def test_file_url_simple(self):
+        with SchemaDescriptor.create(schema_name=""):
+            self.assertEqual(self.storage.url("test.file"), "/base-url/test.file")
+        with SchemaDescriptor.create(schema_name="public"):
+            self.assertEqual(self.storage.url("test.file"), "/base-url/public/test.file")
+        with SchemaDescriptor.create(schema_name="tenant", folder="folder"):
+            self.assertEqual(self.storage.url("test.file"), "/base-url/test.file")
+
+    def test_file_url_complex(self):
+        with SchemaDescriptor.create(schema_name="tenant"):
+            self.assertEqual(
+                self.storage.url(r"~!*()'@#$%^&*abc`+ =.file"),
+                "/base-url/tenant/~!*()'%40%23%24%25%5E%26*abc%60%2B%20%3D.file",
+            )
+            self.assertEqual(self.storage.url("ab\0c"), "/base-url/tenant/ab%00c")
+            self.assertEqual(self.storage.url("a/b\\c.file"), "/base-url/tenant/a/b/c.file")
+            self.assertEqual(self.storage.url(""), "/base-url/tenant/")
+            self.assertEqual(self.storage.url(None), "/base-url/tenant/")