Skip to content

rule_store

Persistent storage for transform rules.

Provides JSON-file based storage for saving, loading, and searching reusable transform rules.

TransformRuleStore

Persistent storage for transform rules.

Stores rules in a JSON file for reuse across sessions and datasets. Supports searching by tags, description similarity, and column patterns.

Parameters:

Name Type Description Default
path str

Path to the JSON file for storage. Defaults to ~/.featcopilot/rules.json

 None

Examples:

>>> store = TransformRuleStore()
>>> store.save_rule(rule)
>>> matching = store.find_matching_rules(columns=["price", "quantity"])
>>> all_rules = store.list_rules()
Source code in featcopilot/stores/rule_store.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 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
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
class TransformRuleStore:
    """
    Persistent storage for transform rules.

    Stores rules in a JSON file for reuse across sessions and datasets.
    Supports searching by tags, description similarity, and column patterns.

    Parameters
    ----------
    path : str, optional
        Path to the JSON file for storage. Defaults to ~/.featcopilot/rules.json

    Examples
    --------
    >>> store = TransformRuleStore()
    >>> store.save_rule(rule)
    >>> matching = store.find_matching_rules(columns=["price", "quantity"])
    >>> all_rules = store.list_rules()
    """

    DEFAULT_PATH = "~/.featcopilot/rules.json"

    def __init__(self, path: Optional[str] = None):
        self.path = Path(os.path.expanduser(path or self.DEFAULT_PATH))
        self._rules: dict[str, TransformRule] = {}
        self._ensure_directory()
        self._load()

    def _ensure_directory(self) -> None:
        """Ensure the storage directory exists."""
        self.path.parent.mkdir(parents=True, exist_ok=True)

    def _load(self) -> None:
        """Load rules from storage file."""
        if self.path.exists():
            try:
                with open(self.path, encoding="utf-8") as f:
                    data = json.load(f)
                    self._rules = {rule_id: TransformRule.from_dict(rule_data) for rule_id, rule_data in data.items()}
                logger.debug(f"Loaded {len(self._rules)} rules from {self.path}")
            except (json.JSONDecodeError, KeyError) as e:
                logger.warning(f"Failed to load rules from {self.path}: {e}")
                self._rules = {}
        else:
            self._rules = {}

    def _save(self) -> None:
        """Save rules to storage file."""
        try:
            with open(self.path, "w", encoding="utf-8") as f:
                data = {rule_id: rule.to_dict() for rule_id, rule in self._rules.items()}
                json.dump(data, f, indent=2)
            logger.debug(f"Saved {len(self._rules)} rules to {self.path}")
        except OSError as e:
            logger.error(f"Failed to save rules to {self.path}: {e}")
            raise

    def save_rule(self, rule: TransformRule) -> str:
        """
        Save a rule to the store.

        Parameters
        ----------
        rule : TransformRule
            The rule to save

        Returns
        -------
        str
            The rule's ID
        """
        self._rules[rule.id] = rule
        self._save()
        logger.info(f"Saved rule '{rule.name}' with ID {rule.id}")
        return rule.id

    def get_rule(self, rule_id: str) -> Optional[TransformRule]:
        """
        Get a rule by ID.

        Parameters
        ----------
        rule_id : str
            The rule's ID

        Returns
        -------
        TransformRule or None
            The rule if found, None otherwise
        """
        return self._rules.get(rule_id)

    def get_rule_by_name(self, name: str) -> Optional[TransformRule]:
        """
        Get a rule by name.

        Parameters
        ----------
        name : str
            The rule's name

        Returns
        -------
        TransformRule or None
            The first rule matching the name, None if not found
        """
        for rule in self._rules.values():
            if rule.name == name:
                return rule
        return None

    def delete_rule(self, rule_id: str) -> bool:
        """
        Delete a rule by ID.

        Parameters
        ----------
        rule_id : str
            The rule's ID

        Returns
        -------
        bool
            True if deleted, False if not found
        """
        if rule_id in self._rules:
            del self._rules[rule_id]
            self._save()
            logger.info(f"Deleted rule {rule_id}")
            return True
        return False

    def list_rules(self, tags: Optional[list[str]] = None) -> list[TransformRule]:
        """
        List all rules, optionally filtered by tags.

        Parameters
        ----------
        tags : list[str], optional
            Filter rules that have all specified tags

        Returns
        -------
        list[TransformRule]
            List of matching rules
        """
        rules = list(self._rules.values())

        if tags:
            rules = [r for r in rules if all(t in r.tags for t in tags)]

        return rules

    def find_matching_rules(
        self,
        columns: Optional[list[str]] = None,
        description: Optional[str] = None,
        tags: Optional[list[str]] = None,
        min_usage: int = 0,
    ) -> list[tuple[TransformRule, dict[str, str]]]:
        """
        Find rules that can be applied to the given context.

        Parameters
        ----------
        columns : list[str], optional
            Available column names to match against
        description : str, optional
            Description to search for (keyword matching)
        tags : list[str], optional
            Required tags
        min_usage : int, default=0
            Minimum usage count

        Returns
        -------
        list[tuple[TransformRule, dict]]
            List of (rule, column_mapping) tuples for applicable rules,
            sorted by usage count (most used first)
        """
        results: list[tuple[TransformRule, dict[str, str]]] = []

        for rule in self._rules.values():
            # Filter by usage count
            if rule.usage_count < min_usage:
                continue

            # Filter by tags
            if tags and not all(t in rule.tags for t in tags):
                continue

            # Filter by description keywords
            if description:
                keywords = description.lower().split()
                rule_text = f"{rule.name} {rule.description}".lower()
                if not any(kw in rule_text for kw in keywords):
                    continue

            # Check column compatibility
            mapping: dict[str, str] = {}
            if columns:
                matches, mapping = rule.matches_columns(columns)
                if not matches:
                    continue

            results.append((rule, mapping))

        # Sort by usage count (descending)
        results.sort(key=lambda x: x[0].usage_count, reverse=True)

        return results

    def search_by_description(self, query: str, limit: int = 10) -> list[TransformRule]:
        """
        Search rules by description similarity.

        Parameters
        ----------
        query : str
            Search query
        limit : int, default=10
            Maximum number of results

        Returns
        -------
        list[TransformRule]
            Matching rules sorted by relevance
        """
        query_words = set(query.lower().split())
        scored_rules: list[tuple[float, TransformRule]] = []

        for rule in self._rules.values():
            rule_words = set(f"{rule.name} {rule.description}".lower().split())

            # Simple word overlap scoring
            overlap = len(query_words & rule_words)
            if overlap > 0:
                score = overlap / len(query_words)
                scored_rules.append((score, rule))

        # Sort by score descending
        scored_rules.sort(key=lambda x: x[0], reverse=True)

        return [rule for _, rule in scored_rules[:limit]]

    def import_rules(self, path: str, merge: bool = True) -> int:
        """
        Import rules from another JSON file.

        Parameters
        ----------
        path : str
            Path to import from
        merge : bool, default=True
            If True, merge with existing rules. If False, replace all.

        Returns
        -------
        int
            Number of rules imported
        """
        import_path = Path(os.path.expanduser(path))

        if not import_path.exists():
            raise FileNotFoundError(f"Import file not found: {path}")

        with open(import_path, encoding="utf-8") as f:
            data = json.load(f)

        if not merge:
            self._rules = {}

        count = 0
        for _rule_id, rule_data in data.items():
            rule = TransformRule.from_dict(rule_data)
            self._rules[rule.id] = rule
            count += 1

        self._save()
        logger.info(f"Imported {count} rules from {path}")

        return count

    def export_rules(self, path: str, tags: Optional[list[str]] = None) -> int:
        """
        Export rules to a JSON file.

        Parameters
        ----------
        path : str
            Path to export to
        tags : list[str], optional
            Only export rules with these tags

        Returns
        -------
        int
            Number of rules exported
        """
        export_path = Path(os.path.expanduser(path))
        export_path.parent.mkdir(parents=True, exist_ok=True)

        rules_to_export = self.list_rules(tags=tags)

        with open(export_path, "w", encoding="utf-8") as f:
            data = {r.id: r.to_dict() for r in rules_to_export}
            json.dump(data, f, indent=2)

        logger.info(f"Exported {len(rules_to_export)} rules to {path}")

        return len(rules_to_export)

    def clear(self) -> None:
        """Remove all rules from the store."""
        self._rules = {}
        self._save()
        logger.info("Cleared all rules")

    def __len__(self) -> int:
        return len(self._rules)

    def __contains__(self, rule_id: str) -> bool:
        return rule_id in self._rules

    def __iter__(self):
        return iter(self._rules.values())

