Skip to content

JSONL Module API

The fuzzy_infer.jsonl module provides JSONL (JSON Lines) serialization for facts and rules, enabling Unix-style streaming and composability.

Module Reference

fuzzy_infer.jsonl

JSONL-based serialization for FuzzyInfer.

Pure JSONL format - one JSON object per line. No backwards compatibility with old JSON/YAML formats.

JSONLStream 

JSONLStream(input_stream: Optional[TextIO] = None)

Stream processor for JSONL data.

Initialize with optional input stream.

Source code in fuzzy_infer/jsonl.py 
130
131
132
def __init__(self, input_stream: Optional[TextIO] = None):
    """Initialize with optional input stream."""
    self.input_stream = input_stream or sys.stdin

read 

read() -> Iterator[Dict[str, Any]]

Read JSONL items from stream.

Source code in fuzzy_infer/jsonl.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def read(self) -> Iterator[Dict[str, Any]]:
    """Read JSONL items from stream."""
    for line_num, line in enumerate(self.input_stream, 1):
        line = line.strip()

        # Skip empty lines and comments
        if not line or line.startswith('#'):
            continue

        try:
            yield json.loads(line)
        except json.JSONDecodeError as e:
            logger.error(f"Line {line_num}: Invalid JSON - {e}")
            continue

write 

write(
    items: Iterator[Dict[str, Any]],
    output: Optional[TextIO] = None,
) -> None

Write items as JSONL.

Source code in fuzzy_infer/jsonl.py 
149
150
151
152
153
154
155
156
157
def write(self, items: Iterator[Dict[str, Any]],
         output: Optional[TextIO] = None) -> None:
    """Write items as JSONL."""
    output = output or sys.stdout

    for item in items:
        json.dump(item, output, separators=(',', ':'))
        output.write('\n')
        output.flush()

parse 

parse() -> Iterator[Union[Fact, Rule, Dict]]

Parse stream into Fact/Rule objects.

Source code in fuzzy_infer/jsonl.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
def parse(self) -> Iterator[Union[Fact, Rule, Dict]]:
    """Parse stream into Fact/Rule objects."""
    for item in self.read():
        item_type = item.get('type', 'fact')

        try:
            if item_type == 'fact':
                # Filter out extra fields
                fact_fields = {'type', 'pred', 'args', 'deg', '_inferred'}
                filtered = {k: v for k, v in item.items() if k in fact_fields}
                yield FactItem(**filtered).to_fact()
            elif item_type == 'rule':
                yield RuleItem(**item).to_rule()
            else:
                # Pass through unknown types
                yield item
        except Exception as e:
            logger.error(f"Failed to parse {item_type}: {e}")
            continue

KnowledgeBaseIO

I/O operations for knowledge bases in JSONL format.

save staticmethod 

save(kb: FuzzyInfer, path: Union[str, Path]) -> None

Save knowledge base to JSONL file.

Source code in fuzzy_infer/jsonl.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
@staticmethod
def save(kb: 'FuzzyInfer', path: Union[str, Path]) -> None:
    """Save knowledge base to JSONL file."""
    path = Path(path)

    with open(path, 'w') as f:
        # Write facts
        for fact in kb.get_facts():
            item = FactItem.from_fact(fact)
            f.write(item.to_jsonl() + '\n')

        # Write rules
        for rule in kb.get_rules():
            item = RuleItem.from_rule(rule)
            f.write(item.to_jsonl() + '\n')

load staticmethod 

load(path: Union[str, Path]) -> FuzzyInfer

Load knowledge base from JSONL file.

Source code in fuzzy_infer/jsonl.py 
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
@staticmethod
def load(path: Union[str, Path]) -> 'FuzzyInfer':
    """Load knowledge base from JSONL file."""
    from fuzzy_infer.core import FuzzyInfer

    path = Path(path)
    kb = FuzzyInfer()

    with open(path) as f:
        stream = JSONLStream(f)
        for obj in stream.parse():
            if isinstance(obj, Fact):
                kb.add_fact(obj)
            elif isinstance(obj, Rule):
                kb.add_rule(obj)

    return kb

