Merge branch 'main' into dependabot/pip/black-24.3.0

modelscan/middlewares/__init__.py

@@ -0,0 +1 @@
+from modelscan.middlewares.format_via_extension import FormatViaExtensionMiddleware

modelscan/middlewares/format_via_extension.py

@@ -0,0 +1,17 @@
+from .middleware import MiddlewareBase
+from modelscan.model import Model
+from typing import Callable
+
+
+class FormatViaExtensionMiddleware(MiddlewareBase):
+    def __call__(self, model: Model, call_next: Callable[[Model], None]) -> None:
+        extension = model.get_source().suffix
+        formats = [
+            format
+            for format, extensions in self._settings["formats"].items()
+            if extension in extensions
+        ]
+        if len(formats) > 0:
+            model.set_context("formats", model.get_context("formats") or [] + formats)
+
+        call_next(model)

modelscan/middlewares/middleware.py

@@ -0,0 +1,59 @@
+import abc
+from modelscan.model import Model
+from typing import Callable, Dict, Any, List
+import importlib
+
+
+class MiddlewareImportError(Exception):
+    pass
+
+
+class MiddlewareBase(metaclass=abc.ABCMeta):
+    _settings: Dict[str, Any]
+
+    def __init__(self, settings: Dict[str, Any]):
+        self._settings = settings
+
+    @abc.abstractmethod
+    def __call__(
+        self,
+        model: Model,
+        call_next: Callable[[Model], None],
+    ) -> None:
+        raise NotImplementedError
+
+
+class MiddlewarePipeline:
+    _middlewares: List[MiddlewareBase]
+
+    def __init__(self) -> None:
+        self._middlewares = []
+
+    @staticmethod
+    def from_settings(middleware_settings: Dict[str, Any]) -> "MiddlewarePipeline":
+        pipeline = MiddlewarePipeline()
+
+        for path, params in middleware_settings.items():
+            try:
+                (modulename, classname) = path.rsplit(".", 1)
+                imported_module = importlib.import_module(
+                    name=modulename, package=classname
+                )
+
+                middleware_class: MiddlewareBase = getattr(imported_module, classname)
+                pipeline.add_middleware(middleware_class(params))  # type: ignore
+            except Exception as e:
+                raise MiddlewareImportError(f"Error importing middleware {path}: {e}")
+
+        return pipeline
+
+    def add_middleware(self, middleware: MiddlewareBase) -> "MiddlewarePipeline":
+        self._middlewares.append(middleware)
+        return self
+
+    def run(self, model: Model) -> None:
+        def runner(model: Model, index: int) -> None:
+            if index < len(self._middlewares):
+                self._middlewares[index](model, lambda model: runner(model, index + 1))
+
+        runner(model, 0)

modelscan/model.py

@@ -0,0 +1,54 @@
+from pathlib import Path
+from typing import Union, Optional, IO, Dict, Any
+
+
+class ModelDataEmpty(ValueError):
+    pass
+
+
+class Model:
+    _source: Path
+    _stream: Optional[IO[bytes]]
+    _source_file_used: bool
+    _context: Dict[str, Any]
+
+    def __init__(self, source: Union[str, Path], stream: Optional[IO[bytes]] = None):
+        self._source = Path(source)
+        self._stream = stream
+        self._source_file_used = False
+        self._context = {"formats": []}
+
+    def set_context(self, key: str, value: Any) -> None:
+        self._context[key] = value
+
+    def get_context(self, key: str) -> Any:
+        return self._context.get(key)
+
+    def open(self) -> "Model":
+        if self._stream:
+            return self
+
+        self._stream = open(self._source, "rb")
+        self._source_file_used = True
+
+        return self
+
+    def close(self) -> None:
+        # Only close the stream if we opened a file (not for IO[bytes] objects passed in)
+        if self._stream and self._source_file_used:
+            self._stream.close()
+
+    def __enter__(self) -> "Model":
+        return self.open()
+
+    def __exit__(self, exc_type, exc_value, traceback) -> None:  # type: ignore
+        self.close()
+
+    def get_source(self) -> Path:
+        return self._source
+
+    def get_stream(self) -> IO[bytes]:
+        if not self._stream:
+            raise ModelDataEmpty("Model data is empty.")
+
+        return self._stream