clear()

Remove all rules from the store.

Source code in featcopilot/stores/rule_store.py 
330
331
332
333
334
def clear(self) -> None:
    """Remove all rules from the store."""
    self._rules = {}
    self._save()
    logger.info("Cleared all rules")

delete_rule(rule_id)

Delete a rule by ID.

Parameters:

Name Type Description Default
rule_id str

The rule's ID

 required

Returns:

Type Description
bool

True if deleted, False if not found
Source code in featcopilot/stores/rule_store.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def delete_rule(self, rule_id: str) -> bool:
    """
    Delete a rule by ID.

    Parameters
    ----------
    rule_id : str
        The rule's ID

    Returns
    -------
    bool
        True if deleted, False if not found
    """
    if rule_id in self._rules:
        del self._rules[rule_id]
        self._save()
        logger.info(f"Deleted rule {rule_id}")
        return True
    return False

export_rules(path, tags=None)

Export rules to a JSON file.

Parameters:

Name Type Description Default
path str

Path to export to

 required
tags list[str]

Only export rules with these tags

 None

Returns:

Type Description
int

Number of rules exported
Source code in featcopilot/stores/rule_store.py 
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
def export_rules(self, path: str, tags: Optional[list[str]] = None) -> int:
    """
    Export rules to a JSON file.

    Parameters
    ----------
    path : str
        Path to export to
    tags : list[str], optional
        Only export rules with these tags

    Returns
    -------
    int
        Number of rules exported
    """
    export_path = Path(os.path.expanduser(path))
    export_path.parent.mkdir(parents=True, exist_ok=True)

    rules_to_export = self.list_rules(tags=tags)

    with open(export_path, "w", encoding="utf-8") as f:
        data = {r.id: r.to_dict() for r in rules_to_export}
        json.dump(data, f, indent=2)

    logger.info(f"Exported {len(rules_to_export)} rules to {path}")

    return len(rules_to_export)

