Add search option (#131)

* Add search option

* Added missing type hints

* Recreate the client object to avoid failure with continuation marker

* Unify calls to list_object

* Filter duplicates

* Add search option to README

* Rename search button

* Update templates/bucket_contents.html

Co-authored-by: Roman Zydyk <31843161+romanzdk@users.noreply.github.com>

* Use dataclass instead of dictionary to allow hashing

* Fix lint issues

* Fix object creation

* Make S3Entry hashable

---------

Co-authored-by: Roman Zydyk <31843161+romanzdk@users.noreply.github.com>

README.md

@@ -20,6 +20,7 @@ S3 Web Browser is a Flask-based web application that allows users to browse AWS
 
 - **List S3 Buckets**: View all S3 buckets available to the AWS account.
 - **Browse Bucket Contents**: Navigate through the contents of any S3 bucket, including folders and files.
+- **Search Bucket Contents**: Search for files/folders in any S3 bucket, The search will be recursive from the point of origin.
 - **Generate Presigned URLs**: Securely generate temporary URLs for S3 objects, making them accessible for a limited time.
 
 ## Run

app.py

@@ -1,9 +1,10 @@
+import dataclasses
 import os
 
 import boto3
 import botocore
 import humanize
-from flask import Flask, render_template
+from flask import Flask, render_template, request
 
 app = Flask(__name__)
 app.secret_key = "your_secure_random_key_here"  # noqa: S105
@@ -26,24 +27,117 @@
 @app.route("/", methods=["GET"])
 def index() -> str:
     s3 = boto3.resource("s3", **AWS_KWARGS)
-    buckets = s3.buckets.all()
-    return render_template("index.html", buckets=buckets)
+    all_buckets = s3.buckets.all()
+    return render_template("index.html", buckets=all_buckets)
 
 
 @app.route("/buckets")
 def buckets() -> str:
     s3 = boto3.resource("s3", **AWS_KWARGS)
-    buckets = s3.buckets.all()
-    return render_template("index.html", buckets=buckets)
+    all_buckets = s3.buckets.all()
+    return render_template("index.html", buckets=all_buckets)
+
+
+@dataclasses.dataclass(eq=True, frozen=True)
+class S3Entry:
+    """Representation of S3 object."""
+
+    name: str
+    type: str
+    url: str = ""
+    size: str = ""
+    date_modified: str = ""
+
+
+def parse_responses(responses: list, s3_client: botocore.client.BaseClient, bucket_name: str, search_param: str) -> list[S3Entry]:
+    contents: set[S3Entry] = set()
+    for response in responses:
+        # Add folders to contents
+        if "CommonPrefixes" in response:
+            for item in response["CommonPrefixes"]:
+                contents.add(S3Entry(name=item["Prefix"], type="folder"))
+
+        # Add files to contents
+        if "Contents" in response:
+            for item in response["Contents"]:
+                if not item["Key"].endswith("/"):
+                    url = s3_client.generate_presigned_url(
+                        "get_object",
+                        Params={"Bucket": bucket_name, "Key": item["Key"]},
+                        ExpiresIn=3600,
+                    )  # URL expires in 1 hour
+                    contents.add(
+                        S3Entry(
+                            name=f'{bucket_name}/{item["Key"]}',
+                            type="file",
+                            url=url,
+                            size=humanize.naturalsize(item["Size"]),
+                            date_modified=item["LastModified"])
+                    )
+
+    contents_list = list(contents)
+    if search_param:
+        contents_list = list(filter(lambda x: search_param in x.name, contents_list))
+    return sorted(contents_list, key=lambda x: x.type, reverse=True)
+
+
+def list_objects(s3_client: botocore.client.BaseClient, bucket_name: str, path: str, delimiter: str = "") -> list[dict]:
+    responses = []
+    list_params = {"Bucket": bucket_name, "Prefix": path}
+    if delimiter:
+        list_params["Delimiter"] = "/"
+
+    while True:
+        response = s3_client.list_objects_v2(**list_params)
+        responses.append(response)
+        if response["IsTruncated"]:
+            list_params["ContinuationToken"] = response["NextContinuationToken"]
+        else:
+            break
+
+    return responses
+
+
+@app.route("/search//buckets/<bucket_name>", defaults={"path": ""})
+@app.route("/search/buckets/<bucket_name>/<path:path>")
+def search_bucket(bucket_name: str, path: str) -> str:
+    s3_client = boto3.client("s3", **AWS_KWARGS)
+    responses = []
+    try:
+        responses.extend(list_objects(s3_client, bucket_name, path))
+        responses.extend(list_objects(s3_client, bucket_name, path, "/"))
+    except botocore.exceptions.ClientError as e:
+        match e.response["Error"]["Code"]:
+            case "AccessDenied":
+                return render_template(
+                    "error.html",
+                    error="You do not have permission to access this bucket.",
+                )
+            case "NoSuchBucket":
+                return render_template("error.html", error="The specified bucket does not exist.")
+            case _:
+                return render_template("error.html", error=f"An unknown error occurred: {e}")
+    except Exception as e:  # noqa: BLE001
+        return render_template("error.html", error=f"An unknown error occurred: {e}")
+
+    search_param = request.args.get("search", "")
+    contents = parse_responses(responses, s3_client, bucket_name, search_param)
+    return render_template(
+        "bucket_contents.html",
+        contents=contents,
+        bucket_name=bucket_name,
+        path=path,
+        search_param=search_param,
+    )
 
 
 @app.route("/buckets/<bucket_name>", defaults={"path": ""})
 @app.route("/buckets/<bucket_name>/<path:path>")
 def view_bucket(bucket_name: str, path: str) -> str:
     s3_client = boto3.client("s3", **AWS_KWARGS)
-
+    responses = []
     try:
-        response = s3_client.list_objects_v2(Bucket=bucket_name, Prefix=path, Delimiter="/")
+        responses.extend(list_objects(s3_client, bucket_name, path, "/"))
     except botocore.exceptions.ClientError as e:
         match e.response["Error"]["Code"]:
             case "AccessDenied":
@@ -57,44 +151,15 @@ def view_bucket(bucket_name: str, path: str) -> str:
                 return render_template("error.html", error=f"An unknown error occurred: {e}")
     except Exception as e:  # noqa: BLE001
         return render_template("error.html", error=f"An unknown error occurred: {e}")
-    contents = []
-
-    # Add folders to contents
-    if "CommonPrefixes" in response:
-        for item in response["CommonPrefixes"]:
-            contents.append(  # noqa: PERF401
-                {
-                    "name": item["Prefix"],
-                    "type": "folder",
-                    "size": 0,
-                    "date_modified": "",
-                }
-            )
-
-    # Add files to contents
-    if "Contents" in response:
-        for item in response["Contents"]:
-            if not item["Key"].endswith("/"):  # Ignore directories
-                url = s3_client.generate_presigned_url(
-                    "get_object",
-                    Params={"Bucket": bucket_name, "Key": item["Key"]},
-                    ExpiresIn=3600,
-                )  # URL expires in 1 hour
-                contents.append(
-                    {
-                        "name": f'{bucket_name}/{item["Key"]}',
-                        "type": "file",
-                        "url": url,
-                        "size": humanize.naturalsize(item["Size"]),
-                        "date_modified": item["LastModified"],
-                    }
-                )
 
+    search_param = request.args.get("search", "")
+    contents = parse_responses(responses, s3_client, bucket_name, search_param)
     return render_template(
         "bucket_contents.html",
         contents=contents,
         bucket_name=bucket_name,
         path=path,
+        search_param=search_param,
     )
 
 

templates/bucket_contents.html

@@ -1,20 +1,25 @@
 {% extends "base.html" %} {% block content %}
+<form
+  action="{{ url_for('search_bucket', bucket_name=bucket_name, path=path) }}">
+  <input class="form-control" name="search" placeholder="{{ search_param }}"
+    autocomplete="off" autofocus="autofocus" type="text">
+  <button class="btn waves-effect waves-light" type="submit" name="action">
+    Search
+  </button>
+</form>
 <h4>Contents of {{ bucket_name }}</h4>
 <a href="{{ url_for('index') }}" class="btn-small blue">Back to Buckets</a>
 {% if path %}
-<a
-  href="{{ url_for('view_bucket', bucket_name=bucket_name, path=path.rstrip('/').rsplit('/', 1)[0] if '/' in path.rstrip('/') else '') }}"
+<a href="{{ url_for('view_bucket', bucket_name=bucket_name, path=path.rstrip('/').rsplit('/', 1)[0] if '/' in path.rstrip('/') else '') }}"
   class="btn-small">Go Up</a>
 {% endif %}
 <ul class="collection">
   {% for item in contents %}
   <li class="collection-item">
     {% if item.type == "folder" %}
-    <a
-      href="{{ url_for('view_bucket', bucket_name=bucket_name, path=item.name) }}">{{
-      item.name }}</a>
-    {% else %} {{ item.date_modified }} | {{ item.size }} | <a
-      href="{{ item.url }}" target="_blank">{{ item.name }}</a>
+      <a href="{{ url_for('view_bucket', bucket_name=bucket_name, path=item.name) }}">{{ item.name }}</a>
+    {% else %}
+      {{ item.date_modified }} | {{ item.size }} | <a href="{{ item.url }}" target="_blank">{{ item.name }}</a>
     {% endif %}
   </li>
   {% endfor %}