X-Git-Url: https://wannabe.guru.org/gitweb/?a=blobdiff_plain;ds=sidebyside;f=logical_search.py;h=2cbe0c781f4ba7889e1b3bcc05373c66a427888e;hb=e11d9b0958b2874bd67360552e9214d965fe0d12;hp=85f946135e406e74137a632bceff2a5a1845c699;hpb=36fea7f15ed17150691b5b3ead75450e575229ef;p=python_utils.git

diff --git a/logical_search.py b/logical_search.py
index 85f9461..2cbe0c7 100644
--- a/logical_search.py
+++ b/logical_search.py
@@ -1,39 +1,45 @@
 #!/usr/bin/env python3
 
-from __future__ import annotations
+# © Copyright 2021-2022, Scott Gasch
 
-from collections import defaultdict
+"""This is a module concerned with the creation of and searching of a
+corpus of documents.  The corpus is held in memory for fast
+searching.
+
+"""
+
+from __future__ import annotations
 import enum
 import sys
-from typing import (
-    Any,
-    Dict,
-    List,
-    NamedTuple,
-    Optional,
-    Set,
-    Sequence,
-    Tuple,
-    Union,
-)
+from collections import defaultdict
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional, Sequence, Set, Tuple, Union
 
 
 class ParseError(Exception):
     """An error encountered while parsing a logical search expression."""
 
     def __init__(self, message: str):
+        super().__init__()
         self.message = message
 
 
-class Document(NamedTuple):
-    """A tuple representing a searchable document."""
+@dataclass
+class Document:
+    """A class representing a searchable document."""
+
+    # A unique identifier for each document.
+    docid: str = ''
+
+    # A set of tag strings for this document.  May be empty.
+    tags: Set[str] = field(default_factory=set)
+
+    # A list of key->value strings for this document.  May be empty.
+    properties: List[Tuple[str, str]] = field(default_factory=list)
 
-    docid: str  # a unique idenfier for the document
-    tags: Set[str]  # an optional set of tags
-    properties: List[
-        Tuple[str, str]
-    ]  # an optional set of key->value properties
-    reference: Any  # an optional reference to something else
+    # An optional reference to something else; interpreted only by
+    # caller code, ignored here.
+    reference: Optional[Any] = None
 
 
 class Operation(enum.Enum):
@@ -98,13 +104,13 @@ class Corpus(object):
     ...          )
     >>> c.query('author:Scott and important')
     {1}
+    >>> c.query('*')
+    {1, 2, 3}
     """
 
     def __init__(self) -> None:
         self.docids_by_tag: Dict[str, Set[str]] = defaultdict(set)
-        self.docids_by_property: Dict[Tuple[str, str], Set[str]] = defaultdict(
-            set
-        )
+        self.docids_by_property: Dict[Tuple[str, str], Set[str]] = defaultdict(set)
         self.docids_with_property: Dict[str, Set[str]] = defaultdict(set)
         self.documents_by_docid: Dict[str, Document] = {}
 
@@ -182,13 +188,7 @@ class Corpus(object):
     def invert_docid_set(self, original: Set[str]) -> Set[str]:
         """Invert a set of docids."""
 
-        return set(
-            [
-                docid
-                for docid in self.documents_by_docid.keys()
-                if docid not in original
-            ]
-        )
+        return {docid for docid in self.documents_by_docid if docid not in original}
 
     def get_doc(self, docid: str) -> Optional[Document]:
         """Given a docid, retrieve the previously added Document."""
@@ -213,6 +213,8 @@ class Corpus(object):
         tag1 and key:*
         """
 
+        if query == '*':
+            return set(self.documents_by_docid.keys())
         try:
             root = self._parse_query(query)
         except ParseError as e:
@@ -268,9 +270,7 @@ class Corpus(object):
                     operation = Operation.from_token(token)
                     operand_count = operation.num_operands()
                     if len(node_stack) < operand_count:
-                        raise ParseError(
-                            f"Incorrect number of operations for {operation}"
-                        )
+                        raise ParseError(f"Incorrect number of operations for {operation}")
                     for _ in range(operation.num_operands()):
                         args.append(node_stack.pop())
                     node = Node(corpus, operation, args)
@@ -297,9 +297,7 @@ class Corpus(object):
                         ok = True
                         break
                 if not ok:
-                    raise ParseError(
-                        "Unbalanced parenthesis in query expression"
-                    )
+                    raise ParseError("Unbalanced parenthesis in query expression")
 
             # and, or, not
             else:
@@ -362,9 +360,7 @@ class Node(object):
                         try:
                             key, value = tag.split(":")
                         except ValueError as v:
-                            raise ParseError(
-                                f'Invalid key:value syntax at "{tag}"'
-                            ) from v
+                            raise ParseError(f'Invalid key:value syntax at "{tag}"') from v
                         if value == "*":
                             r = self.corpus.get_docids_with_property(key)
                         else:
@@ -376,23 +372,17 @@ class Node(object):
                     raise ParseError(f"Unexpected query {tag}")
         elif self.op is Operation.DISJUNCTION:
             if len(evaled_operands) != 2:
-                raise ParseError(
-                    "Operation.DISJUNCTION (or) expects two operands."
-                )
+                raise ParseError("Operation.DISJUNCTION (or) expects two operands.")
             retval.update(evaled_operands[0])
             retval.update(evaled_operands[1])
         elif self.op is Operation.CONJUNCTION:
             if len(evaled_operands) != 2:
-                raise ParseError(
-                    "Operation.CONJUNCTION (and) expects two operands."
-                )
+                raise ParseError("Operation.CONJUNCTION (and) expects two operands.")
             retval.update(evaled_operands[0])
             retval = retval.intersection(evaled_operands[1])
         elif self.op is Operation.INVERSION:
             if len(evaled_operands) != 1:
-                raise ParseError(
-                    "Operation.INVERSION (not) expects one operand."
-                )
+                raise ParseError("Operation.INVERSION (not) expects one operand.")
             _ = evaled_operands[0]
             if isinstance(_, set):
                 retval.update(self.corpus.invert_docid_set(_))