Source code for bibble.io._util

  1#!/usr/bin/env python3
  2"""
  3
  4
  5"""
  6# mypy: disable-error-code="attr-defined"
  7# Imports:
  8from __future__ import annotations
  9
 10# ##-- stdlib imports
 11import datetime
 12import enum
 13import functools as ftz
 14import itertools as itz
 15import logging as logmod
 16import pathlib as pl
 17import re
 18import time
 19import types
 20import collections
 21import contextlib
 22import hashlib
 23from copy import deepcopy
 24from uuid import UUID, uuid1
 25from weakref import ref
 26import atexit # for @atexit.register
 27import faulthandler
 28# ##-- end stdlib imports
 29
 30import bibble._interface as API
 31from bibble.model import MetaBlock
 32
 33# ##-- types
 34# isort: off
 35import abc
 36import collections.abc
 37from typing import TYPE_CHECKING, cast, assert_type, assert_never
 38from typing import Generic, NewType
 39# Protocols:
 40from typing import Protocol, runtime_checkable
 41# Typing Decorators:
 42from typing import no_type_check, final, override, overload
 43
 44if TYPE_CHECKING:
 45    from jgdv import Maybe
 46    from typing import Final
 47    from typing import ClassVar, Any, LiteralString
 48    from typing import Never, Self, Literal
 49    from typing import TypeGuard
 50    from collections.abc import Iterable, Iterator, Callable, Generator
 51    from collections.abc import Sequence, Mapping, MutableMapping, Hashable
 52
 53    from bibtexparser import Library
 54    from bibble._interface import Middleware
 55
 56##--|
 57
 58# isort: on
 59# ##-- end types
 60
 61##-- logging
 62logging = logmod.getLogger(__name__)
 63##-- end logging
 64
 65# Vars:
 66
 67# Body:
 68

[docs]
 69class Runner_m:
 70    """
 71    Shared code for running middlewares
 72    """
 73

[docs]
 74    def _run_writewares(self, library:Library, *, append:Maybe[list[Middleware]]=None) -> Library:
 75        """ Run write transforms on the library before writing,
 76        can handle bidirectional middlewares
 77        """
 78        append     = append or []
 79        fail_count = len(library.failed_blocks)
 80        for middleware in itz.chain(self._middlewares, append):
 81            fail_count = len(library.failed_blocks)
 82            self._logger.debug("- Running Write Middleware: %s", middleware.metadata_key())
 83            if hasattr(middleware, "handle_meta_entry"):
 84                middleware.handle_meta_entry(library)
 85            match middleware:
 86                case API.Middleware_p():
 87                    library = middleware.transform(library=library)
 88                case API.BidirectionalMiddleware_p():
 89                    library = middleware.write_transform(library=library)
 90                case x:
 91                    raise TypeError(type(x))
 92
 93            if fail_count < (new_fcount:=len(library.failed_blocks)):
 94                self._logger.debug("Added %s failures", new_fcount - fail_count)
 95                fail_count = new_fcount
 96
 97        else:
 98            self._record_transform_chain("write_transforms", library, append)
 99            return library

100

[docs]
101    def _run_readwares(self, library:Library, *, append:Maybe[list[Middleware]]=None) -> Library:
102        append     = append or []
103        fail_count = len(library.failed_blocks)
104        for middleware in itz.chain(self._middlewares, append):
105            self._logger.debug("- Running Read Middleware: %s", middleware.metadata_key())
106            if hasattr(middleware, "handle_meta_entry"):
107                middleware.handle_meta_entry(library)
108            match middleware:
109                case API.Middleware_p():
110                    library = middleware.transform(library=library)
111                case API.BidirectionalMiddleware_p():
112                    library = middleware.read_transform(library=library)
113                case x:
114                    raise TypeError(type(x))
115
116            if fail_count < (new_fcount:=len(library.failed_blocks)):
117                self._logger.debug("Added %s failures", new_fcount - fail_count)
118                fail_count = new_fcount
119
120        else:
121            self._record_transform_chain("read_transforms", library, append)
122            return library

123

[docs]
124    def _record_transform_chain(self, meta_key:str, library:Library, append:list[Middleware]) -> None:
125        """
126        Record the metadata keys used on this library in a meta block
127        """
128        keys = [x.metadata_key() for x in itz.chain(self._middlewares, append)]
129        match MetaBlock.find_in(library):
130            case None:
131                library.add(MetaBlock(read_stack=keys))
132            case MetaBlock() as mb if meta_key in mb.data:
133                mb.data[meta_key] += keys
134            case MetaBlock() as mb:
135                mb.data[meta_key] = keys
136            case x:
137                raise TypeError(type(x))


138