Overview

QuantumDatalytica empowers Developers to build modular Quantum Machines, enables Workflow Engineers to design powerful data workflows, and allows Workflow Users to execute and automate tasks seamlessly, all within a secure, scalable, and cloud-native platform.

Workflows are created using DAG-based design, with parent-child machine dependencies and conditional execution logic.

QuantumDatalytica as a unified platform for workflow automation, data processing, and task orchestration using modular, containerized units called Quantum Machines.

Marketplace, the central repository where machines are published, versioned, and reused across workflows and teams, reducing duplication of logic.

Provide a short onboarding path for each role:

• Developers: Setup CLI, create machine, submit for review, publish
• Engineers: Use Data factory and Workflow Designer Screen to design workflows
• Users: Run workflows, track results, view logged usage of infrastructure

Overview

A Machine Developer is a core contributor to the QuantumDatalytica platform, responsible for building, testing, and publishing self-contained units of logic called Quantum Machines. These machines are the foundational building blocks of automated data workflows. They are designed to perform a specific task, such as transforming data, calling APIs, running analytics, or generating documents  and are built with consistency, reusability, and scalability in mind.

Machine Developers primarily work in Python, structure their logic around a standardized interface (CoreEngine), and package everything into lightweight Quantum Blocks. Each Quantum Machine must adhere to a defined project.json, input/output schema and is validated through rigorous testing before publication. These machines are then made available to Workflow Engineers and Users, who can plug them into larger Quantum workflows or trigger them as standalone automation tasks.

The role is both creative and technical. Developers have the freedom to innovate, turning domain expertise or unique algorithms into shareable, monetizable machines, while working within a well-structured ecosystem. With Quantum-Machine CLI tools, testing environments, and version control built into the platform, developers can confidently deploy machines that scale across environments.

Whether you're solving a data challenge, building a PDF generator, or integrating external APIs, as a Machine Developer, your work directly empowers automated decision-making and workflow efficiency at scale.

Before you begin building a Quantum Machine, it's essential to set up a local development environment that mirrors the runtime environment of the QuantumDatalytica platform. This ensures compatibility, smooth testing, and predictable deployment behavior.

Required Tools and Software

Once these tools are installed, you're fully equipped to create and run your first Quantum Machine.

Overview

Quantum-Machine CLI is a command-line interface developed by QuantumDatalytica LLC to help developers build, run, test, log, and manage modular analytics components called Quantum Machines. These machines are the foundation of scalable, distributed data workflows within the QuantumDatalytica ecosystem. The CLI streamlines local development and ensures consistent behavior across environments.

Installation

pip install quantum-machine

Copy

Usage

quantum --help

Copy

Commands

Example Commands

Initialize a machine

quantum init machine HelloWorld

Copy

Creates

• HelloWorld/main.py
• HelloWorld/Project.json
• HelloWorld/requirements.txt
• HelloWorld/Dockerfile
• HelloWorld/input.json
• HelloWorld/output.json

Update requirements.txt

git+https://github.com/QuantumDatalytica-LLC/quantum-core-engine.git@<latest_version>

Copy

Open the generated requirements.txt and update it to include the required quantum-core-engine version.

Install dependencies (no cache)

pip install --no-cache-dir --force-reinstall -r HelloWorld/requirements.txt

Copy

Install all required Python packages without using cache to ensure a clean environment.

Run the machine

quantum run machine HelloWorld

Copy

Build the machine as Docker Image

quantum build machine HelloWorld

Copy

Builds a Docker image with dependencies for the machine.

Test your machine

quantum test machine HelloWorld

Copy

Runs the test suite defined under the machine's directory.

Lint your machine

quantum lint machine HelloWorld

Copy

Applies flake8 or equivalent linting tools to maintain code standards.

Validate machine structure

quantum validate machine HelloWorld\<file_name>

Copy

Ensures the machine has the correct Project.json, required fields, and structure.

Create a Workflow

quantum init workflow my_workflow

Copy

