diff --git a/examples/s3/main.py b/examples/s3/main.py index 948ed7ad0..f1ed33fc5 100644 --- a/examples/s3/main.py +++ b/examples/s3/main.py @@ -47,7 +47,7 @@ def index(): s3_client.upload_fileobj( BytesIO(b"abcdef"), "bucket", - "some-directory/some-file", + "some-directory/some-file-2", ExtraArgs={"ContentType": "text/plain"}, ) @@ -66,6 +66,16 @@ def index(): ) ) + admin.add_view( + S3FileAdmin( + bucket_name=bucket_name, + s3_client=s3_client, + name="S3 with Prefix", + prefix="some-directory/", + endpoint="s3-with-prefix", + ) + ) + # Add Local Directory view admin.add_view(FileAdmin("localdir", name="Local Dir")) diff --git a/flask_admin/contrib/fileadmin/s3.py b/flask_admin/contrib/fileadmin/s3.py index 7a4bd5eef..f0e611baa 100644 --- a/flask_admin/contrib/fileadmin/s3.py +++ b/flask_admin/contrib/fileadmin/s3.py @@ -65,7 +65,9 @@ class MyS3Admin(BaseFileAdmin): fileadmin_view = MyS3Admin(storage=S3Storage(...)) """ - def __init__(self, s3_client: BaseClient, bucket_name: str) -> None: + def __init__( + self, s3_client: BaseClient, bucket_name: str, prefix: str = "" + ) -> None: """ Constructor @@ -85,21 +87,26 @@ def __init__(self, s3_client: BaseClient, bucket_name: str) -> None: self.s3_client = s3_client self.bucket_name = bucket_name self.separator = "/" + prefix = self.normpath(prefix).lstrip("/") + prefix = "" if prefix in [".", "./"] else prefix + if prefix and not prefix.endswith(self.separator): + prefix += self.separator + self.prefix = prefix - @_strip_leading_slash_from("path") - def get_files(self, path: str, directory: str) -> list[t.Any]: - def _strip_path(name: str, path: str) -> str: + @_strip_leading_slash_from("directory") + def get_files(self, directory: str, path: str) -> list[t.Any]: + def _strip_prefix(name: str) -> str: if name.startswith(path): return name.replace(path, "", 1) return name - def _remove_trailing_slash(name: str) -> str: - return name[:-1] + def _rcoat(x: str) -> str: + return x.ljust(len(x) + 1, "/") if not x.endswith("/") else x files = [] directories: list[tuple[str, str, bool, int, int]] = [] - if path and not path.endswith(self.separator): - path += self.separator + directory = _rcoat(directory) if directory else "" + path = _rcoat(path) if path else "" try: paginator = self.s3_client.get_paginator("list_objects_v2") @@ -107,19 +114,20 @@ def _remove_trailing_slash(name: str) -> str: Bucket=self.bucket_name, Prefix=path, Delimiter=self.separator ): for common_prefix in page.get("CommonPrefixes", []): - name = _remove_trailing_slash( - _strip_path(common_prefix["Prefix"], path) + name = common_prefix["Prefix"].removeprefix(path).rstrip("/") + rel_path = ( + common_prefix["Prefix"].removeprefix(self.prefix).rstrip("/") ) - key_name = _remove_trailing_slash(common_prefix["Prefix"]) - directories.append((name, key_name, True, 0, 0)) + directories.append((name, rel_path, True, 0, 0)) for obj in page.get("Contents", []): if obj["Key"] == path: continue last_modified = int(obj["LastModified"].timestamp()) - name = _strip_path(obj["Key"], path) - files.append((name, obj["Key"], False, obj["Size"], last_modified)) + name = obj["Key"].removeprefix(path) + rel_path = obj["Key"].removeprefix(self.prefix) + files.append((name, rel_path, False, obj["Size"], last_modified)) except ClientError as e: raise ValueError(f"Failed to list files: {e}") from e @@ -166,10 +174,10 @@ def path_exists(self, path: str) -> bool: if path == "": return True keys = self._get_path_keys(path) - return path in keys or (path + self.separator) in keys + return any([k.startswith(path) for k in keys]) def get_base_path(self) -> str: - return "" + return self.prefix @_strip_leading_slash_from("path") def get_breadcrumbs(self, path: str) -> list[tuple[str, str]]: @@ -288,6 +296,12 @@ class S3FileAdmin(BaseFileAdmin): :param bucket_name: Name of the bucket that the files are on. + :param prefix: + Optional prefix to use within the bucket. Note that this is different + from the `base_path` parameter of the BaseFileAdmin, which is handled + internally by the S3Storage. Prefix must be specified without leading + slash and will be normalized to end with a slash. + Sample usage:: from flask_admin import Admin @@ -305,8 +319,10 @@ def __init__( self, s3_client: BaseClient, bucket_name: str, + prefix: str = "", *args: t.Any, **kwargs: t.Any, ) -> None: - storage = S3Storage(s3_client, bucket_name) + storage = S3Storage(s3_client, bucket_name, prefix=prefix) + super().__init__(*args, storage=storage, **kwargs) # type: ignore[misc] diff --git a/flask_admin/tests/fileadmin/test_fileadmin_s3.py b/flask_admin/tests/fileadmin/test_fileadmin_s3.py index f7b9a15fb..fabd03775 100644 --- a/flask_admin/tests/fileadmin/test_fileadmin_s3.py +++ b/flask_admin/tests/fileadmin/test_fileadmin_s3.py @@ -209,3 +209,154 @@ class MyFileAdmin(fileadmin_class): # type: ignore[valid-type, misc] # Test access to deep path rv = client.get("/admin/myfileadmin/b/xx/yy/zz") assert rv.status_code == 200 + + def test_prefix(self, app: Flask, admin: Admin, mock_s3_client: t.Any) -> None: + fileadmin_class = self.fileadmin_class() + fileadmin_args, fileadmin_kwargs = self.fileadmin_args() + + class MyFileAdmin(fileadmin_class): # type: ignore[valid-type, misc] + pass + + view_kwargs = dict(fileadmin_kwargs) + view_kwargs["prefix"] = "xx/yy/" + view_kwargs.setdefault("name", "Files") + view = MyFileAdmin(*fileadmin_args, **view_kwargs) + + # create deep path + view.storage.make_dir("", "xx/yy/zz/") + + admin.add_view(view) + + client = app.test_client() + + # upload to deep path + rv = client.post( + "/admin/myfileadmin/upload/zz/", + data=dict(upload=(BytesIO(b"test content"), "test_upload.txt")), + follow_redirects=True, + ) + assert rv.status_code == 200 + assert "Successfully saved file: test_upload.txt" in rv.text + + rv = client.get("/admin/myfileadmin/b/zz/") + assert rv.status_code == 200 + assert "path=zz/test_upload.txt" in rv.text + + # rename in deep path + rv = client.post( + "/admin/myfileadmin/rename/?path=zz/test_upload.txt", + data=dict(name="dummy.txt", path="zz/test_upload.txt"), + ) + assert rv.status_code == 302 + + rv = client.get("/admin/myfileadmin/b/zz/") + assert rv.status_code == 200 + assert "path=zz/dummy.txt" in rv.text + assert "path=test_upload.txt" not in rv.text + + # download from deep path + rv = client.get("/admin/myfileadmin/download/zz/dummy.txt") + assert rv.status_code == 302 + assert rv.headers["Location"].startswith( + "https://my-bucket.s3.amazonaws.com/xx/yy/zz/dummy.txt?AWSAccessKeyId=FOOBARKEY" + ) + + # delete + rv = client.post( + "/admin/myfileadmin/delete", + data=dict(path="zz/dummy.txt"), + follow_redirects=True, + ) + assert rv.status_code == 200 + assert "successfully deleted" in rv.text + + @pytest.mark.parametrize( + "prefix, res_code", + [ + ("xx/yy", 200), + ("xx/yy/", 200), + ("/xx/yy", 200), + ("/xx/yy/", 200), + ("/xx//yy/", 200), + ("xx\\yy", 404), + ("/xx\\yy/", 404), + ("xx/../xx/yy", 200), + ("xx/../xx//yy", 200), + ], + ) + def test_base_path( + self, + app: Flask, + admin: Admin, + mock_s3_client: t.Any, + prefix: str, + res_code: int, + ) -> None: + fileadmin_class = self.fileadmin_class() + fileadmin_args, fileadmin_kwargs = self.fileadmin_args() + + s3 = fileadmin_args[0] + s3.upload_fileobj(BytesIO(b""), _bucket_name, "xx/yy/zz/dummy2.txt") + + class MyFileAdmin(fileadmin_class): # type: ignore[valid-type, misc] + pass + + view_kwargs = dict(fileadmin_kwargs) + view_kwargs["prefix"] = prefix + view_kwargs.setdefault("name", "Files") + view = MyFileAdmin(*fileadmin_args, **view_kwargs) + + admin.add_view(view) + + client = app.test_client() + + # actual s3 prefix is xx/yy/ + rv = client.get("/admin/myfileadmin/") + assert rv.status_code == res_code + if res_code == 200: + assert "dummy2.txt" not in rv.data.decode("utf-8") + + rv = client.get("/admin/myfileadmin/b/zz/") + assert rv.status_code == res_code + if res_code == 200: + assert "dummy2.txt" in rv.data.decode("utf-8") + + rv = client.get("/admin/myfileadmin/b/xx/zz/") + assert rv.status_code == 404 + + @pytest.mark.parametrize( + "prefix", + ["", ".", "/", "./"], + ) + def test_base_path_root( + self, app: Flask, admin: Admin, mock_s3_client: t.Any, prefix: str + ) -> None: + fileadmin_class = self.fileadmin_class() + fileadmin_args, fileadmin_kwargs = self.fileadmin_args() + + s3 = fileadmin_args[0] + s3.upload_fileobj(BytesIO(b""), _bucket_name, "xx/yy/zz/dummy2.txt") + + class MyFileAdmin(fileadmin_class): # type: ignore[valid-type, misc] + pass + + view_kwargs = dict(fileadmin_kwargs) + view_kwargs["prefix"] = prefix + view_kwargs.setdefault("name", "Files") + view = MyFileAdmin(*fileadmin_args, **view_kwargs) + + admin.add_view(view) + + client = app.test_client() + + rv = client.get("/admin/myfileadmin/") + assert rv.status_code == 200 + assert "dummy2.txt" not in rv.data.decode("utf-8") + + rv = client.get("/admin/myfileadmin/b/xx/yy/") + assert rv.status_code == 200 + assert "dummy2.txt" not in rv.data.decode("utf-8") + + rv = client.get("/admin/myfileadmin/b/xx/yy/zz") + assert rv.status_code == 200 + assert "dummy2.txt" in rv.data.decode("utf-8")