Skip to main content

Quixplus is a Python library for building sources and sinksfor Quix Streams.

Project description

Quixplus

This project contains custom data sources for streaming data to QuixStreams using:

  • WebSocket: Streams data from WebSocket endpoints.
  • CSV: Streams data from CSV files.
  • HTTP: Polls data from HTTP APIs at scheduled intervals.

Each source is designed to integrate with QuixStreams for real-time data streaming in Python applications.

Table of Contents

Getting Started

This project requires:

  • Python 3.8 or higher
  • QuixStreams library
  • Access to your data sources (WebSocket endpoints, HTTP APIs, or CSV files)

Environment Setup

Clone this repository:

git clone https://github.com/your-username/your-repo.git
cd your-repo

Install the dependencies:

pip install -r requirements.txt

Set up your environment variables:

To use the application factory method, you need to set up your environment variables.

export BOOTSTRAP_SERVERS=<your-bootstrap-servers>
export SASL_USERNAME=<your-username>
export SASL_PASSWORD=<your-password>
export SASL_MECHANISM=<your-sasl-mechanism>
export SECURITY_PROTOCOL=<your-security-protocol>

Data Sources

WebSocket Source

The WebSocketSource class connects to a WebSocket endpoint, receives data, applies optional transformations, and streams it to QuixStreams.

Key Features:

  • Connects to WebSocket URLs.
  • Validates and transforms incoming messages.
  • Sends data to a specified Kafka topic.

CSV Source

The CSVSource class reads data from a CSV file, transforms each row, and streams it to QuixStreams.

Key Features:

  • Reads from CSV files with configurable delimiters.
  • Extracts each row as a JSON object, using column headers as keys.
  • Supports configurable keys for Kafka messages and value serialization.

HTTP Source

The AIOHTTPSource class polls an HTTP API endpoint at specified intervals, validates the data, and streams it to QuixStreams.

Key Features:

  • Asynchronous polling with configurable intervals.
  • Flexible authentication (Bearer token, Basic, or custom headers).
  • Supports configurable JSON path extraction for Kafka keys and values.

Installation:

Install from PyPI:

pip install quixplus

Add your environment variables to the .env file.

Usage

WebSocket Source Example

To use the WebSocket source, initialize it with the necessary configurations:

from quixstreams import Application
from custom_sources.websocket_source import WebSocketSource

app = Application(broker_address="your_kafka_broker")
ws_source = WebSocketSource(
    topic="your_topic",
    ws_url="wss://your_websocket_endpoint",
    auth_payload={"api_key": "your_key"},
    subscribe_payload={"action": "subscribe", "params": "A.*"}
)
sdf = app.dataframe(source=ws_source)
sdf.print()
app.run()

CSV Source Example

To use the CSV source, provide the path to your CSV file and specify the columns for Kafka keys:

from quixstreams import Application
from custom_sources.csv_source import CSVSource

app = Application(broker_address="your_kafka_broker")
csv_source = CSVSource(
    topic="your_topic",
    csv_path="path/to/your/csv/file.csv",
    key_columns=["column1", "column2"]
)
sdf = app.dataframe(source=csv_source)
sdf.print()
app.run()

HTTP Source Example

To use the HTTP source, set the URL and polling interval:

from quixstreams import Application
from custom_sources.http_source import AIOHTTPSource

app = Application(broker_address="your_kafka_broker")
http_source = AIOHTTPSource(
    url="https://your-api-endpoint",
    poll_interval=10,
    auth_type="bearer",
    auth_credentials="your_bearer_token",
    key_json_path="$.data.id",
    value_json_path="$.data"
)
sdf = app.dataframe(source=http_source)
sdf.print()
app.run()

Contributing

We welcome contributions! Please read our CONTRIBUTING.md for guidelines on contributing to this project.

License

This project is licensed under the MIT License.

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

quixplus-2.0.0.tar.gz (12.0 kB view details)

Uploaded Source

Built Distribution

quixplus-2.0.0-py3-none-any.whl (11.4 kB view details)

Uploaded Python 3

File details

Details for the file quixplus-2.0.0.tar.gz.

File metadata

  • Download URL: quixplus-2.0.0.tar.gz
  • Upload date:
  • Size: 12.0 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: uv/0.4.29

File hashes

Hashes for quixplus-2.0.0.tar.gz
Algorithm Hash digest
SHA256 cbaf2543f38c4faf12a374efe83b4ca2b2838d65aef7542847fda81c5feb9e0b
MD5 9fdf906c69298ca84eb17eab8cf1a3a6
BLAKE2b-256 b10738f99f4f1ec65de612282ca165b0ad77a2071a35b728518859111f41975f

See more details on using hashes here.

File details

Details for the file quixplus-2.0.0-py3-none-any.whl.

File metadata

File hashes

Hashes for quixplus-2.0.0-py3-none-any.whl
Algorithm Hash digest
SHA256 6721cb21f1999b79de9ec4ac4e489592c2c4b019a7ed57a837428bd088ed8bda
MD5 ae21ce48539557396c2c21df74fb0804
BLAKE2b-256 d9e47aefb04cf3fd620b6280c1740b257510f6078ebca209b8e4ff9bf10fd41a

See more details on using hashes here.

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page