mandos.model.search_caches

Complexity

Total Complexity

9

Size/Duplication

Total Lines	55
Duplicated Lines	0 %

Importance

Changes

0

7 Methods

Rating	Name	Duplication	Size	Complexity
A	SearchCache.kill()	0	4	1
A	SearchCache.at()	0	3	1
A	SearchCache.path()	0	3	1
A	SearchCache.__init__()	0	11	2
A	SearchCache._meta_path()	0	3	1
A	SearchCache.save()	0	6	2
A	SearchCache.next()	0	2	1

"""
Temporary caching of search results as they progress.
"""
from pathlib import Path
from typing import Iterator, Sequence

import orjson

Unable to import 'orjson'

from suretime import Suretime

Unable to import 'suretime'

from mandos import logger


class SearchCache:

Missing class docstring

    def __init__(self, path: Path, compounds: Sequence[str]):
        self._path = path
        if self._meta_path.exists():
            self._data = orjson.loads(self._meta_path.read_text(encoding="utf8"))
            logger.caution(f"Resuming {path} with {self.at} completed compounds")
        else:
            self._data = dict(
                start=Suretime.tagged.now_utc_sys().iso, last=None, path=self.path, done=set()
            )
            logger.debug(f"Starting fresh cache for {path}")
        self._queue: Iterator[str] = iter([c for c in compounds if c not in self._data["done"]])

    def next(self) -> str:

Missing function or method docstring

        return next(self._queue)

    def save(self, *compounds: str) -> None:

Missing function or method docstring

        for c in compounds:

Variable name "c" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

            self._data["done"].add(c)
        data = orjson.dumps(self._data).decode(encoding="utf8")
        self._meta_path.write_text(data)
        logger.debug(f"Saved to {self._meta_path}")

    @property
    def path(self) -> Path:

Missing function or method docstring

        return self._path

    @property
    def at(self) -> int:

Missing function or method docstring

Attribute name "at" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

        return len(self._data["done"])

    @property
    def _meta_path(self) -> Path:
        return self._path.parent / (self._path.name + ".meta.json.tmp")

    def kill(self) -> None:

Missing function or method docstring

        self._data = None
        self._meta_path.unlink()
        logger.debug(f"Destroyed search cache {self._meta_path}")


__all__ = ["SearchCache"]