diff --git a/examples/api_view_blueprint_demo.py b/examples/api_view_blueprint_demo.py new file mode 100644 index 00000000..d9c11f51 --- /dev/null +++ b/examples/api_view_blueprint_demo.py @@ -0,0 +1,76 @@ +# -*- coding: utf-8 -*- +# Example demonstrating how to register APIView with APIBlueprint + +from pydantic import BaseModel, Field + +from flask_openapi3 import APIBlueprint, APIView, Info, OpenAPI, Tag + +# Initialize the OpenAPI app +info = Info(title="API View with Blueprint Demo", version="1.0.0") +app = OpenAPI(__name__, info=info) + +# Create an APIBlueprint +api_bp = APIBlueprint("api_bp", __name__, url_prefix="/api") + +# Create an APIView +api_view = APIView(url_prefix="/v1", view_tags=[Tag(name="books")]) + + +# Define models +class BookPath(BaseModel): + book_id: int = Field(..., description="Book ID") + + +class BookQuery(BaseModel): + search: str | None = Field(None, description="Search query") + + +class BookBody(BaseModel): + title: str = Field(..., min_length=1, max_length=100, description="Book title") + author: str = Field(..., min_length=1, max_length=100, description="Book author") + year: int = Field(..., ge=1000, le=9999, description="Publication year") + + +# Define API views +@api_view.route("/books") +class BookListView: + @api_view.doc(summary="Get all books") + def get(self, query: BookQuery): + """Get a list of all books""" + if query.search: + return {"message": f"Searching for: {query.search}", "books": []} + return {"books": [{"id": 1, "title": "Sample Book", "author": "John Doe", "year": 2023}]} + + @api_view.doc(summary="Create a new book") + def post(self, body: BookBody): + """Create a new book""" + return {"message": "Book created", "book": body.model_dump()} + + +@api_view.route("/books/") +class BookDetailView: + @api_view.doc(summary="Get a book by ID") + def get(self, path: BookPath): + """Get details of a specific book""" + return {"book": {"id": path.book_id, "title": "Sample Book", "author": "John Doe", "year": 2023}} + + @api_view.doc(summary="Update a book") + def put(self, path: BookPath, body: BookBody): + """Update an existing book""" + return {"message": f"Book {path.book_id} updated", "book": body.model_dump()} + + @api_view.doc(summary="Delete a book") + def delete(self, path: BookPath): + """Delete a book""" + return {"message": f"Book {path.book_id} deleted"} + + +# Register the APIView with the APIBlueprint +api_bp.register_api_view(api_view) + +# Register the APIBlueprint with the app +app.register_api(api_bp) + +if __name__ == "__main__": + print("Visit http://127.0.0.1:5000/openapi for API documentation") + app.run(debug=True) diff --git a/flask_openapi3/blueprint.py b/flask_openapi3/blueprint.py index c609e9b8..ab32960c 100644 --- a/flask_openapi3/blueprint.py +++ b/flask_openapi3/blueprint.py @@ -20,6 +20,7 @@ parse_parameters, parse_rule, ) +from .view import APIView class APIBlueprint(APIScaffold, Blueprint): @@ -101,6 +102,75 @@ def register_api(self, api: "APIBlueprint") -> None: # Register the nested APIBlueprint as a blueprint self.register_blueprint(api) + def register_api_view( + self, api_view: APIView, url_prefix: str | None = None, view_kwargs: dict[Any, Any] | None = None + ) -> None: + """Register an APIView onto this APIBlueprint.""" + + if view_kwargs is None: + view_kwargs = {} + + # Merge tags from the APIView + for tag in api_view.tags: + if tag.name not in self.tag_names: + self.tags.append(tag) + self.tag_names.append(tag.name) + + # Merge paths with optional url_prefix adjustment + if url_prefix and api_view.url_prefix and url_prefix != api_view.url_prefix: + api_view.paths = {url_prefix + k.removeprefix(api_view.url_prefix): v for k, v in api_view.paths.items()} + api_view.url_prefix = url_prefix + elif url_prefix and not api_view.url_prefix: + api_view.paths = {url_prefix.rstrip("/") + "/" + k.lstrip("/"): v for k, v in api_view.paths.items()} + api_view.url_prefix = url_prefix + + # Apply blueprint's url_prefix to the paths using parse_rule + for path_url, path_item in api_view.paths.items(): + uri = parse_rule(path_url, url_prefix=self.url_prefix) + self.paths[uri] = path_item + + # Merge component schemas + self.components_schemas.update(**api_view.components_schemas) + + # Register URL rules onto the blueprint + for rule, (cls, methods) in api_view.views.items(): + # Optionally adjust rule with a local url_prefix override + _rule = rule + if url_prefix and api_view.url_prefix and url_prefix != api_view.url_prefix: + _rule = url_prefix + rule.removeprefix(api_view.url_prefix) + elif url_prefix and not api_view.url_prefix: + _rule = url_prefix.rstrip("/") + "/" + rule.lstrip("/") + + for method in methods: + func = getattr(cls, method.lower()) + # Select validate_response: method override, else view default + _validate_response = ( + func.validate_response + if getattr(func, "validate_response", None) is not None + else api_view.validate_response + ) + header, cookie, path, query, form, body, raw = parse_parameters(func, doc_ui=False) + view_func = self.create_view_func( + func, + header, + cookie, + path, + query, + form, + body, + raw, + view_class=cls, + view_kwargs=view_kwargs, + responses=getattr(func, "responses", None), + validate_response=_validate_response, + ) + + # Endpoint names for blueprints may NOT contain dots. Flask reserves '.' for + # namespacing: it automatically prefixes endpoints with "{blueprint_name}.". + options = {"endpoint": f"{cls.__name__}_{method.lower()}", "methods": [method.upper()]} + # Use the blueprint's add_url_rule method + self._add_url_rule(_rule, view_func=view_func, **options) + def _add_url_rule( self, rule, diff --git a/tests/test_api_view_blueprint.py b/tests/test_api_view_blueprint.py new file mode 100644 index 00000000..db02c4d0 --- /dev/null +++ b/tests/test_api_view_blueprint.py @@ -0,0 +1,286 @@ +# -*- coding: utf-8 -*- +# Test for registering APIView with root app, parent blueprint, and child blueprint + +import pytest +from pydantic import BaseModel, Field + +from flask_openapi3 import APIBlueprint, APIView, Info, OpenAPI, Tag + +info = Info(title="api view blueprint test", version="1.0.0") + +app = OpenAPI(__name__, info=info) +app.config["TESTING"] = True + +# Create nested blueprint structure +parent_bp = APIBlueprint("parent", __name__, url_prefix="/parent") +child_bp = APIBlueprint("child", __name__, url_prefix="/child") + +# Create three APIViews - one for root, one for parent, one for child +root_api_view = APIView(url_prefix="/root", view_tags=[Tag(name="root-books")]) +parent_api_view = APIView(url_prefix="/v1", view_tags=[Tag(name="parent-books")]) +child_api_view = APIView(url_prefix="/v2", view_tags=[Tag(name="child-books")]) + + +class BookPath(BaseModel): + id: int = Field(..., description="book ID") + + +class BookBody(BaseModel): + title: str = Field(..., min_length=1, max_length=100, description="Book title") + + +# Root level APIView (registered on app) +@root_api_view.route("/books") +class RootBookListAPIView: + @root_api_view.doc(summary="get root book list") + def get(self): + return {"level": "root", "books": []} + + @root_api_view.doc(summary="create root book") + def post(self, body: BookBody): + return {"level": "root", "book": body.model_dump()} + + +@root_api_view.route("/books/") +class RootBookAPIView: + @root_api_view.doc(summary="get root book") + def get(self, path: BookPath): + return {"level": "root", "id": path.id} + + +# Parent level APIView (registered on parent_bp) +@parent_api_view.route("/books") +class ParentBookListAPIView: + @parent_api_view.doc(summary="get parent book list") + def get(self): + return {"level": "parent", "books": []} + + @parent_api_view.doc(summary="create parent book") + def post(self, body: BookBody): + return {"level": "parent", "book": body.model_dump()} + + +@parent_api_view.route("/books/") +class ParentBookAPIView: + @parent_api_view.doc(summary="get parent book") + def get(self, path: BookPath): + return {"level": "parent", "id": path.id} + + @parent_api_view.doc(summary="update parent book") + def put(self, path: BookPath, body: BookBody): + return {"level": "parent", "id": path.id, "book": body.model_dump()} + + @parent_api_view.doc(summary="delete parent book") + def delete(self, path: BookPath): + return {"level": "parent", "deleted": path.id} + + +# Custom error attached to the parent blueprint +class ParentBlueprintError(Exception): + pass + + +@parent_bp.errorhandler(ParentBlueprintError) +def handle_parent_blueprint_error(e: ParentBlueprintError): + return {"error": str(e)}, 418 + + +# Route on the CHILD APIView that always raises the custom error (no child handler) +@child_api_view.route("/boom") +class ChildBoomAPIView: + @child_api_view.doc(summary="trigger parent blueprint error from child") + def get(self): + raise ParentBlueprintError("broken") + + +# Child level APIView (registered on child_bp) +@child_api_view.route("/books") +class ChildBookListAPIView: + @child_api_view.doc(summary="get child book list") + def get(self): + return {"level": "child", "books": []} + + @child_api_view.doc(summary="create child book") + def post(self, body: BookBody): + return {"level": "child", "book": body.model_dump()} + + +@child_api_view.route("/books/") +class ChildBookAPIView: + @child_api_view.doc(summary="get child book") + def get(self, path: BookPath): + return {"level": "child", "id": path.id} + + +# Register APIView with root app (existing functionality) +app.register_api_view(root_api_view) + +# Register APIView with parent blueprint (NEW functionality) +parent_bp.register_api_view(parent_api_view) + +# Register APIView with child blueprint (NEW functionality) +child_bp.register_api_view(child_api_view) + +# Register child blueprint with parent blueprint +parent_bp.register_api(child_bp) + +# Register parent blueprint with app +app.register_api(parent_bp) + + +@pytest.fixture +def client(): + return app.test_client() + + +def test_openapi_paths(): + """Test that paths exist in OpenAPI spec at all three levels""" + spec = app.api_doc + + # Root level paths (registered on app) + assert "/root/books" in spec["paths"] + assert "/root/books/{id}" in spec["paths"] + + # Parent level paths (registered on parent blueprint) + assert "/parent/v1/books" in spec["paths"] + assert "/parent/v1/books/{id}" in spec["paths"] + + # Child level paths (registered on child blueprint, nested under parent) + assert "/parent/child/v2/books" in spec["paths"] + assert "/parent/child/v2/books/{id}" in spec["paths"] + + +def test_root_level_get_list(client): + """Test GET request to root APIView endpoint""" + resp = client.get("/root/books") + assert resp.status_code == 200 + assert resp.json == {"level": "root", "books": []} + + +def test_root_level_post(client): + """Test POST request to root APIView endpoint""" + resp = client.post("/root/books", json={"title": "Root Book"}) + assert resp.status_code == 200 + assert resp.json["level"] == "root" + + +def test_root_level_get_detail(client): + """Test GET request with path parameter to root APIView""" + resp = client.get("/root/books/99") + assert resp.status_code == 200 + assert resp.json == {"level": "root", "id": 99} + + +def test_parent_level_get_list(client): + """Test GET request to parent blueprint APIView endpoint""" + resp = client.get("/parent/v1/books") + assert resp.status_code == 200 + assert resp.json == {"level": "parent", "books": []} + + +def test_parent_level_post(client): + """Test POST request to parent blueprint APIView endpoint""" + resp = client.post("/parent/v1/books", json={"title": "Parent Book"}) + assert resp.status_code == 200 + assert resp.json["level"] == "parent" + + +def test_parent_level_get_detail(client): + """Test GET request with path parameter to parent blueprint APIView""" + resp = client.get("/parent/v1/books/123") + assert resp.status_code == 200 + assert resp.json == {"level": "parent", "id": 123} + + +def test_parent_level_put(client): + """Test PUT request to parent blueprint APIView endpoint""" + resp = client.put("/parent/v1/books/123", json={"title": "Updated Parent Book"}) + assert resp.status_code == 200 + assert resp.json["level"] == "parent" + assert resp.json["id"] == 123 + + +def test_parent_level_delete(client): + """Test DELETE request to parent blueprint APIView endpoint""" + resp = client.delete("/parent/v1/books/123") + assert resp.status_code == 200 + assert resp.json == {"level": "parent", "deleted": 123} + + +def test_child_error_bubbles_to_parent_handler(client): + """Child-raised error should be handled by the parent blueprint error handler""" + resp = client.get("/parent/child/v2/boom") + assert resp.status_code == 418 + assert resp.json == {"error": "broken"} + + +def test_child_level_get_list(client): + """Test GET request to child blueprint APIView endpoint""" + resp = client.get("/parent/child/v2/books") + assert resp.status_code == 200 + assert resp.json == {"level": "child", "books": []} + + +def test_child_level_post(client): + """Test POST request to child blueprint APIView endpoint""" + resp = client.post("/parent/child/v2/books", json={"title": "Child Book"}) + assert resp.status_code == 200 + assert resp.json["level"] == "child" + + +def test_child_level_get_detail(client): + """Test GET request with path parameter to child blueprint APIView""" + resp = client.get("/parent/child/v2/books/456") + assert resp.status_code == 200 + assert resp.json == {"level": "child", "id": 456} + + +def test_tags_merged_from_all_levels(): + """Test that tags from root, parent, and child APIViews are all merged""" + tag_names = [tag.name for tag in app.tags] + assert "root-books" in tag_names + assert "parent-books" in tag_names + assert "child-books" in tag_names + + +def test_blueprint_paths_structure(): + """Test that blueprint paths are correctly structured at parent and child levels""" + # Child blueprint should have its own url_prefix applied + child_paths = list(child_bp.paths.keys()) + assert any("/child/v2/books" in path for path in child_paths) + + # Parent blueprint should have paths from both itself and nested child + parent_paths = list(parent_bp.paths.keys()) + # Parent's own APIView paths + assert any("/parent/v1/books" in path for path in parent_paths) + # Nested child paths with parent prefix + assert any("/parent/child/v2/books" in path for path in parent_paths) + + +# Idempotency test - same pattern as test_api_view.py and test_api_blueprint.py +# Create a view that will be registered multiple times +idempotent_view = APIView(url_prefix="/v1", view_tags=[Tag(name="test")]) + + +@idempotent_view.route("/items") +class ItemView: + @idempotent_view.doc(summary="get items") + def get(self): + return {"items": []} + + +def create_blueprint_with_view(): + app = OpenAPI(__name__, info=info) + bp = APIBlueprint("test_bp", __name__, url_prefix="/test") + bp.register_api_view(idempotent_view, url_prefix="/v2") + app.register_api(bp) + + +# Invoke twice to ensure that call is idempotent +create_blueprint_with_view() +create_blueprint_with_view() + + +def test_register_api_view_idempotency(): + """Test that registering same APIView multiple times updates paths correctly""" + assert list(idempotent_view.paths.keys()) == ["/v2/items"]