patterns_query

patterns_query

Pattern query and lookup mixin for GlobalLearningStore.

Provides methods for querying, filtering, and retrieving patterns: - get_patterns: Query with filtering by type, priority, tags, quarantine, trust - get_pattern_by_id: Single pattern lookup - get_pattern_provenance: Provenance information retrieval - _row_to_pattern_record: Database row to PatternRecord conversion - _row_to_discovery_event: Database row to PatternDiscoveryEvent conversion

This is the "hub" mixin — other pattern sub-mixins depend on its query methods.

Classes

PatternQueryProtocol

Bases: Protocol

Protocol declaring the query methods provided by PatternQueryMixin.

Dependent mixins (TrustMixin, SuccessFactorsMixin, QuarantineMixin, BroadcastMixin) inherit from this under TYPE_CHECKING so mypy knows these methods will exist at runtime through mixin composition.

PatternQueryMixin

Mixin providing pattern query methods for GlobalLearningStore.

This mixin requires that the composed class provides: - _get_connection(): Context manager yielding sqlite3.Connection

Functions

get_patterns

get_patterns(pattern_type=None, min_priority=0.01, limit=20, context_tags=None, quarantine_status=None, exclude_quarantined=False, min_trust=None, max_trust=None, include_inactive=False, instrument_name=None, include_universal=True)

Get patterns from the global store.

v19 Evolution: Extended with quarantine and trust filtering options. v14 (cycle 2): Extended with soft-delete and instrument filtering.

Parameters:

Name	Type	Description	Default
pattern_type	str | None	Optional filter by pattern type.	None
min_priority	float	Minimum priority score to include.	0.01
limit	int	Maximum number of patterns to return.	20
context_tags	list[str] | None	Optional list of tags for context-based filtering. Patterns match if ANY of their tags match ANY query tag. If None or empty, no tag filtering is applied.	None
quarantine_status	QuarantineStatus | None	Filter by specific quarantine status.	None
exclude_quarantined	bool	If True, exclude QUARANTINED patterns.	False
min_trust	float | None	Filter patterns with trust_score >= this value.	None
max_trust	float | None	Filter patterns with trust_score <= this value.	None
include_inactive	bool	If True, include soft-deleted patterns (active=0).	False
instrument_name	str | None	Filter by instrument name. None means no filter.	None
include_universal	bool	If True (default) AND instrument_name is set, also include patterns where instrument_name is NULL (universal patterns applicable to all instruments).	True

Returns:

Type	Description
list[PatternRecord]	List of PatternRecord objects sorted by priority.

Source code in src/marianne/learning/store/patterns_query.py

