Added live-reload flag for updating spec on every request

src/apispec_fromfile/plugin.py

@@ -25,26 +25,55 @@ def operation_helper(self, path=None, operations=None, **kwargs):
         # get specs
         specs = getattr(view, "__apispec__", None)
 
+        if specs is None:
+            get_specs = getattr(view, "__get_apispec__", None)
+
+            if get_specs is not None:
+                specs = get_specs()
+
         # update operations
         if specs is not None:
             operations.update(specs)
 
 
-def from_file(spec_path):
+def read_spec(path: str):
+    # get the file
+    path = Path(path)
+    if not path.exists():
+        return None
+
+    # get the content
+    return path.read_text()
+
+
+def load_spec(func, path):
+    content = read_spec(path)
+    spec = func.__dict__.get("__previous_apispec__", {})
+
+    if content is None:
+        return spec
+
+    spec.update(load_operations_from_docstring(content))
+    func.___previous_apispec__ = spec
+
+    return spec
+
+
+def from_file(spec_path, live_reload: bool = False):
     """ Decorate an endpoint with an OpenAPI spec file to import. """
 
     def wrapper(func):
-        # get the file
-        path = Path(spec_path)
-        if not path.exists():
-            return func
+        # save the content in a special attribute of the function
+        content = read_spec(spec_path)
 
-        # get the content
-        content = path.read_text()
+        if content is None:
+            return func
 
-        # save the content in a special attribute of the function
-        func.__apispec__ = func.__dict__.get("__apispec__", {})
-        func.__apispec__.update(load_operations_from_docstring(content))
+        if live_reload:
+            func.__get_apispec__ = lambda: load_spec(func, spec_path)
+        else:
+            func.__apispec__ = func.__dict__.get("__apispec__", {})
+            func.__apispec__.update(load_operations_from_docstring(content))
 
         return func
 

tests/test_live_reload.py

@@ -0,0 +1,94 @@
+""" Test for apispec_fromfile """
+
+from apispec import APISpec
+from apispec.yaml_utils import load_operations_from_docstring
+
+from apispec_fromfile import FromFilePlugin
+from apispec_fromfile import from_file
+
+
+def write_yaml_file(path, summary: str = 'Hello'):
+    """
+    Generate method spec with given summary and save in an external file
+    """
+
+    yaml_content = f"""
+    ---
+    get:
+      summary: {summary}
+      operationId: hello
+      responses:
+        '200':
+          content:
+            application/json:
+              schema:
+                type: string
+    """
+    yaml_file = path / "hello.yml"
+    yaml_file.write_text(yaml_content)
+
+    return yaml_file, yaml_content
+
+
+def make_spec(func):
+    """
+    Create apispec object with default params for passed '/hello' method handler
+    """
+
+    spec = APISpec(
+        title="Petstore",
+        version="1.0.0",
+        openapi_version="3.0.3",
+        plugins=[FromFilePlugin("func")],
+    )
+
+    spec.path("/hello", func=func)
+
+    return spec
+
+
+def test_spec_is_not_updated_without_live_reload_flag(tmp_path):
+    """
+    Ensure that when external file changed,
+    the method spec is not updated when not using live-reload flag
+    """
+    yaml_file, yaml_content = write_yaml_file(tmp_path)
+
+    @from_file(str(yaml_file))
+    def hello():
+        return "hello"
+
+    assert load_operations_from_docstring(yaml_content) == make_spec(hello).to_dict()['paths']['/hello']
+
+    # update file contents
+    yaml_file, yaml_content_updated = write_yaml_file(tmp_path, summary = 'Hello world')
+
+    # check that yaml content has changed, but the method spec has not
+
+    assert load_operations_from_docstring(yaml_content) != load_operations_from_docstring(yaml_content_updated)
+    assert load_operations_from_docstring(yaml_content) == make_spec(hello).to_dict()['paths']['/hello']
+    assert load_operations_from_docstring(yaml_content_updated) != make_spec(hello).to_dict()['paths']['/hello']
+
+
+def test_spec_is_updated_with_live_reload_flag(tmp_path):
+    """
+    Ensure that when external file changed,
+    the method spec is updated when using live-reload flag
+    """
+
+    yaml_file, yaml_content = write_yaml_file(tmp_path)
+
+    @from_file(str(yaml_file), live_reload = True)
+    def hello():
+        return "hello"
+
+    assert load_operations_from_docstring(yaml_content) == make_spec(hello).to_dict()['paths']['/hello']
+
+    # update file contents
+    yaml_file, yaml_content_updated = write_yaml_file(tmp_path, summary = 'Hello world')
+
+    # check that yaml content has changed, and the method spec as well
+
+    assert load_operations_from_docstring(yaml_content) != load_operations_from_docstring(yaml_content_updated)
+    assert load_operations_from_docstring(yaml_content) != make_spec(hello).to_dict()['paths']['/hello']
+    assert load_operations_from_docstring(yaml_content_updated) == make_spec(hello).to_dict()['paths']['/hello']