append staticmethod 

append(kb: FuzzyInfer, path: Union[str, Path]) -> None

Append knowledge base to existing JSONL file.

Source code in fuzzy_infer/jsonl.py 
217
218
219
220
221
222
223
224
225
226
227
228
229
@staticmethod
def append(kb: 'FuzzyInfer', path: Union[str, Path]) -> None:
    """Append knowledge base to existing JSONL file."""
    path = Path(path)

    with open(path, 'a') as f:
        for fact in kb.get_facts():
            item = FactItem.from_fact(fact)
            f.write(item.to_jsonl() + '\n')

        for rule in kb.get_rules():
            item = RuleItem.from_rule(rule)
            f.write(item.to_jsonl() + '\n')

FactItem dataclass 

FactItem(
    type: str = "fact",
    pred: str = "",
    args: list = None,
    deg: float = 1.0,
    _inferred: Optional[bool] = None,
)

Bases: JSONLItem

JSONL representation of a fact.

from_fact classmethod 

from_fact(fact: Fact) -> FactItem

Create from Fact object.

Source code in fuzzy_infer/jsonl.py 
45
46
47
48
49
50
51
52
@classmethod
def from_fact(cls, fact: Fact) -> 'FactItem':
    """Create from Fact object."""
    return cls(
        pred=fact.predicate,
        args=list(fact.args),
        deg=fact.degree
    )

to_fact 

to_fact() -> Fact

Convert to Fact object.

Source code in fuzzy_infer/jsonl.py 
54
55
56
def to_fact(self) -> Fact:
    """Convert to Fact object."""
    return Fact(self.pred, self.args, self.deg)

RuleItem dataclass 

RuleItem(
    type: str = "rule",
    name: Optional[str] = None,
    cond: list = None,
    actions: list = None,
    priority: int = 50,
)

Bases: JSONLItem

JSONL representation of a rule.

from_rule classmethod 

from_rule(rule: Rule) -> RuleItem

Create from Rule object.

Source code in fuzzy_infer/jsonl.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
@classmethod
def from_rule(cls, rule: Rule) -> 'RuleItem':
    """Create from Rule object."""
    # Convert Condition and Action objects to dicts
    conditions = []
    for cond in rule.conditions:
        if hasattr(cond, 'to_dict'):
            conditions.append(cond.to_dict())
        else:
            conditions.append(cond)

    actions = []
    for action in rule.actions:
        if hasattr(action, 'to_dict'):
            actions.append(action.to_dict())
        else:
            actions.append(action)

    return cls(
        name=rule.name,
        cond=conditions,
        actions=actions,
        priority=rule.priority or 50
    )

to_rule 

to_rule() -> Rule

Convert to Rule object.

Source code in fuzzy_infer/jsonl.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def to_rule(self) -> Rule:
    """Convert to Rule object."""
    from fuzzy_infer.models import Condition, Action

    # Convert conditions to Condition objects
    conditions = []
    for cond in self.cond:
        if isinstance(cond, dict):
            conditions.append(Condition.from_dict(cond))
        else:
            conditions.append(cond)

    # Convert actions to Action objects
    actions = []
    for action in self.actions:
        if isinstance(action, dict):
            actions.append(Action.from_dict(action))
        else:
            actions.append(action)

    return Rule(
        name=self.name,
        conditions=conditions,
        actions=actions,
        priority=self.priority
    )

Quick Start

from fuzzy_infer.jsonl import JSONLStream, KnowledgeBaseIO
from fuzzy_infer import FuzzyInfer, Fact, Rule
import sys

# Save a knowledge base
kb = FuzzyInfer()
kb.add_fact(Fact("is-bird", ["robin"], 0.9))
KnowledgeBaseIO.save(kb, "kb.jsonl")

# Load a knowledge base
kb2 = KnowledgeBaseIO.load("kb.jsonl")

