Adds S3-compatible storage support

Enables the connector to support S3-compatible storage services
like MinIO, Wasabi, DigitalOcean Spaces, and Backblaze B2, by
allowing users to set a custom endpoint URL.

Also updates the Quix platform link.

Author: tomas-quix

python/destinations/quixlake-timeseries/README.md

@@ -13,7 +13,7 @@ This connector consumes time-series data from a Kafka topic and writes it to S3
 
 ## How to run
 
-Create a [Quix](https://portal.platform.quix.io/signup?xlink=github) account or log in and visit the `Connectors` tab to use this connector.
+Create a [Quix](https://portal.cloud.quix.io/signup?xlink=github) account or log in and visit the `Connectors` tab to use this connector.
 
 Clicking `Set up connector` allows you to enter your connection details and runtime parameters.
 
@@ -44,8 +44,13 @@ Then either:
 - **`AWS_REGION`**: AWS region for S3 bucket
   *Default*: `us-east-1`
 
-- **`AWS_ENDPOINT_URL`**: S3 endpoint URL (for non-AWS S3-compatible storage like MinIO)
-  *Default*: None
+- **`AWS_ENDPOINT_URL`**: Custom S3 endpoint URL for non-AWS S3-compatible storage
+  *Examples*:
+  - MinIO: `http://minio.example.com:9000`
+  - Wasabi: `https://s3.wasabisys.com`
+  - DigitalOcean Spaces: `https://nyc3.digitaloceanspaces.com`
+  - Backblaze B2: `https://s3.us-west-004.backblazeb2.com`
+  *Default*: None (uses AWS S3)
 
 ### Data Organization
 
@@ -126,6 +131,46 @@ HIVE_COLUMNS=
 ```
 Creates: `s3://bucket/prefix/table/data_*.parquet`
 
+## Using Non-AWS S3-Compatible Storage
+
+This connector supports any S3-compatible storage service by setting the `AWS_ENDPOINT_URL` environment variable.
+
+### MinIO Example
+```bash
+AWS_ENDPOINT_URL=http://minio.example.com:9000
+AWS_ACCESS_KEY_ID=minioadmin
+AWS_SECRET_ACCESS_KEY=minioadmin
+AWS_REGION=us-east-1
+S3_BUCKET=my-data-lake
+```
+
+### Wasabi Example
+```bash
+AWS_ENDPOINT_URL=https://s3.wasabisys.com
+AWS_ACCESS_KEY_ID=your-wasabi-access-key
+AWS_SECRET_ACCESS_KEY=your-wasabi-secret-key
+AWS_REGION=us-east-1
+S3_BUCKET=my-data-lake
+```
+
+### DigitalOcean Spaces Example
+```bash
+AWS_ENDPOINT_URL=https://nyc3.digitaloceanspaces.com
+AWS_ACCESS_KEY_ID=your-spaces-access-key
+AWS_SECRET_ACCESS_KEY=your-spaces-secret-key
+AWS_REGION=nyc3
+S3_BUCKET=my-data-lake
+```
+
+### Backblaze B2 Example
+```bash
+AWS_ENDPOINT_URL=https://s3.us-west-004.backblazeb2.com
+AWS_ACCESS_KEY_ID=your-b2-key-id
+AWS_SECRET_ACCESS_KEY=your-b2-application-key
+AWS_REGION=us-west-004
+S3_BUCKET=my-data-lake
+```
+
 ## Architecture
 
 The sink uses a batching architecture for high throughput:
@@ -138,7 +183,9 @@ The sink uses a batching architecture for high throughput:
 
 ## Requirements
 
-- S3 bucket access (AWS S3 or S3-compatible storage)
+- S3 bucket access:
+  - AWS S3, or
+  - Any S3-compatible storage (MinIO, Wasabi, DigitalOcean Spaces, Backblaze B2, etc.)
 - Optional: Quix REST Catalog endpoint for data catalog integration
 
 ## Contribute

python/destinations/quixlake-timeseries/main.py

@@ -50,6 +50,10 @@ def parse_hive_columns(columns_str: str) -> list:
     s3_bucket=os.environ["S3_BUCKET"],
     s3_prefix=os.getenv("S3_PREFIX", "data"),
     table_name=table_name,
+    aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
+    aws_secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY"),
+    aws_region=os.getenv("AWS_REGION", "us-east-1"),
+    s3_endpoint_url=os.getenv("AWS_ENDPOINT_URL"),
     hive_columns=hive_columns,
     timestamp_column=os.getenv("TIMESTAMP_COLUMN", "ts_ms"),
     catalog_url=os.getenv("CATALOG_URL"),

python/destinations/quixlake-timeseries/quixlake_sink.py

@@ -8,7 +8,6 @@
 import time
 import logging
 import uuid
-import os
 from typing import List, Dict, Any, Optional
 from datetime import datetime, timezone
 from io import BytesIO
@@ -36,6 +35,10 @@ def __init__(
         s3_bucket: str,
         s3_prefix: str,
         table_name: str,
+        aws_access_key_id: Optional[str] = None,
+        aws_secret_access_key: Optional[str] = None,
+        aws_region: str = "us-east-1",
+        s3_endpoint_url: Optional[str] = None,
         hive_columns: List[str] = None,
         timestamp_column: str = "ts_ms",
         catalog_url: Optional[str] = None,
@@ -52,6 +55,11 @@ def __init__(
             s3_bucket: S3 bucket name
             s3_prefix: S3 prefix/path for data files
             table_name: Table name for registration
+            aws_access_key_id: AWS access key ID
+            aws_secret_access_key: AWS secret access key
+            aws_region: AWS region (default: "us-east-1")
+            s3_endpoint_url: Custom S3 endpoint URL for non-AWS S3-compatible storage
+                           (e.g., MinIO, Wasabi, DigitalOcean Spaces)
             hive_columns: List of columns to use for Hive partitioning. Include 'year', 'month',
                          'day', 'hour' to extract these from timestamp_column
             timestamp_column: Column containing timestamp to extract time partitions from
@@ -62,10 +70,10 @@ def __init__(
             auto_create_bucket: if True, create bucket in S3 if missing.
             max_workers: Maximum number of parallel upload threads (default: 10)
         """
-        self._aws_region = os.getenv('AWS_REGION', 'us-east-1')
-        self._aws_access_key_id = os.getenv("AWS_ACCESS_KEY_ID")
-        self._aws_secret_access_key = os.getenv("AWS_SECRET_ACCESS_KEY")
-        self._aws_endpoint_url = os.getenv("AWS_ENDPOINT_URL", None)
+        self._aws_region = aws_region
+        self._aws_access_key_id = aws_access_key_id
+        self._aws_secret_access_key = aws_secret_access_key
+        self._aws_endpoint_url = s3_endpoint_url
         self._credentials = {
             "region_name": self._aws_region,
             "aws_access_key_id": self._aws_access_key_id,
@@ -100,6 +108,9 @@ def setup(self):
         logger.info(f"S3 Target: s3://{self.s3_bucket}/{self.s3_prefix}/{self.table_name}")
         logger.info(f"Partitioning: hive_columns={self.hive_columns}")
 
+        if self._aws_endpoint_url:
+            logger.info(f"Using custom S3 endpoint: {self._aws_endpoint_url}")
+
         if self._catalog and self.auto_discover:
             logger.info(f"Table will be auto-registered in REST Catalog on first write")