Creates a workflow.yaml file to define machine dependencies.

Add DAG Machine to Workflow

quantum add machine HelloWord -w my_workflow
quantum add machine 2nd_Machine -w my_workflow
quantum add machine 3rd_Machine -p HelloWorld --workflow  my_workflow
quantum add machine 4th_Machine -parent 2nd_Machine -w my_workflow
quantum add machine 5th_Machine -p 3rd_Machine 4th_Machine -w my_workflow

Copy

Run a Workflow

quantum run workflow my_workflow

Copy

Executes machines in the correct DAG order as defined in workflow.yaml.

View machine logs

quantum logs machine HelloWord

Copy

Displays the logs from the most recent execution of the HelloWorld machine.

Overview

The quantum-core-engine is a required Python package that provides the foundational structure for all Quantum Machines. It must be installed in your development environment before writing or testing any machine logic

pip install

https://github.com/QuantumDatalytica-LLC/quantum-core-engine.git@<latest_version>

Copy

Run this command inside your virtual environment or project directory to keep dependencies clean and isolated. Replace <latest_version> with the latest version available on the Quantum-Core-Engine Releases page.

Objective

The Quantum-Core-Engine is more than just a library, it defines how every Quantum Machine behaves. Its goals include:

By using CoreEngine, developers don’t have to reinvent basic execution logic or worry about input/output parsing. They can focus purely on building business logic, while the framework handles the rest, making machines easier to validate, test, and integrate into Quantum Workflows.

How to use Quantum-Core-Engine

from quantum.CoreEngine import CoreEngine

class MyMachine(CoreEngine):
"""
Inherit from CoreEngine to define your machine's logic.
Implement the required methods such as 'receiving', ‘pre_processing’ 
and many more to handle input and produce output.
"""

if __name__ == "__main__":
# Initialize and execute machine
machine = MyMachine()
machine.start()
                                                    

Copy

Breakdown of Key Elements

• from quantum.CoreEngine import CoreEngine:
  Imports the base class required for all Quantum Machines.
• class MyMachine(CoreEngine):
  Subclasses CoreEngine to inherit standardized structure, input/output flow, error handling, and lifecycle of execution flow.
• if __name__ == "__main__":
  Ensures the script can be run directly during local testing or inside Quantum Cluster.
  machine.start() initializes the engine and runs it in Quantum Workflow.

Before writing any code, it's important to understand what a Quantum Machine really is: a self-contained, containerized Python base code that takes structured input, performs a task, and returns structured output. Each machine follows a common design pattern, making it predictable, testable, and reusable.

Development starts by using the Quantum Machine CLI to scaffold a new machine with a predefined structure. You'll write your logic in main.py, prepare project.json which defines key parameters of the Quantum Machine, and input.json specifies the expected input parameters for the Quantum Machine, while output.json defines the structure and content of the machine's resulting output.

Use CLI commands to test your machine locally, simulating its execution within the actual Quantum Datalytica runtime environment.

Machine Scaffold

HelloWord/

• main.py # Core logic of quantum-machine
• Project.json# Machine metadata
• input.json# Example input for testing
• output.json# Expected output structure
• Dockerfile# Cluster container definition
• requirements.txt# Python dependencies
• README.md# Documentation

Sample Machine (main.py)

from quantum.CoreEngine import CoreEngine