# Stream processing
stream = JSONLStream(sys.stdin)
for obj in stream.parse():
    if isinstance(obj, Fact):
        print(f"Fact: {obj}")
    elif isinstance(obj, Rule):
        print(f"Rule: {obj}")

Classes

JSONLStream

Stream processor for JSONL data.

Initialize with optional input stream.

Source code in fuzzy_infer/jsonl.py 
130
131
132
def __init__(self, input_stream: Optional[TextIO] = None):
    """Initialize with optional input stream."""
    self.input_stream = input_stream or sys.stdin

input_stream instance-attribute 

input_stream = input_stream or stdin

read 

read() -> Iterator[Dict[str, Any]]

Read JSONL items from stream.

Source code in fuzzy_infer/jsonl.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def read(self) -> Iterator[Dict[str, Any]]:
    """Read JSONL items from stream."""
    for line_num, line in enumerate(self.input_stream, 1):
        line = line.strip()

        # Skip empty lines and comments
        if not line or line.startswith('#'):
            continue

        try:
            yield json.loads(line)
        except json.JSONDecodeError as e:
            logger.error(f"Line {line_num}: Invalid JSON - {e}")
            continue

write 

write(
    items: Iterator[Dict[str, Any]],
    output: Optional[TextIO] = None,
) -> None

Write items as JSONL.

Source code in fuzzy_infer/jsonl.py 
149
150
151
152
153
154
155
156
157
def write(self, items: Iterator[Dict[str, Any]],
         output: Optional[TextIO] = None) -> None:
    """Write items as JSONL."""
    output = output or sys.stdout

    for item in items:
        json.dump(item, output, separators=(',', ':'))
        output.write('\n')
        output.flush()

parse 

parse() -> Iterator[Union[Fact, Rule, Dict]]

Parse stream into Fact/Rule objects.

Source code in fuzzy_infer/jsonl.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
def parse(self) -> Iterator[Union[Fact, Rule, Dict]]:
    """Parse stream into Fact/Rule objects."""
    for item in self.read():
        item_type = item.get('type', 'fact')

        try:
            if item_type == 'fact':
                # Filter out extra fields
                fact_fields = {'type', 'pred', 'args', 'deg', '_inferred'}
                filtered = {k: v for k, v in item.items() if k in fact_fields}
                yield FactItem(**filtered).to_fact()
            elif item_type == 'rule':
                yield RuleItem(**item).to_rule()
            else:
                # Pass through unknown types
                yield item
        except Exception as e:
            logger.error(f"Failed to parse {item_type}: {e}")
            continue

KnowledgeBaseIO

I/O operations for knowledge bases in JSONL format.

save staticmethod 

save(kb: FuzzyInfer, path: Union[str, Path]) -> None

Save knowledge base to JSONL file.

Source code in fuzzy_infer/jsonl.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
@staticmethod
def save(kb: 'FuzzyInfer', path: Union[str, Path]) -> None:
    """Save knowledge base to JSONL file."""
    path = Path(path)

    with open(path, 'w') as f:
        # Write facts
        for fact in kb.get_facts():
            item = FactItem.from_fact(fact)
            f.write(item.to_jsonl() + '\n')

        # Write rules
        for rule in kb.get_rules():
            item = RuleItem.from_rule(rule)
            f.write(item.to_jsonl() + '\n')

load staticmethod 

load(path: Union[str, Path]) -> FuzzyInfer

Load knowledge base from JSONL file.

Source code in fuzzy_infer/jsonl.py 
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
@staticmethod
def load(path: Union[str, Path]) -> 'FuzzyInfer':
    """Load knowledge base from JSONL file."""
    from fuzzy_infer.core import FuzzyInfer

    path = Path(path)
    kb = FuzzyInfer()

    with open(path) as f:
        stream = JSONLStream(f)
        for obj in stream.parse():
            if isinstance(obj, Fact):
                kb.add_fact(obj)
            elif isinstance(obj, Rule):
                kb.add_rule(obj)

    return kb

append staticmethod 

append(kb: FuzzyInfer, path: Union[str, Path]) -> None

