Index

Bases: Record, Generic[R]

An Index is a data structure that allows for fast lookup of records by their properties. It is used internally by the Record class and once created, is automatically updated when records are created or updated.

It is not recommended using the Index class directly, but rather through the where_eq method of the Record class.

Example

MyRecord.where_eq(foo='bar')

If MyRecord has an index on the property foo, this will return all records of type MyRecord where foo is equal to 'bar'.

Creates a new index given a record_type and a single or multiple properties to index on.

Example

Index(MyRecord, 'foo')

Indexes all records of type MyRecord on the property ‘foo’.

This will create an index named 'MyRecord.foo', which is stored in the MyRecord class.

Parameters:

Name	Type	Description	Default
record_type	Type[R]	The record_type to index.	required
on	Union[str, List[str], Tuple[str]]	The properties to index on. Can be a single property or a list/tuple of properties.	required

Source code in aars/core.py

def __init__(self, record_type: Type[R], on: Union[str, List[str], Tuple[str]]):
    """
    Creates a new index given a record_type and a single or multiple properties to index on.

    Example:
        ```python
        Index(MyRecord, 'foo')
        ```
        Indexes all records of type MyRecord on the property 'foo'.

    This will create an index named `'MyRecord.foo'`, which is stored in the `MyRecord` class.

    Args:
        record_type: The record_type to index.
        on: The properties to index on. Can be a single property or a list/tuple of properties.
    """
    if isinstance(on, str):
        on = [on]
    # check if all properties exist
    for prop in on:
        if prop not in record_type.__fields__:
            raise ValueError(f"Property {prop} does not exist on {record_type.__name__}")
    super(Index, self).__init__(record_type=record_type, index_on=sorted(on))
    record_type.add_index(self)

add_record(obj: R)

Adds a record to the index.

Source code in aars/core.py

def add_record(self, obj: R):
    """Adds a record to the index."""
    assert issubclass(type(obj), Record)
    assert obj.id_hash is not None
    key = attrgetter(*self.index_on)(obj)
    if isinstance(key, str):
        key = (key,)
    if isinstance(key, list):
        key = tuple(key)
    if key not in self.hashmap:
        self.hashmap[key] = set()
    self.hashmap[key].add(obj.id_hash)

lookup(query: IndexQuery) -> AsyncIterator[R]

Fetches records with given values for the indexed properties.

Example

index = Index(MyRecord, 'foo')
index_query = IndexQuery(MyRecord, **{'foo': 'bar'})
index.lookup(index_query)

Returns all records of type MyRecord where foo is equal to 'bar'. This is an anti-pattern, as the IndexQuery should be created by calling MyRecord.where_eq(foo='bar') instead.

Parameters:

Name	Type	Description	Default
query	IndexQuery	The query to execute.	required

Returns:

Type	Description
AsyncIterator[R]	An async iterator over the records.

Source code in aars/core.py

def lookup(self, query: IndexQuery) -> AsyncIterator[R]:
    """
    Fetches records with given values for the indexed properties.

    Example:
        ```python
        index = Index(MyRecord, 'foo')
        index_query = IndexQuery(MyRecord, **{'foo': 'bar'})
        index.lookup(index_query)
        ```
        Returns all records of type `MyRecord` where `foo` is equal to `'bar'`.
        This is an anti-pattern, as the `IndexQuery` should be created by calling `MyRecord.where_eq(foo='bar')`
        instead.

    Args:
        query: The query to execute.
    Returns:
        An async iterator over the records.
    """
    assert query.record_type == self.record_type
    id_hashes: Optional[Set[str]]
    needs_filtering = False

    subquery = query
    if repr(self) != query.get_index_name():
        subquery = query.get_subquery(self.index_on)
        needs_filtering = True
    id_hashes = self.hashmap.get(tuple(subquery.values()))

    if id_hashes is None:
        return EmptyAsyncIterator()

    items = AARS.fetch_records(self.record_type, list(id_hashes))

    if needs_filtering:
        return self._filter_index_items(items, query)
    return items

regenerate(items: List[R])

Regenerates the index with given items.

Source code in aars/core.py

def regenerate(self, items: List[R]):
    """Regenerates the index with given items."""
    self.hashmap = {}
    for item in items:
        self.add_record(item)

remove_record(obj: R)

Removes a record from the index, i.e. when it is forgotten.

Source code in aars/core.py

def remove_record(self, obj: R):
    """Removes a record from the index, i.e. when it is forgotten."""
    assert obj.id_hash is not None
    key = attrgetter(*self.index_on)(obj)
    if isinstance(key, str):
        key = (key,)
    if key in self.hashmap:
        self.hashmap[key].remove(obj.id_hash)