class MyMachine(CoreEngine):

    input_data = {}
    dependent_machine_data = {}

    def receiving(self, input_data, dependent_machine_data, callback):
        """Receiving
        :param input_data: Configure parameter values
        :param dependent_machine_data: Dependant/Previous machine data values
        :return: callback method to pass data and error into next step
        """
        data = {}
        error_list = []
        try:
            # Review Final data and Error list
            data = self.get_final_data()
            error_list = self.get_error_list()

            self.input_data = input_data
            self.dependent_machine_data = dependent_machine_data

        except Exception as e:
            err_msg = f"Error : {str(e)}"
            error_list.append(err_msg)

        finally:
            callback(data, error_list)

    def pre_processing(self, callback):
        """Pre-Processing
        :return: callback method to pass data and error into next step
        """
        data = {}
        error_list = []
        try:
            # Updated Final data and Error list
            data = self.get_final_data()
            error_list = self.get_error_list()

        except Exception as e:
            err_msg = f"Error : {str(e)}"
            error_list.append(err_msg)

        finally:
            callback(data, error_list)

    def processing(self, callback):
        """Processing
        :return: callback method to pass data and error into next step
        """
        data = {}
        error_list = []
        try:
            # Updated Final data and Error list
            data = self.get_final_data()
            error_list = self.get_error_list()

        except Exception as e:
            err_msg = f"Error : {str(e)}"
            error_list.append(err_msg)

        finally:
            callback(data, error_list)

    def post_processing(self, callback):
        """Post-Processing
        :return: callback method to pass data and error into next step
        """
        data = {}
        error_list = []
        try:
            # Updated Final data and Error list
            data = self.get_final_data()
            error_list = self.get_error_list()

        except Exception as e:
            err_msg = f"Error : {str(e)}"
            error_list.append(err_msg)

        finally:
            callback(data, error_list)

    def packaging_shipping(self, callback):
        """Packaging & Shipping
        :return: callback method to pass data and error into next step, This is 
        final data to use into next machine
        """
        data = {}
        error_list = []
        try:
            # Updated Final data and Error list
            data = self.get_final_data()
            error_list = self.get_error_list()

        except Exception as e:
            err_msg = f"Error : {str(e)}"
            error_list.append(err_msg)

        finally:
            callback(data, error_list)

if __name__ == '__main__':
    # Create a machine instance and start the process
    machine = MyMachine()
    machine.start()

Copy

Breakdown of key component

from quantum.CoreEngine import CoreEngine

Copy

This imports the CoreEngine base class from the Quantum-Core-Engine SDK. Inheriting from CoreEngine standardizes the structure, input/output flow, error handling, and lifecycle of your Quantum Machine.

input_data = {}Dependent_machine_data = {}

Copy

input_data: Holds the raw input parameters passed into the machine. These typically come from input.json or are set by the workflow engine.

dependent_machine_data: Captures output from any parent/previous machines in the workflow. Useful when a machine depends on upstream results.

receiving(self, input_data, dependent_machine_data, callback)

This method acts as the initial handshake for the machine’s execution. It captures the raw input and any output from parent machines, setting the stage for downstream logic.

pre_processing(self, callback)

Pre-processing prepares and transforms the initial inputs into a shape more suitable for the main computation. This stage is often used for validation, enrichment, and staging.

processing(self, callback)

This is the core execution phase where the machine performs its primary logic or task.

post_processing(self, callback)

In this phase, the machine performs final adjustments or logging after processing is complete, preparing the output for final packaging.

packaging_shipping(self, callback)

This is the last phase in the lifecycle — it prepares and packages the machine’s output into a structure compatible with QuantumDatalytica’s output schema and downstream machines.

if __name__ == '__main__':
machine = MyMachine()
machine.start()

Copy

This allows the machine to run when executed directly (locally or in a Quantum Cluster).

• start() method that initiates the lifecycle: receiving → pre-processing → processing → post-processing → packaging.

Overview

Quantum Machines often need to handle files such as data files, PDFs, Excel sheets, logs, or any file format. The Quantum-Core-Engine (QCE) provides a unified and developer-friendly interface for reading and writing files, while automatically handling:

• Small files inline (Base64-encoded, transparent to developer)
• Large files via efficient streaming (raw bytes, no Base64 overhead)
• Atomic writes (ensures data consistency)
• Simple, consistent APIs

Core Concepts

1. Unified API

• Use read_file(file_name, as_base64=False) to read any file.
• Use write_file(file_name, content, overwrite=True) to write any file.

The QCE decides automatically whether to inline (Base64) or stream (raw bytes) based on file size.

2. Automatic Switching