def get_patterns(
    self,
    pattern_type: str | None = None,
    min_priority: float = 0.01,
    limit: int = 20,
    context_tags: list[str] | None = None,
    quarantine_status: QuarantineStatus | None = None,
    exclude_quarantined: bool = False,
    min_trust: float | None = None,
    max_trust: float | None = None,
    include_inactive: bool = False,
    instrument_name: str | None = None,
    include_universal: bool = True,
) -> list[PatternRecord]:
    """Get patterns from the global store.

    v19 Evolution: Extended with quarantine and trust filtering options.
    v14 (cycle 2): Extended with soft-delete and instrument filtering.

    Args:
        pattern_type: Optional filter by pattern type.
        min_priority: Minimum priority score to include.
        limit: Maximum number of patterns to return.
        context_tags: Optional list of tags for context-based filtering.
                     Patterns match if ANY of their tags match ANY query tag.
                     If None or empty, no tag filtering is applied.
        quarantine_status: Filter by specific quarantine status.
        exclude_quarantined: If True, exclude QUARANTINED patterns.
        min_trust: Filter patterns with trust_score >= this value.
        max_trust: Filter patterns with trust_score <= this value.
        include_inactive: If True, include soft-deleted patterns (active=0).
        instrument_name: Filter by instrument name. None means no filter.
        include_universal: If True (default) AND instrument_name is set,
                          also include patterns where instrument_name is NULL
                          (universal patterns applicable to all instruments).

    Returns:
        List of PatternRecord objects sorted by priority.
    """
    with self._get_connection() as conn:
        wb = WhereBuilder()
        wb.add("priority_score >= ?", min_priority)

        # v14: Soft-delete filter — COALESCE handles NULL for pre-v14 rows
        if not include_inactive:
            wb.add("COALESCE(active, 1) = 1")

        if pattern_type:
            wb.add("pattern_type = ?", pattern_type)

        # v14: Instrument name filter — when specified, match exact instrument
        # or (if include_universal) also include NULL (universal) patterns
        if instrument_name is not None:
            if include_universal:
                wb.add("(instrument_name = ? OR instrument_name IS NULL)", instrument_name)
            else:
                wb.add("instrument_name = ?", instrument_name)

        # v19: Quarantine status filtering
        if quarantine_status is not None:
            wb.add("quarantine_status = ?", quarantine_status.value)
        elif exclude_quarantined:
            wb.add("quarantine_status != ?", QuarantineStatus.QUARANTINED.value)

        # v19: Trust score filtering
        if min_trust is not None:
            wb.add("trust_score >= ?", min_trust)
        if max_trust is not None:
            wb.add("trust_score <= ?", max_trust)

        # Context tag filtering: match if ANY pattern tag matches ANY query tag
        # Uses json_each() to iterate over the JSON array stored in context_tags
        if context_tags is not None and len(context_tags) > 0:
            tag_placeholders = ", ".join("?" for _ in context_tags)
            wb.add(
                f"""EXISTS (
                    SELECT 1 FROM json_each(context_tags)
                    WHERE json_each.value IN ({tag_placeholders})
                )""",
                *context_tags,
            )

        where_sql, params = wb.build()
        cursor = conn.execute(
            f"""
            SELECT * FROM patterns
            WHERE {where_sql}
            ORDER BY priority_score DESC
            LIMIT ?
            """,
            (*params, limit),
        )

        return [self._row_to_pattern_record(row) for row in cursor.fetchall()]

get_pattern_by_id

get_pattern_by_id(pattern_id)

Get a single pattern by its ID.

Parameters:

Name	Type	Description	Default
pattern_id	str	The pattern ID to retrieve.	required

Returns:

Type	Description
PatternRecord | None	PatternRecord if found, None otherwise.

Source code in src/marianne/learning/store/patterns_query.py

def get_pattern_by_id(self, pattern_id: str) -> PatternRecord | None:
    """Get a single pattern by its ID.

    Args:
        pattern_id: The pattern ID to retrieve.

    Returns:
        PatternRecord if found, None otherwise.
    """
    with self._get_connection() as conn:
        cursor = conn.execute(
            "SELECT * FROM patterns WHERE id = ?",
            (pattern_id,),
        )
        row = cursor.fetchone()
        if row:
            return self._row_to_pattern_record(row)
        return None

get_pattern_provenance

get_pattern_provenance(pattern_id)

Get provenance information for a pattern.

Returns details about the pattern's origin and lifecycle.

Parameters:

Name	Type	Description	Default
pattern_id	str	The pattern ID to query.	required

Returns:

Type	Description
dict[str, Any] | None	Dict with provenance info, or None if pattern not found.

Source code in src/marianne/learning/store/patterns_query.py

def get_pattern_provenance(self, pattern_id: str) -> dict[str, Any] | None:
    """Get provenance information for a pattern.

    Returns details about the pattern's origin and lifecycle.

    Args:
        pattern_id: The pattern ID to query.

    Returns:
        Dict with provenance info, or None if pattern not found.
    """
    pattern = self.get_pattern_by_id(pattern_id)
    if not pattern:
        return None

    return {
        "pattern_id": pattern.id,
        "pattern_name": pattern.pattern_name,
        "quarantine_status": pattern.quarantine_status.value,
        "first_seen": pattern.first_seen.isoformat(),
        "last_seen": pattern.last_seen.isoformat(),
        "last_confirmed": pattern.last_confirmed.isoformat(),
        "provenance_job_hash": pattern.provenance_job_hash,
        "provenance_sheet_num": pattern.provenance_sheet_num,
        "quarantined_at": pattern.quarantined_at.isoformat()
        if pattern.quarantined_at
        else None,
        "validated_at": pattern.validated_at.isoformat()
        if pattern.validated_at
        else None,
        "quarantine_reason": pattern.quarantine_reason,
        "trust_score": pattern.trust_score,
        "trust_calculation_date": pattern.trust_calculation_date.isoformat()
        if pattern.trust_calculation_date
        else None,
    }