remake loading from yaml files. (#35)

* remake loading from yaml files.

Author: flask-pro

CHANGES.md

@@ -1,3 +1,7 @@
+## Version 0.19.0
+
+* Remake loading from yaml files.
+
 ## Version 0.18.2
 
 * Add setting `FIRST_DATETIME_FORMAT`.

pyproject.toml

@@ -23,7 +23,7 @@ license = {file = "LICENSE"}
 name = "Flask-First"
 readme = "README.md"
 requires-python = ">=3.9"
-version = "0.18.2"
+version = "0.19.0"
 
 [project.optional-dependencies]
 dev = [

src/flask_first/first/exceptions.py

@@ -5,8 +5,12 @@ class FirstException(Exception):
     """Common exception."""
 
 
-class FirstOpenAPIResolverError(FirstException):
-    """Exception for specification resolver error."""
+class FirstYAMLReaderError(FirstException):
+    """Exception for yaml file loading error."""
+
+
+class FirstResolverError(FirstException):
+    """Exception for specification from yaml file resolver error."""
 
 
 class FirstOpenAPIValidation(FirstException):

src/flask_first/first/loaders/__init__.py

@@ -0,0 +1,3 @@
+from .yaml_loader import load_from_yaml
+
+__all__ = ['load_from_yaml']

src/flask_first/first/loaders/yaml_loader.py

@@ -0,0 +1,127 @@
+from collections.abc import Hashable
+from functools import reduce
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+from ..exceptions import FirstResolverError
+from ..exceptions import FirstYAMLReaderError
+
+
+class YAMLReader:
+    """
+    Open OpenAPI specification from yaml file. The specification from multiple files is supported.
+    """
+
+    def __init__(self, path: Path):
+        self.path = path
+        self.root_file_name = self.path.name
+        self.store = {}
+
+    @staticmethod
+    def _yaml_to_dict(path: Path) -> dict:
+        with open(path) as f:
+            s = yaml.safe_load(f)
+        return s
+
+    def add_file_to_store(self, ref: str) -> None:
+        try:
+            file_path, node_path = ref.split('#/')
+        except (AttributeError, ValueError):
+            raise FirstYAMLReaderError(f'"$ref" with value <{ref}> is not valid.')
+
+        if file_path and file_path not in self.store:
+            path_to_spec_file = Path(self.path.parent, file_path)
+
+            try:
+                self.store[file_path] = self._yaml_to_dict(path_to_spec_file)
+            except FileNotFoundError:
+                raise FirstYAMLReaderError(f'No such file or directory: <{file_path}>')
+
+    def search_file(self, obj: dict or list) -> None:
+        if isinstance(obj, dict):
+            ref = obj.get('$ref')
+            if ref:
+                self.add_file_to_store(ref)
+            else:
+                for _, v in obj.items():
+                    self.search_file(v)
+
+        elif isinstance(obj, list):
+            for item in obj:
+                self.search_file(item)
+
+        else:
+            return
+
+    def load(self) -> 'YAMLReader':
+        root_file = self._yaml_to_dict(self.path)
+        self.store[self.root_file_name] = root_file
+        self.search_file(root_file)
+        return self
+
+
+class Resolver:
+    def __init__(self, yaml_reader: YAMLReader):
+        self.yaml_reader = yaml_reader
+        self.resolved_spec = None
+
+    def _get_schema_via_local_ref(self, file_path: str, node_path: str) -> dict:
+        keys = node_path.split('/')
+
+        def get_value_of_key_from_dict(source_dict: dict, key: Hashable) -> Any:
+            return source_dict[key]
+
+        try:
+            return reduce(get_value_of_key_from_dict, keys, self.yaml_reader.store[file_path])
+        except KeyError:
+            raise FirstResolverError(f'No such path: "{node_path}"')
+
+    def _get_schema(self, root_file_path: str, ref: str) -> Any:
+        try:
+            file_path, node_path = ref.split('#/')
+        except (AttributeError, ValueError):
+            raise FirstResolverError(
+                f'"$ref" with value <{ref}> is not valid in file <{root_file_path}>'
+            )
+
+        if file_path and node_path:
+            obj = self._get_schema_via_local_ref(file_path, node_path)
+
+        elif node_path and not file_path:
+            obj = self._get_schema_via_local_ref(root_file_path, node_path)
+
+        else:
+            raise NotImplementedError
+
+        return obj
+
+    def _resolving_all_refs(self, file_path: str, obj: Any) -> Any:
+        if isinstance(obj, dict):
+            ref = obj.get('$ref', ...)
+            if ref is not ...:
+                obj = self._resolving_all_refs(file_path, self._get_schema(file_path, ref))
+            else:
+                for key, value in obj.items():
+                    obj[key] = self._resolving_all_refs(file_path, value)
+
+        if isinstance(obj, list):
+            objs = []
+            for item_obj in obj:
+                objs.append(self._resolving_all_refs(file_path, item_obj))
+            obj = objs
+
+        return obj
+
+    def resolving(self) -> 'Resolver':
+        root_file_path = self.yaml_reader.root_file_name
+        root_spec = self.yaml_reader.store[root_file_path]
+        self.resolved_spec = self._resolving_all_refs(root_file_path, root_spec)
+        return self
+
+
+def load_from_yaml(path: Path) -> Resolver:
+    yaml_reader = YAMLReader(path).load()
+    resolved_obj = Resolver(yaml_reader).resolving()
+    return resolved_obj.resolved_spec

src/flask_first/first/specification.py

@@ -1,95 +1,17 @@
-from collections.abc import Hashable
-from collections.abc import Mapping
 from copy import deepcopy
-from functools import reduce
 from pathlib import Path
-from typing import Any
 from typing import Optional
 
 from openapi_spec_validator import validate
-from openapi_spec_validator.readers import read_from_filename
 from openapi_spec_validator.validation.exceptions import OpenAPIValidationError
 
 from ..schema.schema_maker import make_marshmallow_schema
-from .exceptions import FirstOpenAPIResolverError
 from .exceptions import FirstOpenAPIValidation
+from .loaders import load_from_yaml
 from .validator import OpenAPI310ValidationError
 from .validator import Validator
 
 
-class Resolver:
-    """
-    This class creates a dictionary from the specification that contains resolved schema references.
-    Specification from several files are supported.
-    """
-
-    def __init__(self, abs_path: Path or str):
-        self.abs_path = Path(abs_path)
-        self.root_dir = self.abs_path.resolve().parent
-
-    @staticmethod
-    def file_to_dict(abs_path: Path) -> Mapping[Hashable, Any]:
-        try:
-            spec_as_dict, _ = read_from_filename(str(abs_path))
-        except OSError as e:
-            raise FirstOpenAPIResolverError(e)
-        return spec_as_dict
-
-    @staticmethod
-    def get_value_of_key_from_dict(source_dict: dict, key: Hashable) -> Any or KeyError:
-        return source_dict[key]
-
-    def _get_schema_via_local_ref(self, root_schema: dict, keys: dict) -> dict:
-        try:
-            return reduce(self.get_value_of_key_from_dict, keys, root_schema)
-        except KeyError:
-            raise FirstOpenAPIResolverError(f'No such path: {keys}')
-
-    def _get_schema_from_file_ref(self, root_dir: Path, relative_path: Path, keys: dict) -> dict:
-        abs_path_file = Path(root_dir, relative_path)
-        root_schema = self.file_to_dict(abs_path_file)
-        return self._get_schema_via_local_ref(root_schema, keys)
-
-    def _resolving(self, schema: dict, relative_path_to_file_schema: str) -> dict or list[dict]:
-        if isinstance(schema, dict):
-            if '$ref' in schema:
-                try:
-                    relative_file_path_from_ref, local_path = schema['$ref'].split('#/')
-                except AttributeError:
-                    raise FirstOpenAPIResolverError(f'"$ref" <{schema["$ref"]}> must be string.')
-
-                local_path_parts = local_path.split('/')
-
-                if relative_file_path_from_ref:
-                    schema_from_ref = self._get_schema_from_file_ref(
-                        self.root_dir, relative_file_path_from_ref, local_path_parts
-                    )
-                    schema = self._resolving(schema_from_ref, relative_file_path_from_ref)
-                else:
-                    schema_from_ref = self._get_schema_from_file_ref(
-                        self.root_dir, relative_path_to_file_schema, local_path_parts
-                    )
-                    schema = self._resolving(schema_from_ref, relative_path_to_file_schema)
-
-            else:
-                for key, value in schema.items():
-                    schema[key] = self._resolving(value, relative_path_to_file_schema)
-
-            return schema
-
-        if isinstance(schema, list):
-            schemas = []
-            for item in schema:
-                schemas.append(self._resolving(item, relative_path_to_file_schema))
-            return schemas
-
-        return schema
-
-    def resolve(self) -> Mapping[Hashable, Any]:
-        schema = self.file_to_dict(self.abs_path)
-        return self._resolving(schema, self.abs_path)
-
-
 class Specification:
     def __init__(
         self,
@@ -100,7 +22,7 @@ def __init__(
         self.path = path
         self.datetime_format = datetime_format
         self.experimental_validator = experimental_validator
-        self.raw_spec = Resolver(self.path).resolve()
+        self.raw_spec = load_from_yaml(self.path)
         self._validating_openapi_file(self.path, self.experimental_validator)
         self.resolved_spec = self._convert_parameters_to_schema(self.raw_spec)
         self.serialized_spec = self._convert_schemas(self.resolved_spec)

tests/first/loaders/yaml_loader/conftest.py

@@ -0,0 +1,51 @@
+from pathlib import Path
+
+import pytest
+import yaml
+from openapi_spec_validator import validate as osv_validate
+from openapi_spec_validator.readers import read_from_filename
+
+
+@pytest.fixture
+def fx_spec_minimal():
+    payload = {
+        'openapi': '3.1.1',
+        'info': {'title': 'API for testing Flask-First', 'version': '1.0.0'},
+        'paths': {
+            '/endpoint': {
+                'get': {
+                    'operationId': 'endpoint',
+                    'responses': {
+                        '200': {
+                            'description': 'OK',
+                            'content': {
+                                'application/json': {
+                                    'schema': {
+                                        'type': 'object',
+                                        'properties': {'message': {'type': 'string'}},
+                                    }
+                                }
+                            },
+                        }
+                    },
+                }
+            }
+        },
+    }
+    return payload
+
+
+@pytest.fixture
+def fx_spec_as_file(tmp_path):
+    def create(spec: dict, validate: bool = True, file_name: str = 'openapi.yaml') -> Path:
+        spec_path = Path(tmp_path, file_name)
+        with open(spec_path, 'w+') as f:
+            yaml.dump(spec, f)
+
+        spec_as_dict, _ = read_from_filename(str(spec_path))
+        if validate:
+            osv_validate(spec_as_dict)
+
+        return spec_path
+
+    return create

tests/first/loaders/yaml_loader/test_loaders__yaml_loader__multiple_files.py

@@ -0,0 +1,27 @@
+import pytest
+
+from src.flask_first.first.exceptions import FirstYAMLReaderError
+from src.flask_first.first.loaders.yaml_loader import load_from_yaml
+
+
+def test_loaders__yaml__multiple__response_ref(fx_spec_minimal, fx_spec_as_file):
+    endpoint_spec = {'endpoint': fx_spec_minimal['paths']['/endpoint']}
+    endpoint_spec_file = fx_spec_as_file(endpoint_spec, validate=False, file_name='endpoint.yaml')
+
+    fx_spec_minimal['paths']['/endpoint'] = {'$ref': f'{endpoint_spec_file.name}#/endpoint'}
+    spec_file = fx_spec_as_file(fx_spec_minimal, validate=False)
+    spec_obj = load_from_yaml(spec_file)
+
+    assert spec_obj['paths']['/endpoint'].get('$ref') is None
+    assert spec_obj['paths']['/endpoint'] == endpoint_spec['endpoint']
+
+
+def test_loader__internal__multiple__non_exist_file(fx_spec_minimal, fx_spec_as_file):
+    non_exist_file_name = 'non_exist_file.yaml'
+    fx_spec_minimal['paths']['/endpoint'] = {'$ref': f'{non_exist_file_name}#/endpoint'}
+    spec_file = fx_spec_as_file(fx_spec_minimal, validate=False)
+
+    with pytest.raises(FirstYAMLReaderError) as e:
+        load_from_yaml(spec_file)
+
+    assert str(e.value) == f'No such file or directory: <{non_exist_file_name}>'