• Small files (< 10MB by default) → returned inline as Base64.
• Large files (≥ 10MB) → streamed as raw byte chunks.
• The threshold is configurable via _qce_max_inline_bytes.

3. Developer-Friendly Parameters

• No need to manage Base64 manually.
• Just set as_base64=True if you prefer raw bytes for small files too.
• Streaming for big files is automatic.

Sample Machine (read_file())

The read_file function provides a unified interface for reading files in a Quantum Machine environment. It automatically handles small files inline (e.g., base64-encoded JSON) and large files with streaming (chunked binary reading). This ensures developers can read files of any size efficiently without worrying about memory overhead.

Parameters

Return Value

The function returns a dictionary (dict) with metadata and file content. The structure varies slightly depending on whether the file was inlined or streamed.

Writing Files (write_file())

The write_file() method is the central file write API for Quantum Machines. It supports multiple input formats (raw bytes, base64, iterators, or existing file paths) and automatically decides the safest and most efficient way to persist the data.

Parameters

Return Value

The function always returns a dictionary containing metadata about the read operation, and depending on parameters, either inline content or a streaming reference.

Behavior Summary

• If file size ≤ inline_limit → returns data inline.
• If file size > inline_limit → switches to chunked streaming, returns chunks list instead of data.
• as_base64 toggles whether inline/chunk content is Base64 strings or raw bytes.

Examples

Write a small file (auto streaming)

self.write_file("hello.txt", data=b"Hello, Quantum!")

Copy

Write from Base64 (rarely needed manually)

content_b64 = base64.b64encode(b"My secret file").decode()
self.write_file("secret.txt", base64_data=content_b64)

Copy

Write a large file via streaming

def file_iterator(path, chunk_size=1024*1024):  # 1MB chunks
    with open(path, "rb") as f:
        while chunk := f.read(chunk_size):
            yield chunk

self.write_file("big_backup.zip", data_iter=file_iterator("local_backup.zip"))

Copy

The engine detects iterators and automatically performs a streaming write.

Developer Example

The following examples demonstrate how to use CoreEngine’s file management APIs in a Quantum Machine. These cover both small inline files and large streamed files.

Write a small file (string/bytes)

Upload Machine

from quantum.CoreEngine import CoreEngine

class MyMachine(CoreEngine):

    input_data = {}
    dependent_machine_data = {}

    def receiving(self, input_data, dependent_machine_data, callback):
        """Receiving
        :param input_data: Configure parameter values
        :param dependent_machine_data: Dependant/Previous machine data values
        :return: callback method to pass data and error into next step
        """
        data = {}
        error_list = []
        try:
            # Review Final data and Error list
            data = self.get_final_data()
            error_list = self.get_error_list()

        except Exception as e:
            err_msg = f"Error : {str(e)}"
            error_list.append(err_msg)
        finally:
            callback(data, error_list)

    def pre_processing(self, callback):
        """Pre-Processing
        :return: callback method to pass data and error into next step
        """
        data = {}
        error_list = []
        try:
            # Updated Final data and Error list
            data = self.get_final_data()
            error_list = self.get_error_list()


        except Exception as e:
            err_msg = f"Error : {str(e)}"
            error_list.append(err_msg)
        finally:
            callback(data, error_list)

    def processing(self, callback):
        """Processing
        :return: callback method to pass data and error into next step
        """
        data = {}
        error_list = []
        try:
            # Updated Final data and Error list
            data = self.get_final_data()
            error_list = self.get_error_list()

            def file_iterator(path, chunk_size=1024*1024):  # 1MB chunks
                with open(path, "rb") as f:
                    while chunk := f.read(chunk_size):
                        yield chunk

            res = self.write_file("UploadSampleFile.zip",
             data_iter=file_iterator("D:\SampleFile.zip"))
            data = res

        except Exception as e:
            err_msg = f"Error : {str(e)}"
            error_list.append(err_msg)
        finally:
            callback(data, error_list)

    def post_processing(self, callback):
        """Post-Processing
        :return: callback method to pass data and error into next step
        """
        data = {}
        error_list = []
        try:
            # Updated Final data and Error list
            data = self.get_final_data()
            error_list = self.get_error_list()


        except Exception as e:
            err_msg = f"Error : {str(e)}"
            error_list.append(err_msg)
        finally:
            callback(data, error_list)

    def packaging_shipping(self, callback):
        """Packaging & Shipping
        :return: callback method to pass data and error into next step,
         This is final data to use into next machine
        """
        data = {}
        error_list = []
        try:
            # Updated Final data and Error list
            data = self.get_final_data()
            error_list = self.get_error_list()


        except Exception as e:
            err_msg = f"Error : {str(e)}"
            error_list.append(err_msg)
        finally:
            callback(data, error_list)

