“

Series: S12 Developer Tutorials · Article 5 | Level: Beginner | Reading Time: 15 min

Your First Rule: Creating Automation with YAML

#Introduction

In coomia-dip, Rules are the core mechanism for automation and business logic. By declaratively defining rules in YAML, you can make the platform automatically respond to data changes, enforce business policies, and trigger workflows. This tutorial will guide you through creating your first rule and understanding how the coomia-dip reasoning engine works.

#Core Concepts

#What is a Rule?

A Rule is a "condition-action" pair: when a condition is met, the corresponding action is automatically executed. The coomia-dip rule engine runs on the Intelligence Layer (Reasoning & Decision Layer), built with Python + FastAPI, communicating with other Layers via gRPC.

#Rule Types

coomia-dip supports three rule types:

Event-Driven: Triggered by data changes
Scheduled: Executed on time-based schedules
Inference: Based on knowledge graph reasoning

#Creating Event-Driven Rules

#Scenario: Critical Project Auto-Notification

YAML

# rules/critical_project_alert.yaml
name: critical_project_alert
display_name: Critical Project Alert
description: Auto-notify when project priority becomes critical

trigger:
  type: data_change
  object_type: Project
  events: [update]
  watch_properties: [priority]

condition:
  all:
    - property: priority
      operator: eq
      value: critical
    - property: status
      operator: in
      values: [planning, in_progress]

actions:
  - type: execute_action
    action_name: send_notification
    parameters:
      channel: "slack"
      template: "critical_project_alert"
      recipients:
        query:
          relation: owns_project
          direction: source
          target: $trigger.object_rid

  - type: update_object
    object_type: Project
    target: $trigger.object_rid
    properties:
      tags:
        append: "needs-attention"

metadata:
  priority: high
  enabled: true

#Loading and Activating Rules

Python

from ontology_sdk import OntoPlatform

platform = OntoPlatform(
    control_plane_url="localhost:50051",
    intelligence_plane_url="localhost:50053"
)

rule = platform.rules.load_from_yaml("rules/critical_project_alert.yaml")
platform.rules.activate(rule.name)

# Test the rule
result = platform.rules.test(
    rule_name="critical_project_alert",
    test_data={
        "object_type": "Project",
        "event": "update",
        "changed_properties": {
            "priority": {"old": "high", "new": "critical"}
        }
    }
)
print(f"Would trigger: {result.would_trigger}")

#Creating Scheduled Rules

#Scenario: Daily Project Progress Check

YAML

# rules/daily_project_check.yaml
name: daily_project_check
display_name: Daily Project Progress Check
description: Check all in-progress projects every morning at 9 AM

trigger:
  type: schedule
  cron: "0 9 * * *"
  timezone: UTC

condition:
  always: true

actions:
  - type: query_and_process
    query:
      object_type: Project
      filter:
        status: {eq: in_progress}
        end_date: {lt: $today_plus_7d}
    for_each: project
    actions:
      - type: execute_action
        action_name: send_notification
        parameters:
          channel: email
          template: project_deadline_reminder
          recipients:
            query:
              relation: participates_in
              direction: source
              target: $project.rid
              filter: {role: owner}
          data:
            project_name: $project.properties.name
            days_remaining: $project.properties.end_date.diff_days($today)

metadata:
  priority: medium
  enabled: true

#Creating Inference Rules

#Scenario: Auto-Calculate Department Risk Level

YAML

# rules/department_risk_inference.yaml
name: department_risk_inference
display_name: Department Risk Inference
description: Infer department risk level based on project statuses

trigger:
  type: data_change
  object_type: Project
  events: [create, update, delete]
  watch_properties: [status, priority]

actions:
  - type: inference
    logic: |
      department = get_related(
          relation="owns_project",
          direction="source",
          target=$trigger.object_rid
      )
      projects = query(
          object_type="Project",
          relation="owns_project",
          source=department.rid
      )
      critical_count = count(projects, priority="critical")
      overdue_count = count(projects, status="in_progress", end_date < today())

      if critical_count >= 3 or overdue_count >= 2:
          risk_level = "high"
      elif critical_count >= 1 or overdue_count >= 1:
          risk_level = "medium"
      else:
          risk_level = "low"

      update(department, derived_risk_level=risk_level)

#Rule Chaining

Multiple rules can form chain reactions:

YAML

name: cascade_risk_escalation
display_name: Cascade Risk Escalation
trigger:
  type: data_change
  object_type: Department
  events: [update]
  watch_properties: [derived_risk_level]

condition:
  property: derived_risk_level
  operator: eq
  value: high

actions:
  - type: query_and_process
    query:
      relation: parent_department
      source: $trigger.object_rid
    for_each: parent_dept
    actions:
      - type: execute_action
        action_name: send_notification
        parameters:
          channel: slack
          template: risk_escalation

#Condition Expression Syntax

YAML

# Simple condition
condition:
  property: status
  operator: eq
  value: active

# AND condition
condition:
  all:
    - property: priority
      operator: eq
      value: critical
    - property: budget
      operator: gt
      value: 100000

# OR condition
condition:
  any:
    - property: status
      operator: eq
      value: on_hold
    - property: end_date
      operator: lt
      value: $today

# Nested conditions
condition:
  all:
    - any:
        - property: priority
          operator: eq
          value: critical
        - property: priority
          operator: eq
          value: high
    - property: status
      operator: neq
      value: completed

#Rule Management

Python

# List all rules
rules = platform.rules.list()
for r in rules:
    status = "Active" if r.enabled else "Disabled"
    print(f"  [{status}] {r.name}: {r.display_name}")

# View execution logs
logs = platform.rules.get_execution_logs(rule_name="critical_project_alert", limit=20)
for log in logs:
    print(f"  {log.timestamp}: {log.status}")

# Pause / Resume / Delete
platform.rules.deactivate("critical_project_alert")
platform.rules.activate("critical_project_alert")
platform.rules.delete("critical_project_alert")

#Best Practices

Single Responsibility: Each rule should do one thing
Precise Conditions: Avoid overly broad triggers
Idempotent Actions: Rule actions should be idempotent
Error Handling: Configure alerts for critical rules
Test First: Use rules.test() to validate before activation
Documentation: Write clear descriptions for every rule

#Summary

This tutorial covered the coomia-dip rule engine, including three rule types (event-driven, scheduled, inference), YAML declarative definitions, condition expression syntax, and rule management. Rules are the key capability for business automation, elevating your data platform from "passive storage" to "active intelligence."

This is Article 5 in the coomia-dip Developer Tutorial series. Repository: https://github.com/coomia-dip/coomia-dip↗ | License: Apache License 2.0