v0.3.1 greynet API refactor

Author: CameleoGrey

examples/greynet/greynet_example_aggregations.py

@@ -16,7 +16,7 @@ class SalesTransaction:
 @builder.constraint("Sales Regional Analysis")
 def sales_analysis():
     return (
-        builder.from_facts(SalesTransaction)
+        builder.for_each(SalesTransaction)
             .group_by(
                 lambda tx: tx.region,
                 Collectors.compose({

examples/greynet/greynet_example_all_collectors.py

@@ -51,7 +51,7 @@ def count_collector_example():
     """Demonstrates: CountCollector
     Counts the number of sales transactions for each product. Penalizes if a product has more than 3 sales.
     """
-    return (cb.from_facts(Sale)
+    return (cb.for_each(Sale)
             .group_by(lambda s: s.product_id, Collectors.count())
             .filter(lambda product_id, count: count > 3)
             .penalize_simple(lambda product_id, count: count)
@@ -62,7 +62,7 @@ def sum_collector_example():
     """Demonstrates: SumCollector
     Calculates the total revenue (price * quantity) for each product.
     """
-    return (cb.from_facts(Sale)
+    return (cb.for_each(Sale)
             .group_by(lambda s: s.product_id, Collectors.sum(lambda s: s.price * s.quantity))
             .filter(lambda product_id, total_revenue: total_revenue > 0)
             .penalize_simple(lambda product_id, total_revenue: 0) # Use penalty 0 to just report
@@ -73,7 +73,7 @@ def min_max_avg_collectors_example():
     """Demonstrates: MinCollector, MaxCollector, AvgCollector
     Finds the minimum, maximum, and average sale price for each product.
     """
-    return (cb.from_facts(Sale)
+    return (cb.for_each(Sale)
             .group_by(lambda s: s.product_id, Collectors.compose({
                 "min_price": Collectors.min(lambda s: s.price),
                 "max_price": Collectors.max(lambda s: s.price),
@@ -88,7 +88,7 @@ def stddev_variance_collectors_example():
     """Demonstrates: StdDevCollector, VarianceCollector
     Calculates the standard deviation and variance of prices for each product.
     """
-    return (cb.from_facts(Sale)
+    return (cb.for_each(Sale)
             .group_by(lambda s: s.product_id, Collectors.compose({
                 "price_stddev": Collectors.stddev(lambda s: s.price),
                 "price_variance": Collectors.variance(lambda s: s.price)
@@ -102,7 +102,7 @@ def list_collector_example():
     """Demonstrates: ListCollector
     Collects all `Sale` objects for each product into a list.
     """
-    return (cb.from_facts(Sale)
+    return (cb.for_each(Sale)
             .group_by(lambda s: s.product_id, Collectors.to_list())
             .filter(lambda product_id, sales_list: len(sales_list) > 0)
             .penalize_simple(lambda product_id, sales_list: 0) # Reporting only
@@ -113,7 +113,7 @@ def set_collector_example():
     """Demonstrates: SetCollector and MappingCollector
     Collects the unique set of customer IDs for each product.
     """
-    return (cb.from_facts(Sale)
+    return (cb.for_each(Sale)
             .group_by(
                 lambda s: s.product_id,
                 Collectors.mapping(
@@ -130,7 +130,7 @@ def distinct_collector_example():
     """Demonstrates: DistinctCollector
     Collects a list of unique customer IDs for each product, preserving insertion order.
     """
-    return (cb.from_facts(Sale)
+    return (cb.for_each(Sale)
             .group_by(
                 lambda s: s.product_id,
                 Collectors.mapping(
@@ -147,7 +147,7 @@ def filtering_collector_example():
     """Demonstrates: FilteringCollector
     Counts only the sales where the quantity is greater than 2.
     """
-    return (cb.from_facts(Sale)
+    return (cb.for_each(Sale)
             .group_by(
                 lambda s: s.product_id,
                 Collectors.filtering(
@@ -164,7 +164,7 @@ def consecutive_sequences_collector_example():
     """Demonstrates: consecutive_sequences
     Finds consecutive sequences of shipment numbers for each order.
     """
-    return (cb.from_facts(Shipment)
+    return (cb.for_each(Shipment)
             .group_by(
                 lambda s: s.order_id,
                 Collectors.consecutive_sequences(lambda s: s.shipment_no)
@@ -178,7 +178,7 @@ def connected_ranges_collector_example():
     """Demonstrates: connected_ranges
     Finds groups of overlapping or adjacent maintenance windows for each machine.
     """
-    return (cb.from_facts(Maintenance)
+    return (cb.for_each(Maintenance)
             .group_by(
                 lambda m: m.machine_id,
                 Collectors.connected_ranges(
@@ -205,7 +205,7 @@ def get_window_key(timestamp: datetime) -> datetime:
         window_start_ts = epoch.timestamp() + window_index * window_size_sec
         return datetime.fromtimestamp(window_start_ts, tz=timezone.utc)
 
-    return (cb.from_facts(UserEvent)
+    return (cb.for_each(UserEvent)
             .group_by(
                 group_key_function=lambda e: get_window_key(e.timestamp),
                 collector_supplier=Collectors.avg(lambda e: e.value)

examples/greynet/greynet_example_blom_filter.py

@@ -20,11 +20,11 @@ class UserBan:
 @builder.constraint("login_not_banned")
 def login_not_banned():
     # Start from UserLogin facts
-    login_stream = builder.from_facts(UserLogin)
+    login_stream = builder.for_each(UserLogin)
     # Join with UserBan facts, using NOT_EQUAL on user_id
     # This will use the CountingBloomFilter under the hood
     not_banned_stream = login_stream.join(
-        builder.from_facts(UserBan),
+        builder.for_each(UserBan),
         JoinerType.NOT_EQUAL,
         left_key_func=lambda login: login.user_id,
         right_key_func=lambda ban: ban.user_id

examples/greynet/greynet_example_composite_keys.py

@@ -56,7 +56,7 @@ def room_double_booking():
     at the exact same time. This is a classic composite key equality join.
     The composite key is (room_id, start_time).
     """
-    appointments = builder.from_facts(Appointment)
+    appointments = builder.for_each(Appointment)
     return (
         appointments.join(
             appointments,
@@ -78,10 +78,10 @@ def resource_allocation_mismatch():
     This demonstrates a multi-stage join process, followed by filtering.
     """
     # Create streams for all our base facts
-    appointments = builder.from_facts(Appointment)
-    doctors = builder.from_facts(Doctor)
-    rooms = builder.from_facts(Room)
-    procedures = builder.from_facts(ProcedureInfo)
+    appointments = builder.for_each(Appointment)
+    doctors = builder.for_each(Doctor)
+    rooms = builder.for_each(Room)
+    procedures = builder.for_each(ProcedureInfo)
 
     # Stage 1: Join Appointments with Doctors
     app_with_doc = appointments.join(
@@ -125,7 +125,7 @@ def scheduling_priority_inversion():
     case on the same day. This is achieved by first grouping appointments by
     doctor and day, and then filtering for priority/time inversions.
     """
-    appointments = builder.from_facts(Appointment)
+    appointments = builder.for_each(Appointment)
 
     # Define the composite key for grouping: (doctor_id, date)
     def get_grouping_key(app: Appointment) -> tuple:

examples/greynet/greynet_example_core.py

@@ -42,9 +42,9 @@ class CompanyPolicy:
 @builder.constraint("Required skill missing")
 def required_skill_missing():
     return (
-        builder.from_facts(Shift)
+        builder.for_each(Shift)
             .join(
-                builder.from_facts(Employee),
+                builder.for_each(Employee),
                 JoinerType.EQUAL,
                 left_key_func=lambda shift: shift.employee_name,
                 right_key_func=lambda employee: employee.name
@@ -57,7 +57,7 @@ def required_skill_missing():
 @builder.constraint("Shift for non-existent employee")
 def shift_for_non_existent_employee():
     return (
-        builder.from_facts(Shift)
+        builder.for_each(Shift)
             .if_not_exists(
                 Employee,
                 # FIX: Removed JoinerType.EQUAL, which was an invalid argument.
@@ -71,9 +71,9 @@ def shift_for_non_existent_employee():
 @builder.constraint("Scheduled on unavailable day")
 def scheduled_on_unavailable_day():
     return (
-        builder.from_facts(Shift)
+        builder.for_each(Shift)
             .join(
-                builder.from_facts(Employee),
+                builder.for_each(Employee),
                 JoinerType.EQUAL,
                 left_key_func=lambda shift: shift.employee_name,
                 right_key_func=lambda employee: employee.name
@@ -86,9 +86,9 @@ def scheduled_on_unavailable_day():
 @builder.constraint("Overlapping shifts")
 def overlapping_shifts():
     return (
-        builder.from_facts(Shift)
+        builder.for_each(Shift)
             .join(
-                builder.from_facts(Shift),
+                builder.for_each(Shift),
                 JoinerType.EQUAL,
                 left_key_func=lambda s: (s.employee_name, s.shift_date),
                 right_key_func=lambda s: (s.employee_name, s.shift_date)
@@ -108,7 +108,7 @@ def max_consecutive_days():
     )
 
     return (
-        builder.from_facts(Shift)
+        builder.for_each(Shift)
             .group_by(
                 lambda shift: shift.employee_name,
                 Collectors.consecutive_sequences(
@@ -117,7 +117,7 @@ def max_consecutive_days():
                 )
             )
             .join(
-                 builder.from_facts(CompanyPolicy),
+                 builder.for_each(CompanyPolicy),
                  JoinerType.GREATER_THAN, # A dummy join to bring the policy into the stream
                  left_key_func=lambda name, sequences: 1,
                  right_key_func=lambda policy: 0

examples/greynet/greynet_example_node_sharing.py

@@ -56,7 +56,7 @@ def create_rules(builder: ConstraintBuilder, num_rules: int, with_sharing: bool)
             # All rules share the exact same filter condition.
             @builder.constraint(constraint_id, default_weight=1.0)
             def shared_rule():
-                return (builder.from_facts(Transaction)
+                return (builder.for_each(Transaction)
                                .filter(lambda t: t.amount > 5000)
                                .penalize_simple(lambda t: t.amount))
         else:
@@ -68,7 +68,7 @@ def create_unique_rule(rule_index):
 
                 @builder.constraint(f"high_value_tx_unique_{rule_index}", default_weight=1.0)
                 def unique_rule():
-                    return (builder.from_facts(Transaction)
+                    return (builder.for_each(Transaction)
                                    .filter(unique_filter) # Use the unique filter object
                                    .penalize_simple(lambda t: t.amount))
                 return unique_rule

examples/greynet/greynet_example_stress_1.py

@@ -38,23 +38,23 @@ def define_constraints(builder: ConstraintBuilder):
     # Constraint 1: Simple filter for high-value transactions.
     @builder.constraint("high_value_transaction")
     def high_value_transaction():
-        return builder.from_facts(Transaction)\
+        return builder.for_each(Transaction)\
             .filter(lambda tx: tx.amount > 45000)\
             .penalize_simple(lambda tx: tx.amount / 1000)
 
     # Constraint 2: Group transactions by customer and check for excessive activity.
     @builder.constraint("excessive_transactions_per_customer")
     def excessive_transactions():
-        return builder.from_facts(Transaction)\
+        return builder.for_each(Transaction)\
             .group_by(lambda tx: tx.customer_id, Collectors.count())\
             .filter(lambda cid, count: count > 25)\
             .penalize_simple(lambda cid, count: (count - 25) * 10)
 
     # Constraint 3: Join transactions with security alerts on location.
     @builder.constraint("transaction_in_alerted_location")
     def suspicious_location_tx():
-        return builder.from_facts(Transaction)\
-            .join(builder.from_facts(SecurityAlert), 
+        return builder.for_each(Transaction)\
+            .join(builder.for_each(SecurityAlert), 
                   JoinerType.EQUAL, 
                   lambda tx: tx.location, 
                   lambda alert: alert.location)\
@@ -63,9 +63,9 @@ def suspicious_location_tx():
     # Constraint 4: Join to find transactions from inactive customers.
     @builder.constraint("inactive_customer_transaction")
     def inactive_customer_tx():
-        return builder.from_facts(Customer)\
+        return builder.for_each(Customer)\
             .filter(lambda c: c.status == 'inactive')\
-            .join(builder.from_facts(Transaction), 
+            .join(builder.for_each(Transaction), 
                   JoinerType.EQUAL, 
                   lambda c: c.id, 
                   lambda tx: tx.customer_id)\
@@ -76,9 +76,9 @@ def inactive_customer_tx():
     # that does NOT have a security alert.
     @builder.constraint("high_risk_transaction_without_alert")
     def high_risk_no_alert():
-        return builder.from_facts(Customer)\
+        return builder.for_each(Customer)\
             .filter(lambda c: c.risk_level == 'high')\
-            .join(builder.from_facts(Transaction), 
+            .join(builder.for_each(Transaction), 
                   JoinerType.EQUAL, 
                   lambda c: c.id, 
                   lambda tx: tx.customer_id)\

examples/greynet/greynet_example_stress_2.py

@@ -114,7 +114,7 @@ def define_advanced_constraints(builder: ConstraintBuilder):
     # --- Rule 1: Market Manipulation ("Painting the Tape") ---
     @builder.constraint("market_manipulation_painting_the_tape")
     def painting_the_tape():
-        return builder.from_facts(Trade) \
+        return builder.for_each(Trade) \
             .group_by(
                 lambda trade: (trade.trader_id, trade.ticker),
                 Collectors.to_list()
@@ -128,7 +128,7 @@ def painting_the_tape():
     # --- Rule 2: Exceeding Daily Risk Limits ---
     @builder.constraint("trader_exceeds_daily_risk_limit")
     def trader_risk_limit():
-        daily_trade_volumes = builder.from_facts(Trade) \
+        daily_trade_volumes = builder.for_each(Trade) \
             .group_by(
                 lambda trade: (trade.trader_id, trade.timestamp.date()),
                 Collectors.compose({
@@ -141,7 +141,7 @@ def trader_risk_limit():
         # and join the Trader facts to it. This uses a different node
         # that preserves all necessary data.
         return daily_trade_volumes \
-            .join(builder.from_facts(Trader),
+            .join(builder.for_each(Trader),
                   JoinerType.EQUAL,
                   # Left key func: extracts trader_id from the group key
                   lambda key, result: key[0],
@@ -161,15 +161,15 @@ def trader_risk_limit():
     # --- Rule 3: Potential Insider Trading ---
     @builder.constraint("potential_insider_trading")
     def insider_trading():
-        insider_trades = builder.from_facts(Trader)\
+        insider_trades = builder.for_each(Trader)\
             .filter(lambda trader: trader.is_insider)\
-            .join(builder.from_facts(Trade),
+            .join(builder.for_each(Trade),
                 JoinerType.EQUAL,
                 lambda trader: trader.id,
                 lambda trade: trade.trader_id
             )
 
-        return insider_trades.join(builder.from_facts(MarketNews),
+        return insider_trades.join(builder.for_each(MarketNews),
                 JoinerType.EQUAL,
                 lambda trader, trade: trade.ticker,
                 lambda news: news.related_tickers[0] if news.related_tickers else None
@@ -182,9 +182,9 @@ def insider_trading():
     # --- Rule 4: Restricted Security Trading Bursts ---
     @builder.constraint("restricted_security_trading_burst")
     def restricted_trading_burst():
-        restricted_trades = builder.from_facts(Trade) \
+        restricted_trades = builder.for_each(Trade) \
             .if_exists(
-                builder.from_facts(Security).filter(lambda s: s.is_restricted),
+                builder.for_each(Security).filter(lambda s: s.is_restricted),
                 lambda trade: trade.ticker,
                 lambda sec: sec.ticker
             )

examples/greynet/greynet_example_subqueries.py

@@ -57,16 +57,16 @@ class TrainingCompletion:
 # ---
 @builder.constraint("Missing Invoice for VIP Project")
 def missing_invoice_for_vip_project():
-    active_assignments = builder.from_facts(Assignment).filter(lambda a: a.status == 'Active')
+    active_assignments = builder.for_each(Assignment).filter(lambda a: a.status == 'Active')
 
     vip_projects = (
         active_assignments.join(
-            builder.from_facts(Project),
+            builder.for_each(Project),
             JoinerType.EQUAL,
             left_key_func=lambda a: a.project_id,
             right_key_func=lambda p: p.project_id
         ).join(
-            builder.from_facts(Client),
+            builder.for_each(Client),
             JoinerType.EQUAL,
             left_key_func=lambda a, p: p.client_id,
             right_key_func=lambda c: c.client_id
@@ -86,13 +86,13 @@ def missing_invoice_for_vip_project():
 # ---
 @builder.constraint("Training Dead-End Assignment")
 def training_dead_end():
-    assignments = builder.from_facts(Assignment).join(
-        builder.from_facts(Employee),
+    assignments = builder.for_each(Assignment).join(
+        builder.for_each(Employee),
         JoinerType.EQUAL,
         lambda a: a.employee_id,
         lambda e: e.employee_id
     ).join(
-        builder.from_facts(Project),
+        builder.for_each(Project),
         JoinerType.EQUAL,
         lambda a, e: a.project_id,
         lambda p: p.project_id
@@ -117,10 +117,10 @@ def training_dead_end():
 @builder.constraint("Bonus Flag: VIP Project Completion by Trained Employee")
 def bonus_for_vip_completion():
     completed_vip_assignments = (
-        builder.from_facts(Assignment)
+        builder.for_each(Assignment)
             .filter(lambda a: a.status == 'Completed')
-            .join(builder.from_facts(Project), JoinerType.EQUAL, lambda a: a.project_id, lambda p: p.project_id)
-            .join(builder.from_facts(Client), JoinerType.EQUAL, lambda a, p: p.client_id, lambda c: c.client_id)
+            .join(builder.for_each(Project), JoinerType.EQUAL, lambda a: a.project_id, lambda p: p.project_id)
+            .join(builder.for_each(Client), JoinerType.EQUAL, lambda a, p: p.client_id, lambda c: c.client_id)
             .filter(lambda a, p, c: c.status == 'VIP')
     )