Library Functionality

Last Updated: June 25, 2026

Logging

import careflow
careflow.logger.debug('testing')
# 2024-08-08 11:21:54,697 [sub] DEBUG | testing

Careflow includes a built-in logging mechanism for tracking events, errors, and processing steps.

• Default log file: careflow.log (configured via logger.file), written to the working directory /careflow
• Log level is configured via logger.level.

To check logs inside a running container:

docker logs -f <container_name>
docker exec -it <container_name> cat /careflow/careflow.log

Configuration

import careflow
config = careflow.config.get()
# Returns a Munch (attribute-accessible dict) of the full config, e.g.: 
# config.dbroker.BASE_URL
# config.environment
# config.ws_gateway.BASE_URL

storage = careflow.storage
# Munch({'database': [], 'block': [], 'blob': []})

careflow.config.get() returns the full configuration object. Refer to Section 2: Configuration for the complete schema and field reference. careflow.storage is a shortcut to config.get().storage.

Data Communication

The careflow.start() function initializes the Careflow service and begins handling incoming events.

Callback and Function

Start the webhook server and set up the callback:

import careflow

def my_callback(data):
   careflow.logger.info(f"Received data: {data}")

def my_control_callback(request):
   careflow.logger.info(f"Received controller request: {request}")

careflow.start(
    callback=my_callback,
    control_callback=my_control_callback
)

careflow.start(callback, sample_parser=None, control_callback=None)

This function initializes and starts the appropriate connector based on the environment configuration. It runs in the foreground (blocking).

Parameters:

callback (required):

• The callback(s) to handle incoming data from subscribed topics.
• Each callback receives one argument: the data payload

sample_parser (optional):

• The parser/template values for the sample connector (local mode). If not provided, defaults to an empty dict. Used to substitute placeholders in the sample data template. May be a single dict or a list of dicts (iterated per interval; loops if local.LOOP is true).

control_callback (optional):

• A callback invoked with control messages received on dbroker.CONTROL_TOPIC.

Start data through the Data Broker when complete:

import careflow

careflow.send({
   "event": "alert",
   "status": "active"
})

careflow.send(data)

Sends data to the Data Broker or writes it to the Sample Output, depending on the environment.

• If the environment is set to local and dbroker.ENABLED == false:
  • The data is appended to the sample output file <CAREFLOW_CONFIG_PATH>/<PUB_TOPIC>.json.
• For other environments:
  • The data is published to the Data Broker on dbroker.PUB_TOPIC.

Parameters

data (required):

• A JSON-serializable object (dict/list) to send.

Example – Multi-topic setup

import careflow

def callback_input1(data):
    careflow.logger.info(f"[input_1] {data}")

def callback_input2(data):
    careflow.logger.info(f"[input_2] {data}")

def control_callback(request):
    careflow.logger.info(f"[control] {request}")

careflow.start(
    callback=[callback_input1, callback_input2],  # mapped by SUB_TOPIC order
    control_callback=control_callback
)

Get data from the Data Broker without subscribing

import careflow

data = careflow.get()

careflow.get()

Fetches data directly from the subscribed topic(s) on demand, without a webhook or server. Iterates every topic in SUB_TOPIC and returns a combined list of records.

careflow.flush()

Flushes every input topic’s index to the latest offset on Data Broker for the deployment. Does not return any data from the input topic.

Example

import careflow

# Skip any backlog and advance each subscribed topic to the latest offset.
careflow.flush()

# Subsequent reads only see data published after the flush.
data = careflow.get()

S3

S3-compatible object storage (AWS S3, MinIO, Wasabi, etc.) is available via careflow.s3 when storage is configured.

Initialization

• The careflow.s3 object is automatically initialized during the container startup (S3Service.launch())
• Configuration is read from the storage.blob section of config.yaml (see Section 2)
• Credentials/endpoint can be supplied per blob via ENDPOINT, ACCESS_ID, ACCESS_KEY, REGION_NAME, or sourced from the standard AWS credential chain (environment/instance role).

Single-blob Mode

If exactly one entry is defined under storage.blob, careflow.s3 acts as a direct client:

Multi-blob Mode

If multiple entries are defined under storage.blob, select a specific store by its LABEL:

output_s3 = careflow.s3.switch("careflow-data-output")
# or dictionary-style:
temp_s3 = careflow.s3["careflow-data-temporary"]

output_s3.save("local/results.csv", "export/result.csv")
data = temp_s3.get("cache/latest.json")

careflow.s3.labels() # list available labels

If an ENDPOINT is configured (e.g. MinIO), the URI returned by save()/delete() includes the endpoint (http://localhost:9000/bucket/path/file.json); otherwise it uses the standard s3://bucket/key format.