if __name__ == '__main__':
    # Create a machine instance and start the process
    machine = MyMachine()
    machine.start()

Copy

Download Machine

from quantum.CoreEngine import CoreEngine

class MyMachine(CoreEngine):

    input_data = {}
    dependent_machine_data = {}

    # file_path get from previous machine or configure parameter
    file_path = None

    def receiving(self, input_data, dependent_machine_data, callback):
        """Receiving
        :param input_data: Configure parameter values
        :param dependent_machine_data: Dependant/Previous machine data values
        :return: callback method to pass data and error into next step
        """
        data = {}
        error_list = []
        try:
            # Review Final data and Error list
            data = self.get_final_data()
            error_list = self.get_error_list()

            self.input_data = input_data
            self.dependent_machine_data = dependent_machine_data

            self.file_path = self.input_data.get("file_path", None)
            if not self.file_path:
                err_msg = "File path is not provided."
                error_list.append(err_msg)

        except Exception as e:
            err_msg = f"Error : {str(e)}"
            error_list.append(err_msg)
        finally:
            callback(data, error_list)

    def pre_processing(self, callback):
        """Pre-Processing
        :return: callback method to pass data and error into next step
        """
        data = {}
        error_list = []
        try:
            # Updated Final data and Error list
            data = self.get_final_data()
            error_list = self.get_error_list()


        except Exception as e:
            err_msg = f"Error : {str(e)}"
            error_list.append(err_msg)
        finally:
            callback(data, error_list)

    def processing(self, callback):
        """Processing
        :return: callback method to pass data and error into next step
        """
        data = {}
        error_list = []
        try:
            # Updated Final data and Error list
            data = self.get_final_data()
            error_list = self.get_error_list()

            # Read chunk data
            resp = self.read_file(self.file_path)
            if resp.get("streamed"):
                with open("DownloadSampleFile.zip", "wb") as out:
                    for chunk in resp["iterator"]:
                        out.write(chunk)


        except Exception as e:
            err_msg = f"Error : {str(e)}"
            error_list.append(err_msg)
        finally:
            callback(data, error_list)

    def post_processing(self, callback):
        """Post-Processing
        :return: callback method to pass data and error into next step
        """
        data = {}
        error_list = []
        try:
            # Updated Final data and Error list
            data = self.get_final_data()
            error_list = self.get_error_list()


        except Exception as e:
            err_msg = f"Error : {str(e)}"
            error_list.append(err_msg)
        finally:
            callback(data, error_list)

    def packaging_shipping(self, callback):
        """Packaging & Shipping
        :return: callback method to pass data and error into next step,
         This is final data to use into next machine
        """
        data = {}
        error_list = []
        try:
            # Updated Final data and Error list
            data = self.get_final_data()
            error_list = self.get_error_list()


        except Exception as e:
            err_msg = f"Error : {str(e)}"
            error_list.append(err_msg)
        finally:
            callback(data, error_list)

if __name__ == '__main__':
    # Create a machine instance and start the process
    machine = MyMachine()
    machine.start()

Copy

With these utilities, Quantum Machine Developers can handle any file size, any format, safely and efficiently without needing to think about encoding or streaming logistics.