Adding test flask cos event listener

Author: greyhoundforty

apps/cos-event-trigger/.mise.toml

@@ -0,0 +1,29 @@
+[tools]
+python = "3.12"
+usage = "latest"
+
+[env]
+_.python.venv = { path = ".venv", create = true }
+PROJECT_NAME = "{{ config_root | basename }}"
+
+[tasks."uv:reqs"]
+description = "Install dependencies from requirements file"
+alias = "uvr"
+run = "uv pip install -r requirements.txt"
+
+[tasks."uv:freeze"]
+description = "Create requirements.txt from currently installed modules"
+alias = "uvf"
+run = "uv pip freeze > requirements.txt"
+
+[tasks."uv:install"]
+description = "Install pip packages"
+alias = "uvi"
+run = "uv pip install"
+
+[tasks.info]
+description = "Print project information"
+run = '''
+echo "Project: $PROJECT_NAME"
+echo "Virtual Environment: $VIRTUAL_ENV"
+'''

apps/cos-event-trigger/Dockerfile

@@ -0,0 +1,39 @@
+# Stage 1: Build stage
+FROM python:3.12-slim AS builder
+
+WORKDIR /app
+
+# Install venv
+RUN python -m venv /opt/venv
+# Activate venv
+ENV PATH="/opt/venv/bin:$PATH"
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir --upgrade pip && \
+    pip install --no-cache-dir -r requirements.txt
+
+# Stage 2: Runtime stage
+FROM python:3.12-slim
+
+WORKDIR /app
+
+# Create non-root user
+RUN adduser --disabled-password --gecos "" appuser
+
+# Copy virtual environment from builder stage
+COPY --from=builder /opt/venv /opt/venv
+
+# Set environment variables
+ENV PATH="/opt/venv/bin:$PATH" \
+    PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1
+
+# Copy application code
+COPY app.py .
+
+# Switch to non-root user
+USER appuser
+
+EXPOSE 8080
+
+CMD ["python", "app.py"]

apps/cos-event-trigger/app.py

@@ -0,0 +1,53 @@
+#!/usr/bin/env python3
+from flask import Flask, request, jsonify
+import json
+from datetime import datetime
+
+app = Flask(__name__)
+
+# Equivalent to EventStats in Go
+class EventStats:
+    def __init__(self):
+        self.by_bucket = {}
+        self.by_type = {}
+        self.by_object = {}
+    
+    def to_dict(self):
+        return {
+            "by_bucket": self.by_bucket,
+            "by_type": self.by_type,
+            "by_object": self.by_object
+        }
+
+# Global stats object
+stats = EventStats()
+
+@app.route('/stats', methods=['GET'])
+def get_stats():
+    return jsonify(stats.to_dict())
+
+@app.route('/', methods=['POST'])
+def handle_event():
+    current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+    body = request.data.decode('utf-8')
+    
+    # Parse the event data
+    event = json.loads(body)
+    
+    # Update stats
+    bucket = event.get('bucket', 'unknown')
+    operation = event.get('operation', 'unknown')
+    key = event.get('key', 'unknown')
+    
+    stats.by_bucket[bucket] = stats.by_bucket.get(bucket, 0) + 1
+    stats.by_type[operation] = stats.by_type.get(operation, 0) + 1
+    stats.by_object[key] = stats.by_object.get(key, 0) + 1
+    
+    print(f"{current_time} - Received:")
+    print(f"\nBody: {body}")
+    
+    return "OK"
+
+if __name__ == '__main__':
+    print("Listening on port 8080")
+    app.run(host='0.0.0.0', port=8080)