find_matching_rules(columns=None, description=None, tags=None, min_usage=0)

Find rules that can be applied to the given context.

Parameters:

Name Type Description Default
columns list[str]

Available column names to match against

 None
description str

Description to search for (keyword matching)

 None
tags list[str]

Required tags

 None
min_usage int

Minimum usage count

 0

Returns:

Type Description
list[tuple[TransformRule, dict]]

List of (rule, column_mapping) tuples for applicable rules, sorted by usage count (most used first)
Source code in featcopilot/stores/rule_store.py 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
def find_matching_rules(
    self,
    columns: Optional[list[str]] = None,
    description: Optional[str] = None,
    tags: Optional[list[str]] = None,
    min_usage: int = 0,
) -> list[tuple[TransformRule, dict[str, str]]]:
    """
    Find rules that can be applied to the given context.

    Parameters
    ----------
    columns : list[str], optional
        Available column names to match against
    description : str, optional
        Description to search for (keyword matching)
    tags : list[str], optional
        Required tags
    min_usage : int, default=0
        Minimum usage count

    Returns
    -------
    list[tuple[TransformRule, dict]]
        List of (rule, column_mapping) tuples for applicable rules,
        sorted by usage count (most used first)
    """
    results: list[tuple[TransformRule, dict[str, str]]] = []

    for rule in self._rules.values():
        # Filter by usage count
        if rule.usage_count < min_usage:
            continue

        # Filter by tags
        if tags and not all(t in rule.tags for t in tags):
            continue

        # Filter by description keywords
        if description:
            keywords = description.lower().split()
            rule_text = f"{rule.name} {rule.description}".lower()
            if not any(kw in rule_text for kw in keywords):
                continue

        # Check column compatibility
        mapping: dict[str, str] = {}
        if columns:
            matches, mapping = rule.matches_columns(columns)
            if not matches:
                continue

        results.append((rule, mapping))

    # Sort by usage count (descending)
    results.sort(key=lambda x: x[0].usage_count, reverse=True)

    return results

get_rule(rule_id)

Get a rule by ID.

Parameters:

Name Type Description Default
rule_id str

The rule's ID

 required

Returns:

Type Description
TransformRule or None

