Dream

aws-iceberg-sink/app.yaml

@@ -33,4 +33,3 @@ variables:
 dockerfile: dockerfile
 runEntryPoint: main.py
 defaultFile: main.py
-libraryItemId: s3-iceberg-destination

http-api-source/README.md

@@ -0,0 +1,33 @@
+# HTTP API Source
+
+[This code sample](https://github.com/quixio/quix-samples/tree/main/python/sources/http_source) demonstrates how to run a Flask HTTP API as a web gateway and use it to publish to a Kafka topic via HTTP POST requests.
+
+## How to run
+
+Create a [Quix](https://portal.platform.quix.io/#?xlink=github) account or log-in and visit the Samples to use this project
+
+Clicking `Edit code` on the Sample, forks the project to your own Git repo so you can customize it before deploying.
+
+You can test your endpoint by sending a message via curl:
+`curl -X POST -H "Content-Type: application/json" -d '{"sessionId": "000001", "name": "Tony Hawk", "purchase": "skateboard" }' https://<your-deployment-url>/data/
+`
+
+## Environment variables
+
+The code sample uses the following environment variables:
+
+- **output**: This is the output topic for hello world data.
+
+## Contribute
+
+Submit forked projects to the Quix [GitHub](https://github.com/quixio/quix-samples) repo. Any new project that we accept will be attributed to you and you'll receive $200 in Quix credit.
+
+## Open source
+
+This project is open source under the Apache 2.0 license and available in our [GitHub](https://github.com/quixio/quix-samples) repo.
+
+Please star us and mention us on social to show your appreciation.
+
+
+
+image: Flaticon.com

http-api-source/app.yaml

@@ -0,0 +1,12 @@
+name: http-api-source
+language: python
+variables:
+  - name: output
+    inputType: OutputTopic
+    description: This is the output topic for hello world data
+    defaultValue: http-source
+    required: true
+dockerfile: dockerfile
+runEntryPoint: main.py
+defaultFile: main.py
+libraryItemId: web-api-gateway

http-api-source/dockerfile

@@ -0,0 +1,28 @@
+FROM python:3.12.5-slim-bookworm
+
+# Set environment variables for non-interactive setup and unbuffered output
+ENV DEBIAN_FRONTEND=noninteractive \
+    PYTHONUNBUFFERED=1 \
+    PYTHONIOENCODING=UTF-8 \
+    PYTHONPATH="/app"
+
+# Build argument for setting the main app path
+ARG MAINAPPPATH=.
+
+# Set working directory inside the container
+WORKDIR /app
+
+# Copy requirements to leverage Docker cache
+COPY "${MAINAPPPATH}/requirements.txt" "${MAINAPPPATH}/requirements.txt"
+
+# Install dependencies without caching
+RUN pip install --no-cache-dir -r "${MAINAPPPATH}/requirements.txt"
+
+# Copy entire application into container
+COPY . .
+
+# Set working directory to main app path
+WORKDIR "/app/${MAINAPPPATH}"
+
+# Define the container's startup command
+ENTRYPOINT ["python3", "main.py"]

http-api-source/main.py

@@ -0,0 +1,110 @@
+import os
+import datetime
+import json
+from flask import Flask, request, Response, redirect
+from flasgger import Swagger
+from waitress import serve
+import time
+
+from flask_cors import CORS
+
+from setup_logging import get_logger
+from quixstreams import Application
+
+# for local dev, load env vars from a .env file
+from dotenv import load_dotenv
+load_dotenv()
+
+service_url = os.environ["Quix__Deployment__Network__PublicUrl"]
+
+quix_app = Application()
+topic = quix_app.topic(os.environ["output"])
+producer = quix_app.get_producer()
+
+logger = get_logger()
+
+app = Flask(__name__)
+
+# Enable CORS for all routes and origins by default
+CORS(app)
+
+app.config['SWAGGER'] = {
+    'title': 'HTTP API Source',
+    'description': 'Test your HTTP API with this Swagger interface. Send data and see it arrive in Quix.',
+    'uiversion': 3
+}
+
+swagger = Swagger(app)
+
+@app.route("/", methods=['GET'])
+def redirect_to_swagger():
+    return redirect("/apidocs/")
+
+@app.route("/data/", methods=['POST'])
+def post_data_without_key():
+    """
+    Post data without key
+    ---
+    parameters:
+      - in: body
+        name: body
+        schema:
+          type: object
+          properties:
+            some_value:
+              type: string
+    responses:
+      200:
+        description: Data received successfully
+    """
+    data = request.json
+    logger.debug(f"{data}")
+
+    producer.produce(topic.name, json.dumps(data))
+
+    # Return a normal 200 response; CORS headers are added automatically by Flask-CORS 
+    return Response(status=200)
+
+@app.route("/data/<key>", methods=['POST'])
+def post_data_with_key(key: str):
+    """
+    Post data with a key
+    ---
+    parameters:
+      - in: path
+        name: key
+        type: string
+        required: true
+      - in: body
+        name: body
+        schema:
+          type: object
+          properties:
+            some_value:
+              type: string
+    responses:
+      200:
+        description: Data received successfully
+    """
+    data = request.json
+    logger.debug(f"{data}")
+
+    producer.produce(topic.name, json.dumps(data), key.encode())
+
+    return Response(status=200)
+
+if __name__ == '__main__':
+    print("=" * 60)
+    print(" " * 20 + "CURL EXAMPLE")
+    print("=" * 60)
+    print(
+        f"""
+curl -L -X POST \\
+    -H 'Content-Type: application/json' \\
+    -d '{{"key": "value"}}' \\
+    {service_url}/data
+    """
+    )
+    print("=" * 60)
+
+    serve(app, host="0.0.0.0", port=80)

http-api-source/requirements.txt

@@ -0,0 +1,6 @@
+quixstreams==3.2.1
+flask
+flask_cors
+flasgger==0.9.7b2
+waitress
+python-dotenv

http-api-source/setup_logging.py

@@ -0,0 +1,23 @@
+import logging
+
+
+def get_logger():
+
+    logging.basicConfig(
+        level=logging.DEBUG,
+        format='[%(asctime)s] [%(levelname)s]: %(message)s',
+        datefmt='%Y-%m-%d %H:%M:%S'
+    )
+
+    # Set up logging
+    logger = logging.getLogger('waitress')
+    logger.setLevel(logging.DEBUG)  # Set to DEBUG for more detailed output
+    logger.propagate = False  # Prevent the log messages from propagating to the root logger
+
+    # Create handlers (console and file handler for example)
+    console_handler = logging.StreamHandler()
+    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+    console_handler.setFormatter(formatter)
+    logger.addHandler(console_handler)
+
+    return logger

http-transformation/README.md

@@ -0,0 +1,30 @@
+# Event Detection
+
+[This code sample](https://github.com/quixio/quix-samples/tree/main/python/transformations/Event-Detection) demonstrates how to consume data from a Kafka topic, filter the data based on some criteria and publish the result to an output topic.
+
+It shows how to make real-time decisions on your data by looking at the `Brake` value and if a condition is met will output JSON message to the output topic.
+
+You can adapt this code to suit your needs.
+
+## How to run
+
+Create a [Quix](https://portal.platform.quix.io/#?xlink=github) account or log-in and visit the Samples to use this project.
+
+Clicking `Edit code` on the Sample, forks the project to your own Git repo so you can customize it before deploying.
+
+## Environment variables
+
+The code sample uses the following environment variables:
+
+- **input**: This is the input topic for f1 data.
+- **output**: This is the output topic for hard braking events.
+
+## Contribute
+
+Submit forked projects to the Quix [GitHub](https://github.com/quixio/quix-samples) repo. Any new project that we accept will be attributed to you and you'll receive $200 in Quix credit.
+
+## Open source
+
+This project is open source under the Apache 2.0 license and available in our [GitHub](https://github.com/quixio/quix-samples) repo.
+
+Please star us and mention us on social to show your appreciation.

http-transformation/app.yaml

@@ -0,0 +1,18 @@
+name: HTTP Transformation
+language: python
+variables:
+  - name: input
+    inputType: InputTopic
+    description: This is the input topic for f1 data
+    defaultValue: http-source
+    required: true
+  - name: output
+    inputType: OutputTopic
+    multiline: false
+    description: This is the output topic for hard braking events
+    defaultValue: sensor-data-table
+    required: true
+dockerfile: dockerfile
+runEntryPoint: main.py
+defaultFile: quix_function.py
+libraryItemId: event-detection-transformation

http-transformation/dockerfile

@@ -0,0 +1,28 @@
+FROM python:3.12.5-slim-bookworm
+
+# Set environment variables for non-interactive setup and unbuffered output
+ENV DEBIAN_FRONTEND=noninteractive \
+    PYTHONUNBUFFERED=1 \
+    PYTHONIOENCODING=UTF-8 \
+    PYTHONPATH="/app"
+
+# Build argument for setting the main app path
+ARG MAINAPPPATH=.
+
+# Set working directory inside the container
+WORKDIR /app
+
+# Copy requirements to leverage Docker cache
+COPY "${MAINAPPPATH}/requirements.txt" "${MAINAPPPATH}/requirements.txt"
+
+# Install dependencies without caching
+RUN pip install --no-cache-dir -r "${MAINAPPPATH}/requirements.txt"
+
+# Copy entire application into container
+COPY . .
+
+# Set working directory to main app path
+WORKDIR "/app/${MAINAPPPATH}"
+
+# Define the container's startup command
+ENTRYPOINT ["python3", "main.py"]

http-transformation/main.py

@@ -0,0 +1,45 @@
+import os
+from quixstreams import Application
+
+# for local dev, load env vars from a .env file
+from dotenv import load_dotenv
+load_dotenv()
+
+app = Application(consumer_group="v1.1", 
+            auto_offset_reset="earliest", 
+            use_changelog_topics=False)
+
+input_topic = app.topic(os.environ["input"], key_deserializer="str")
+output_topic = app.topic(os.environ["output"])
+
+sdf = app.dataframe(input_topic)
+
+sdf = sdf.apply(lambda row: row["payload"], expand=True)
+
+# Calculate hopping window of 5 seconds with 10 second buffer delay.
+sdf = sdf.tumbling_window(1000, 10000).collect().final()
+
+sdf = sdf.apply(lambda row: sorted(row["value"], key=lambda row: row["time"]), expand=True)
+
+def transpose(row, key, *_):
+
+    for axis, value in row["values"].items():
+        yield {
+            "device_id": key,
+            "sensor": row["name"],
+            "axis": axis,
+            "location": "na",
+            "timestamp": row["time"],
+            "value": value
+        }
+
+sdf = sdf.apply(transpose, metadata=True, expand=True) 
+
+# Print JSON messages in console.
+sdf.print_table(metadata=False)
+
+# Send the message to the output topic
+sdf.to_topic(output_topic)
+
+if __name__ == "__main__":
+    app.run()

http-transformation/requirements.txt

@@ -0,0 +1,2 @@
+quixstreams==3.13.1
+python-dotenv