Append knowledge base to existing JSONL file.

Source code in fuzzy_infer/jsonl.py 
217
218
219
220
221
222
223
224
225
226
227
228
229
@staticmethod
def append(kb: 'FuzzyInfer', path: Union[str, Path]) -> None:
    """Append knowledge base to existing JSONL file."""
    path = Path(path)

    with open(path, 'a') as f:
        for fact in kb.get_facts():
            item = FactItem.from_fact(fact)
            f.write(item.to_jsonl() + '\n')

        for rule in kb.get_rules():
            item = RuleItem.from_rule(rule)
            f.write(item.to_jsonl() + '\n')

FactItem

Bases: JSONLItem

JSONL representation of a fact.

type class-attribute instance-attribute 

type: str = 'fact'

pred class-attribute instance-attribute 

pred: str = ''

args class-attribute instance-attribute 

args: list = None

deg class-attribute instance-attribute 

deg: float = 1.0

_inferred class-attribute instance-attribute 

_inferred: Optional[bool] = None

__post_init__ 

__post_init__()
Source code in fuzzy_infer/jsonl.py 
40
41
42
43
def __post_init__(self):
    if self.args is None:
        self.args = []
    self.deg = max(0.0, min(1.0, self.deg))

from_fact classmethod 

from_fact(fact: Fact) -> FactItem

Create from Fact object.

Source code in fuzzy_infer/jsonl.py 
45
46
47
48
49
50
51
52
@classmethod
def from_fact(cls, fact: Fact) -> 'FactItem':
    """Create from Fact object."""
    return cls(
        pred=fact.predicate,
        args=list(fact.args),
        deg=fact.degree
    )

to_fact 

to_fact() -> Fact

Convert to Fact object.

Source code in fuzzy_infer/jsonl.py 
54
55
56
def to_fact(self) -> Fact:
    """Convert to Fact object."""
    return Fact(self.pred, self.args, self.deg)

RuleItem

Bases: JSONLItem

JSONL representation of a rule.

type class-attribute instance-attribute 

type: str = 'rule'

name class-attribute instance-attribute 

name: Optional[str] = None

cond class-attribute instance-attribute 

cond: list = None

actions class-attribute instance-attribute 

actions: list = None

priority class-attribute instance-attribute 

priority: int = 50

__post_init__ 

__post_init__()
Source code in fuzzy_infer/jsonl.py 
68
69
70
71
72
def __post_init__(self):
    if self.cond is None:
        self.cond = []
    if self.actions is None:
        self.actions = []

from_rule classmethod 

from_rule(rule: Rule) -> RuleItem

Create from Rule object.

Source code in fuzzy_infer/jsonl.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
@classmethod
def from_rule(cls, rule: Rule) -> 'RuleItem':
    """Create from Rule object."""
    # Convert Condition and Action objects to dicts
    conditions = []
    for cond in rule.conditions:
        if hasattr(cond, 'to_dict'):
            conditions.append(cond.to_dict())
        else:
            conditions.append(cond)

    actions = []
    for action in rule.actions:
        if hasattr(action, 'to_dict'):
            actions.append(action.to_dict())
        else:
            actions.append(action)

    return cls(
        name=rule.name,
        cond=conditions,
        actions=actions,
        priority=rule.priority or 50
    )

to_rule 

to_rule() -> Rule

Convert to Rule object.

Source code in fuzzy_infer/jsonl.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def to_rule(self) -> Rule:
    """Convert to Rule object."""
    from fuzzy_infer.models import Condition, Action

    # Convert conditions to Condition objects
    conditions = []
    for cond in self.cond:
        if isinstance(cond, dict):
            conditions.append(Condition.from_dict(cond))
        else:
            conditions.append(cond)

    # Convert actions to Action objects
    actions = []
    for action in self.actions:
        if isinstance(action, dict):
            actions.append(Action.from_dict(action))
        else:
            actions.append(action)

    return Rule(
        name=self.name,
        conditions=conditions,
        actions=actions,
        priority=self.priority
    )