The rule if found, None otherwise
Source code in featcopilot/stores/rule_store.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def get_rule(self, rule_id: str) -> Optional[TransformRule]:
    """
    Get a rule by ID.

    Parameters
    ----------
    rule_id : str
        The rule's ID

    Returns
    -------
    TransformRule or None
        The rule if found, None otherwise
    """
    return self._rules.get(rule_id)

get_rule_by_name(name)

Get a rule by name.

Parameters:

Name Type Description Default
name str

The rule's name

 required

Returns:

Type Description
TransformRule or None

The first rule matching the name, None if not found
Source code in featcopilot/stores/rule_store.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def get_rule_by_name(self, name: str) -> Optional[TransformRule]:
    """
    Get a rule by name.

    Parameters
    ----------
    name : str
        The rule's name

    Returns
    -------
    TransformRule or None
        The first rule matching the name, None if not found
    """
    for rule in self._rules.values():
        if rule.name == name:
            return rule
    return None

import_rules(path, merge=True)

Import rules from another JSON file.

Parameters:

Name Type Description Default
path str

Path to import from

 required
merge bool

If True, merge with existing rules. If False, replace all.

 True

Returns:

Type Description
int

Number of rules imported
Source code in featcopilot/stores/rule_store.py 
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
def import_rules(self, path: str, merge: bool = True) -> int:
    """
    Import rules from another JSON file.

    Parameters
    ----------
    path : str
        Path to import from
    merge : bool, default=True
        If True, merge with existing rules. If False, replace all.

    Returns
    -------
    int
        Number of rules imported
    """
    import_path = Path(os.path.expanduser(path))

    if not import_path.exists():
        raise FileNotFoundError(f"Import file not found: {path}")

    with open(import_path, encoding="utf-8") as f:
        data = json.load(f)

    if not merge:
        self._rules = {}

    count = 0
    for _rule_id, rule_data in data.items():
        rule = TransformRule.from_dict(rule_data)
        self._rules[rule.id] = rule
        count += 1

    self._save()
    logger.info(f"Imported {count} rules from {path}")

    return count

list_rules(tags=None)

List all rules, optionally filtered by tags.

Parameters:

Name Type Description Default
tags list[str]

Filter rules that have all specified tags

 None

Returns:

Type Description
list[TransformRule]

List of matching rules
Source code in featcopilot/stores/rule_store.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
def list_rules(self, tags: Optional[list[str]] = None) -> list[TransformRule]:
    """
    List all rules, optionally filtered by tags.

    Parameters
    ----------
    tags : list[str], optional
        Filter rules that have all specified tags

    Returns
    -------
    list[TransformRule]
        List of matching rules
    """
    rules = list(self._rules.values())

    if tags:
        rules = [r for r in rules if all(t in r.tags for t in tags)]

    return rules

save_rule(rule)

Save a rule to the store.

Parameters:

Name Type Description Default
rule TransformRule

The rule to save

 required

Returns:

Type Description
str

The rule's ID
Source code in featcopilot/stores/rule_store.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def save_rule(self, rule: TransformRule) -> str:
    """
    Save a rule to the store.

    Parameters
    ----------
    rule : TransformRule
        The rule to save

    Returns
    -------
    str
        The rule's ID
    """
    self._rules[rule.id] = rule
    self._save()
    logger.info(f"Saved rule '{rule.name}' with ID {rule.id}")
    return rule.id

search_by_description(query, limit=10)

Search rules by description similarity.

Parameters:

Name Type Description Default
query str

Search query

 required
limit int

Maximum number of results

 10

Returns:

Type Description
list[TransformRule]

Matching rules sorted by relevance
Source code in featcopilot/stores/rule_store.py 
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
def search_by_description(self, query: str, limit: int = 10) -> list[TransformRule]:
    """
    Search rules by description similarity.

    Parameters
    ----------
    query : str
        Search query
    limit : int, default=10
        Maximum number of results

    Returns
    -------
    list[TransformRule]
        Matching rules sorted by relevance
    """
    query_words = set(query.lower().split())
    scored_rules: list[tuple[float, TransformRule]] = []

    for rule in self._rules.values():
        rule_words = set(f"{rule.name} {rule.description}".lower().split())

        # Simple word overlap scoring
        overlap = len(query_words & rule_words)
        if overlap > 0:
            score = overlap / len(query_words)
            scored_rules.append((score, rule))

    # Sort by score descending
    scored_rules.sort(key=lambda x: x[0], reverse=True)

    return [rule for _, rule in scored_rules[:limit]]