from pathlib import Path from typing import List, Optional def ensure_has_columns(df: "pandas.DataFrame", columns: List[str]): # We don't want to import pandas too soon import pandas as pd all_columns = columns # print("all_columns={}".format(all_columns)) # print("df={}".format(df.columns.tolist())) for c in reversed(columns): if not c in df.columns.tolist(): df.insert(0, column=c, value=pd.Series()) # print("df={}".format(df.columns.tolist())) return df def run_filters(filters, obj): for f in filters: if not f(obj): return False return True def any(filters): def f(obj): for f in filters: if f(obj): return True return False return f def all(filters): def f(obj): for f in filters: if not f(obj): return False return True return f class HttpCache(object): def __init__(self, path: Path): self.path = path def lookup(self, key): cache_path = self._make_path(key) if cache_path.exists(): with open(str(cache_path), "r") as f: return f.read() def save(self, key, value): cache_path = self._make_path(key) cache_path.parent.mkdir(parents=True, exist_ok=True) with cache_path.open("w") as f: f.write(value) def _make_path(self, key) -> Path: return self.path / "{}.html".format(key) class EmptyHttpCache(object): @staticmethod def lookup(key): return None def save(self, key, value): pass def maybe_cache(path: Optional[Path]) -> HttpCache: return HttpCache(path) if path is not None else EmptyHttpCache()