diff --git a/.github/pull_request_template.md b/.github/pull_request_template.md index 7e288c54..5599b01e 100644 --- a/.github/pull_request_template.md +++ b/.github/pull_request_template.md @@ -2,5 +2,5 @@ Checklist: - [ ] Run `pytest tests` and no failed. - [ ] Run `flake8 flask_openapi3 tests examples` and no failed. -- [ ] Run `mkdocs serve` and no failed. - [ ] Run `mypy flask_openapi3` and no failed. +- [ ] Run `mkdocs serve` and no failed. diff --git a/docs/Example.md b/docs/Example.md index 23a8d24c..237de032 100644 --- a/docs/Example.md +++ b/docs/Example.md @@ -127,8 +127,7 @@ class BookResponse(BaseModel): tags=[book_tag], summary='new summary', description='new description', - responses={"200": BookResponse}, - extra_responses={"200": {"content": {"text/csv": {"schema": {"type": "string"}}}}}, + responses={"200": BookResponse, "201": {"content": {"text/csv": {"schema": {"type": "string"}}}}}, security=security ) def get_book(path: BookPath): @@ -189,7 +188,6 @@ from typing import Optional from pydantic import BaseModel, Field from flask_openapi3 import APIBlueprint, OpenAPI -from flask_openapi3 import HTTPBearer from flask_openapi3 import Tag, Info info = Info(title='book API', version='1.0.0') @@ -238,7 +236,7 @@ def get_book(): return {"code": 0, "message": "ok"} -@api.post('/book', extra_responses={"200": {"content": {"text/csv": {"schema": {"type": "string"}}}}}) +@api.post('/book', responses={"201": {"content": {"text/csv": {"schema": {"type": "string"}}}}}) def create_book(body: BookBody): assert body.age == 3 return {"code": 0, "message": "ok"} diff --git a/docs/Tutorial/Openapi_extra.md b/docs/Tutorial/Openapi_extra.md new file mode 100644 index 00000000..b8a98a47 --- /dev/null +++ b/docs/Tutorial/Openapi_extra.md @@ -0,0 +1,105 @@ +*New in v2.4.0* + +The [BaseModel](https://docs.pydantic.dev/latest/usage/models/) in [Pydantic](https://github.com/pydantic/pydantic) +supports some custom configurations([Model Config](https://docs.pydantic.dev/latest/usage/model_config/)), +so we can use the `openapi_extra` to extend OpenAPI Specification. + +The `openapi_extra` will be merged with the automatically generated OpenAPI schema. + +## form + +```python +class UploadFilesForm(BaseModel): + file: FileStorage + str_list: List[str] + + class Config: + openapi_extra = { + # "example": {"a": 123}, + "examples": { + "Example 01": { + "summary": "An example", + "value": { + "file": "Example-01.jpg", + "str_list": ["a", "b", "c"] + } + }, + "Example 02": { + "summary": "Another example", + "value": { + "str_list": ["1", "2", "3"] + } + } + } + } +``` + +Effect in Redoc: + +![](../assets/Snipaste_2023-06-02_11-05-11.png) + +## body + +```python +class BookBody(BaseModel): + age: int + author: str + + class Config: + openapi_extra = { + "description": "This is post RequestBody", + "example": {"age": 12, "author": "author1"}, + "examples": { + "example1": { + "summary": "example summary1", + "description": "example description1", + "value": { + "age": 24, + "author": "author2" + } + }, + "example2": { + "summary": "example summary2", + "description": "example description2", + "value": { + "age": 48, + "author": "author3" + } + } + + }} +``` + +Effect in swagger: + +![](../assets/Snipaste_2023-06-02_11-06-59.png) + +## responses + +```python +class Message(BaseModel): + message: str = Field(..., description="The message") + + class Config: + openapi_extra = { + # "example": {"message": "aaa"}, + "examples": { + "example1": { + "summary": "example1 summary", + "value": { + "message": "bbb" + } + }, + "example2": { + "summary": "example2 summary", + "value": { + "message": "ccc" + } + } + } + } +``` + +Effect in swagger: + +![](../assets/Snipaste_2023-06-02_11-08-40.png) \ No newline at end of file diff --git a/docs/Tutorial/Operation.md b/docs/Tutorial/Operation.md index 421af14d..f8c4b113 100644 --- a/docs/Tutorial/Operation.md +++ b/docs/Tutorial/Operation.md @@ -127,6 +127,12 @@ def create_book(body: BookBody): ## extra_form +*new in v2.4.0* + +!!! Deprecated-Warning warning + + `extra_form` will be deprecated in v3.x, please use `openapi_extra` instead. + *new in v2.1.0* Extra form information can be provided using `extra_form` as in the following sample: @@ -148,6 +154,12 @@ def create_book(body: BookForm): ## extra_body +*new in v2.4.0* + +!!! Deprecated-Warning warning + + `extra_body` will be deprecated in v3.x, please use `openapi_extra` instead. + *new in v2.1.0* Extra body information can be provided using `extra_body` as in the following sample: diff --git a/docs/Tutorial/Response.md b/docs/Tutorial/Response.md index 0849a012..5fc0b9f9 100644 --- a/docs/Tutorial/Response.md +++ b/docs/Tutorial/Response.md @@ -15,7 +15,13 @@ class BookResponse(BaseModel): data: BookBodyWithID -@app.get('/book/', tags=[book_tag], responses={"200": BookResponse}) +@app.get('/book/', + tags=[book_tag], + responses={ + "200": BookResponse, + # Version 2.4.0 starts supporting response for dictionary types + "201": {"content": {"text/csv": {"schema": {"type": "string"}}}} + }) def get_book(path: BookPath, query: BookBody): """get a book get book by id, age or author @@ -26,9 +32,14 @@ def get_book(path: BookPath, query: BookBody): ![image-20210526104627124](../assets/image-20210526104627124.png) - ## extra_responses +*New in v2.4.0* + +!!! Deprecated-Warning warning + + `extra_responses` have been merged into the `responses`, and `extra_responses` will be deprecated in v3.x. + *New in v1.0.0* You can pass to your path operation decorators a parameter `extra_responses`. @@ -43,14 +54,14 @@ Like this: '/book/', tags=[book_tag], responses={"200": BookResponse}, - extra_responses={"200": {"content": {"text/csv": {"schema": {"type": "string"}}}}}, + extra_responses={"201": {"content": {"text/csv": {"schema": {"type": "string"}}}}}, security=security ) def get_book(path: BookPath): ... -@api.post('/book', extra_responses={"200": {"content": {"text/csv": {"schema": {"type": "string"}}}}}) +@api.post('/book', extra_responses={"201": {"content": {"text/csv": {"schema": {"type": "string"}}}}}) def create_book(body: BookBody): ... ``` diff --git a/docs/Tutorial/Specification.md b/docs/Tutorial/Specification.md index 65457fdd..80de9525 100644 --- a/docs/Tutorial/Specification.md +++ b/docs/Tutorial/Specification.md @@ -252,6 +252,14 @@ def endpoint(): You can also use [responses ](./Response.md#responses) and [extra_responses](./Response.md#extra_responses) in your api. +*New in v2.4.0* + +!!! Deprecated-Warning warning + + `extra_responses` have been merged into the `responses`, and `extra_responses` will be deprecated in v3.x. + + + ## doc_ui You can pass `doc_ui=False` to disable the `OpenAPI spec` when init `OpenAPI `. diff --git a/docs/assets/Snipaste_2023-06-02_11-05-11.png b/docs/assets/Snipaste_2023-06-02_11-05-11.png new file mode 100644 index 00000000..a171bcc1 Binary files /dev/null and b/docs/assets/Snipaste_2023-06-02_11-05-11.png differ diff --git a/docs/assets/Snipaste_2023-06-02_11-06-59.png b/docs/assets/Snipaste_2023-06-02_11-06-59.png new file mode 100644 index 00000000..5c698f4b Binary files /dev/null and b/docs/assets/Snipaste_2023-06-02_11-06-59.png differ diff --git a/docs/assets/Snipaste_2023-06-02_11-08-40.png b/docs/assets/Snipaste_2023-06-02_11-08-40.png new file mode 100644 index 00000000..a6aa0aa2 Binary files /dev/null and b/docs/assets/Snipaste_2023-06-02_11-08-40.png differ diff --git a/examples/api_blueprint_demo.py b/examples/api_blueprint_demo.py index 378dcfa9..8d837e71 100644 --- a/examples/api_blueprint_demo.py +++ b/examples/api_blueprint_demo.py @@ -55,7 +55,7 @@ def get_book(): return {"code": 0, "message": "ok"} -@api.post('/book', extra_responses={"200": {"content": {"text/csv": {"schema": {"type": "string"}}}}}) +@api.post('/book', responses={"200": {"content": {"text/csv": {"schema": {"type": "string"}}}}}) def create_book(body: BookBody): assert body.age == 3 return {"code": 0, "message": "ok"} diff --git a/examples/openapi_extra.py b/examples/openapi_extra.py new file mode 100644 index 00000000..98992084 --- /dev/null +++ b/examples/openapi_extra.py @@ -0,0 +1,81 @@ +# -*- coding: utf-8 -*- +# @Author : llc +# @Time : 2023/6/1 15:04 +from typing import List + +from pydantic import BaseModel + +from flask_openapi3 import OpenAPI, FileStorage + +app = OpenAPI(__name__) + + +class UploadFilesForm(BaseModel): + file: FileStorage + str_list: List[str] + + class Config: + openapi_extra = { + # "example": {"a": 123}, + "examples": { + "Example 01": { + "summary": "An example", + "value": { + "file": "Example-01.jpg", + "str_list": ["a", "b", "c"] + } + }, + "Example 02": { + "summary": "Another example", + "value": { + "str_list": ["1", "2", "3"] + } + } + } + } + + +class BookBody(BaseModel): + age: int + author: str + + class Config: + openapi_extra = { + "description": "This is post RequestBody", + "example": {"age": 12, "author": "author1"}, + "examples": { + "example1": { + "summary": "example summary1", + "description": "example description1", + "value": { + "age": 24, + "author": "author2" + } + }, + "example2": { + "summary": "example summary2", + "description": "example description2", + "value": { + "age": 48, + "author": "author3" + } + } + + }} + + +@app.post('/upload/files') +def upload_files(form: UploadFilesForm): + print(form.file) + print(form.str_list) + return {"code": 0, "message": "ok"} + + +@app.post('/book', ) +def create_book(body: BookBody): + print(body) + return {"code": 0, "message": "ok"} + + +if __name__ == "__main__": + app.run(debug=True) diff --git a/examples/response_demo.py b/examples/response_demo.py index 04de6f89..4ef40838 100644 --- a/examples/response_demo.py +++ b/examples/response_demo.py @@ -1,5 +1,5 @@ # -*- coding: utf-8 -*- -# @Author : [martinatseequent](https://github.com/martinatseequent) +# @Author : llc # @Time : 2021/6/22 9:32 import json @@ -11,7 +11,7 @@ from flask_openapi3 import Info from flask_openapi3 import OpenAPI, APIBlueprint -app = OpenAPI(__name__, info=Info(title="Hello API", version="1.0.0"), ) +app = OpenAPI(__name__, info=Info(title="Hello API", version="1.0.0")) bp = APIBlueprint("Hello BP", __name__) @@ -23,8 +23,28 @@ class HelloPath(BaseModel): class Message(BaseModel): message: str = Field(..., description="The message") - -@bp.get("/hello/", responses={"200": Message}) + class Config: + openapi_extra = { + # "example": {"message": "aaa"}, + "examples": { + "example1": { + "summary": "example1 summary", + "value": { + "message": "bbb" + } + }, + "example2": { + "summary": "example2 summary", + "value": { + "message": "ccc" + } + } + } + } + + +@bp.get("/hello/", + responses={"200": Message, "201": {"content": {"text/csv": {"schema": {"type": "string"}}}}}) def hello(path: HelloPath): message = {"message": f"""Hello {path.name}!"""} diff --git a/examples/rest_demo.py b/examples/rest_demo.py index 48673690..905f4eba 100644 --- a/examples/rest_demo.py +++ b/examples/rest_demo.py @@ -6,7 +6,7 @@ from pydantic import BaseModel, Field -from flask_openapi3 import ExternalDocumentation, ExtraRequestBody, Example +from flask_openapi3 import ExternalDocumentation from flask_openapi3 import Info, Tag, Server from flask_openapi3 import OpenAPI @@ -95,9 +95,7 @@ class BookResponse(BaseModel): external_docs=ExternalDocumentation( url="https://www.openapis.org/", description="Something great got better, get excited!"), - responses={"200": BookResponse}, - extra_responses={"200": {"content": {"text/csv": {"schema": {"type": "string"}}}}}, security=security, servers=[Server(url="https://www.openapis.org/", description="openapi")] ) @@ -128,32 +126,7 @@ def get_books(query: BookQuery): } -extra_body = ExtraRequestBody( - description="This is post RequestBody", - required=True, - example="ttt", - examples={ - "example1": Example( - summary="example summary1", - description="example description1", - value={ - "age": 24, - "author": "author1" - } - ), - "example2": Example( - summary="example summary2", - description="example description2", - value={ - "age": 48, - "author": "author2" - } - ) - } -) - - -@app.post('/book', tags=[book_tag], responses={"200": BookResponse}, extra_body=extra_body) +@app.post('/book', tags=[book_tag], responses={"200": BookResponse}) def create_book(body: BookBody): print(body) return {"code": 0, "message": "ok"}, HTTPStatus.OK diff --git a/examples/upload_file_demo.py b/examples/upload_file_demo.py index 3c04f8b4..c4ab2d63 100644 --- a/examples/upload_file_demo.py +++ b/examples/upload_file_demo.py @@ -6,8 +6,7 @@ from pydantic import BaseModel, Field -from flask_openapi3 import OpenAPI, FileStorage, ExtraRequestBody -from flask_openapi3 import Encoding +from flask_openapi3 import OpenAPI, FileStorage app = OpenAPI(__name__) @@ -23,14 +22,6 @@ class UploadFilesForm(BaseModel): int_list: List[int] -extra_form = ExtraRequestBody( - description="This is form RequestBody", - required=True, - # replace style (default to form) - encoding={"str_list": Encoding(style="simple")} -) - - @app.post('/upload/file') def upload_file(form: UploadFileForm): print(form.file.filename) @@ -39,7 +30,7 @@ def upload_file(form: UploadFileForm): return {"code": 0, "message": "ok"} -@app.post('/upload/files', extra_form=extra_form) +@app.post('/upload/files') def upload_files(form: UploadFilesForm): print(form.files) print(form.str_list) diff --git a/flask_openapi3/blueprint.py b/flask_openapi3/blueprint.py index 759060d1..42dfcccf 100644 --- a/flask_openapi3/blueprint.py +++ b/flask_openapi3/blueprint.py @@ -3,7 +3,7 @@ # @Time : 2022/4/1 16:54 import re from copy import deepcopy -from typing import Optional, List, Dict, Any, Type, Callable, Tuple +from typing import Optional, List, Dict, Any, Type, Callable, Tuple, Union from flask import Blueprint from pydantic import BaseModel @@ -25,7 +25,7 @@ def __init__( *, abp_tags: Optional[List[Tag]] = None, abp_security: Optional[List[Dict[str, List[str]]]] = None, - abp_responses: Optional[Dict[str, Optional[Type[BaseModel]]]] = None, + abp_responses: Optional[Dict[str, Union[Type[BaseModel], Dict[Any, Any], None]]] = None, doc_ui: bool = True, operation_id_callback: Callable = get_operation_id_for_path, **kwargs: Any @@ -35,49 +35,63 @@ def __init__( Arguments: name: The name of the blueprint. It Will be prepared to each endpoint name. - import_name: The name of the blueprint package, usually - ``__name__``. This helps locate the ``root_path`` for the blueprint. - abp_tags: APIBlueprint tags for every api - abp_security: APIBlueprint security for every api - abp_responses: APIBlueprint response model - doc_ui: Add openapi document UI(swagger, rapidoc and redoc). Defaults to True. + import_name: The name of the blueprint package, usually ``__name__``. + This helps locate the ``root_path`` for the blueprint. + abp_tags: APIBlueprint tags for every API. + abp_security: APIBlueprint security for every API. + abp_responses: API responses should be either a subclass of BaseModel, a dictionary, or None. + doc_ui: Enable OpenAPI document UI (Swagger UI, Redoc and Rapidoc). Defaults to True. operation_id_callback: Callback function for custom operation_id generation. - Receives name (str), path (str) and method (str) parameters. - Defaults to `get_operation_id_for_path` from utils + Receives name (str), path (str) and method (str) parameters. + Defaults to `get_operation_id_for_path` from utils **kwargs: Flask Blueprint kwargs """ super(APIBlueprint, self).__init__(name, import_name, **kwargs) + + # Initialize instance variables self.paths: Dict = dict() self.components_schemas: Dict = dict() self.tags: List[Tag] = [] self.tag_names: List[str] = [] + # Set values from arguments or default values self.abp_tags = abp_tags or [] self.abp_security = abp_security or [] self.abp_responses = abp_responses or {} self.doc_ui = doc_ui + + # Set the operation ID callback function self.operation_id_callback: Callable = operation_id_callback def register_api(self, api: "APIBlueprint") -> None: """Register a nested APIBlueprint""" + + # Check if the APIBlueprint is being registered on itself if api is self: raise ValueError("Cannot register a api blueprint on itself") + # Merge tags from the nested APIBlueprint for tag in api.tags: if tag.name not in self.tag_names: self.tags.append(tag) + # Merge paths from the nested APIBlueprint for path_url, path_item in api.paths.items(): trail_slash = path_url.endswith("/") - # merge url_prefix and new api blueprint path url + + # Merge url_prefix and new API blueprint path url uri = self.url_prefix.rstrip("/") + "/" + path_url.lstrip("/") if self.url_prefix else path_url - # strip the right slash + + # Strip the right slash if not trail_slash: uri = uri.rstrip("/") + self.paths[uri] = path_item + # Merge component schemas from the nested APIBlueprint self.components_schemas.update(**api.components_schemas) + # Register the nested APIBlueprint as a blueprint self.register_blueprint(api) def _do_decorator( @@ -92,7 +106,7 @@ def _do_decorator( operation_id: Optional[str] = None, extra_form: Optional[ExtraRequestBody] = None, extra_body: Optional[ExtraRequestBody] = None, - responses: Optional[Dict[str, Optional[Type[BaseModel]]]] = None, + responses: Optional[Dict[str, Union[Type[BaseModel], Dict[Any, Any], None]]] = None, extra_responses: Optional[Dict[str, Dict]] = None, deprecated: Optional[bool] = None, security: Optional[List[Dict[str, List[Any]]]] = None, @@ -102,7 +116,8 @@ def _do_decorator( method: str = HTTPMethod.GET ) -> Tuple[Type[BaseModel], Type[BaseModel], Type[BaseModel], Type[BaseModel], Type[BaseModel], Type[BaseModel]]: """ - Collect openapi specification information + Collects OpenAPI specification information for Flask routes and view functions. + Arguments: rule: Flask route func: Flask view_func @@ -113,7 +128,7 @@ def _do_decorator( operation_id: Unique string used to identify the operation. extra_form: Extra information describing the request body(application/form). extra_body: Extra information describing the request body(application/json). - responses: response's model must be pydantic BaseModel. + responses: API responses should be either a subclass of BaseModel, a dictionary, or None. extra_responses: Extra information for responses. deprecated: Declares this operation to be deprecated. security: A declaration of which security mechanisms can be used for this operation. @@ -126,34 +141,34 @@ def _do_decorator( responses = {} if extra_responses is None: extra_responses = {} - # global response combine api responses + # Global response: combine API responses combine_responses = deepcopy(self.abp_responses) combine_responses.update(**responses) - # create operation + # Create operation operation = get_operation( func, summary=summary, description=description, openapi_extensions=openapi_extensions ) - # set external docs + # Set external docs operation.externalDocs = external_docs # Unique string used to identify the operation. operation.operationId = operation_id or self.operation_id_callback( name=func.__name__, path=rule, method=method ) - # only set `deprecated` if True otherwise leave it as None + # Only set `deprecated` if True, otherwise leave it as None operation.deprecated = deprecated - # add security + # Add security if security is None: security = [] operation.security = security + self.abp_security or None - # add servers + # Add servers operation.servers = servers - # store tags + # Store tags tags = tags + self.abp_tags if tags else self.abp_tags parse_and_store_tags(tags, self.tags, self.tag_names, operation) - # parse parameters + # Parse parameters header, cookie, path, query, form, body = parse_parameters( func, extra_form=extra_form, @@ -161,19 +176,19 @@ def _do_decorator( components_schemas=self.components_schemas, operation=operation ) - # parse response + # Parse response get_responses(combine_responses, extra_responses, self.components_schemas, operation) - # /pet/ --> /pet/{petId} + # Convert route parameter format from /pet/ to /pet/{petId} uri = re.sub(r"<([^<:]+:)?", "{", rule).replace(">", "}") trail_slash = uri.endswith("/") - # merge url_prefix and uri + # Merge url_prefix and uri uri = self.url_prefix.rstrip("/") + "/" + uri.lstrip("/") if self.url_prefix else uri if not trail_slash: uri = uri.rstrip("/") - # parse method + # Parse method parse_method(uri, method, self.paths, operation) return header, cookie, path, query, form, body else: - # parse parameters + # Parse parameters header, cookie, path, query, form, body = parse_parameters(func, doc_ui=False) return header, cookie, path, query, form, body diff --git a/flask_openapi3/commands.py b/flask_openapi3/commands.py index d6eb6247..ad4b7dd8 100644 --- a/flask_openapi3/commands.py +++ b/flask_openapi3/commands.py @@ -10,28 +10,34 @@ from .markdown import openapi_to_markdown -@click.command(name='openapi') -@click.option('--output', '-o', type=click.Path(), help='The output file path.') -@click.option('--format', '-f', type=click.Choice(['json', 'yaml', 'markdown']), help='The output file format.') -@click.option('--indent', '-i', type=int, help='The indentation for JSON dumps.') -@click.option('--ensure_ascii', '-a', is_flag=True, help='Ensure ASCII characters or not. Defaults to False.') +@click.command(name="openapi") +@click.option("--output", "-o", type=click.Path(), help="The output file path.") +@click.option("--format", "-f", type=click.Choice(["json", "yaml", "markdown"]), help="The output file format.") +@click.option("--indent", "-i", type=int, help="The indentation for JSON dumps.") +@click.option("--ensure_ascii", "-a", is_flag=True, help="Ensure ASCII characters or not. Defaults to False.") @with_appcontext def openapi_command(output, format, indent, ensure_ascii): """Export the OpenAPI Specification to console or a file""" - if hasattr(current_app, 'api_doc'): + + # Check if the current app has an api_doc attribute + if hasattr(current_app, "api_doc"): obj = current_app.api_doc - if format == 'yaml': + + # Generate the OpenAPI Specification based on the specified format + if format == "yaml": try: import yaml # type: ignore except ImportError: raise ImportError("pyyaml must be installed.") openapi = yaml.safe_dump(obj, allow_unicode=True) - elif format == 'markdown': + elif format == "markdown": openapi = openapi_to_markdown(obj) else: openapi = json.dumps(obj, indent=indent, ensure_ascii=ensure_ascii) + + # Save the OpenAPI Specification to a file if the output path is provided if output: - with open(output, 'w', encoding='utf8') as f: + with open(output, "w", encoding="utf8") as f: f.write(openapi) click.echo(f"Saved to {output}.") else: diff --git a/flask_openapi3/models/common.py b/flask_openapi3/models/common.py index 0f998705..3dff7441 100644 --- a/flask_openapi3/models/common.py +++ b/flask_openapi3/models/common.py @@ -1,10 +1,13 @@ # -*- coding: utf-8 -*- # @Author : llc # @Time : 2021/4/30 11:46 +import warnings from typing import Optional, List, Any, Union, Dict from pydantic import BaseModel, Field +warnings.simplefilter("once") + class ExternalDocumentation(BaseModel): description: Optional[str] = None @@ -129,3 +132,7 @@ class ExtraRequestBody(BaseModel): example: Optional[Any] = None examples: Optional[Dict[str, Union[Example, Reference]]] = None encoding: Optional[Dict[str, Encoding]] = None + + def __new__(cls, *args, **kwargs): + warnings.warn(f"""\n{cls.__name__} will be deprecated in v3.x.""", DeprecationWarning) + return super().__new__(cls) diff --git a/flask_openapi3/openapi.py b/flask_openapi3/openapi.py index 752d70ec..25279ef2 100644 --- a/flask_openapi3/openapi.py +++ b/flask_openapi3/openapi.py @@ -32,7 +32,7 @@ def __init__( info: Optional[Info] = None, security_schemes: Optional[Dict[str, Union[SecurityScheme, Dict[str, Any]]]] = None, oauth_config: Optional[OAuthConfig] = None, - responses: Optional[Dict[str, Optional[Type[BaseModel]]]] = None, + responses: Optional[Dict[str, Union[Type[BaseModel], Dict[Any, Any], None]]] = None, doc_ui: bool = True, doc_expansion: str = "list", doc_prefix: str = "/openapi", @@ -48,62 +48,83 @@ def __init__( **kwargs: Any ) -> None: """ - Based on Flask. Provide REST api, swagger-ui and redoc. + OpenAPI class that provides REST API functionality along with Swagger UI and Redoc. Arguments: - import_name: Just flask import_name - info: See https://spec.openapis.org/oas/v3.0.3#info-object - security_schemes: See https://spec.openapis.org/oas/v3.0.3#security-scheme-object - oauth_config: OAuth 2.0 configuration, - see https://github.com/swagger-api/swagger-ui/blob/master/docs/usage/oauth2.md - responses: OpenAPI response model - doc_ui: Add openapi document UI(swagger and redoc). Defaults to True. - doc_expansion: String=["list"*, "full", "none"]. - Controls the default expansion setting for the operations and tags. - It can be "list" (expands only the tags), - "full" (expands the tags and operations) or "none" (expands nothing). - see https://github.com/swagger-api/swagger-ui/blob/master/docs/usage/configuration.md + import_name: The import name for the Flask application. + info: Information about the API (title, version, etc.). + See https://spec.openapis.org/oas/v3.0.3#info-object. + security_schemes: Security schemes for the API. + See https://spec.openapis.org/oas/v3.0.3#security-scheme-object. + oauth_config: OAuth 2.0 configuration for authentication. + See https://github.com/swagger-api/swagger-ui/blob/master/docs/usage/oauth2.md. + responses: API responses should be either a subclass of BaseModel, a dictionary, or None. + doc_ui: Enable OpenAPI document UI (Swagger UI and Redoc). Defaults to True. + doc_expansion: Default expansion setting for operations and tags ("list", "full", or "none"). + See https://github.com/swagger-api/swagger-ui/blob/master/docs/usage/configuration.md. doc_prefix: URL prefix used for OpenAPI document and UI. Defaults to "/openapi". - api_doc_url: The OpenAPI Spec documentation. Defaults to "/openapi.json". - swagger_url: The Swagger UI documentation. Defaults to `/swagger`. - redoc_url: The Redoc UI documentation. Defaults to `/redoc`. - rapidoc_url: The RapiDoc UI documentation. Defaults to `/rapidoc`. - ui_templates: Custom UI templates, which are used to overwrite or add UI documents. - servers: An array of Server Objects, which provide connectivity information to a target server. - external_docs: Allows referencing an external resource for extended documentation. - See: https://spec.openapis.org/oas/v3.0.3#external-documentation-object - operation_id_callback: Callback function for custom operation_id generation. - Receives name (str), path (str) and method (str) parameters. - Default to `get_operation_id_for_path` from utils - openapi_extensions: Allows extensions to the OpenAPI Schema. - See https://spec.openapis.org/oas/v3.0.3#specification-extensions - **kwargs: Flask kwargs + api_doc_url: URL for accessing the OpenAPI specification document in JSON format. + Defaults to "/openapi.json". + swagger_url: URL for accessing the Swagger UI documentation. Defaults to "/swagger". + redoc_url: URL for accessing the Redoc UI documentation. Defaults to "/redoc". + rapidoc_url: URL for accessing the RapiDoc UI documentation. Defaults to "/rapidoc". + ui_templates: Custom UI templates to override or add UI documents. + servers: An array of Server objects providing connectivity information to a target server. + external_docs: External documentation for the API. + See: https://spec.openapis.org/oas/v3.0.3#external-documentation-object. + operation_id_callback: Callback function for custom operation ID generation. + Receives name (str), path (str), and method (str) parameters. + Defaults to `get_operation_id_for_path` from utils. + openapi_extensions: Extensions to the OpenAPI Schema. + See https://spec.openapis.org/oas/v3.0.3#specification-extensions. + **kwargs: Additional kwargs to be passed to Flask. """ super(OpenAPI, self).__init__(import_name, **kwargs) + + # Set OpenAPI version and API information self.openapi_version = "3.0.3" self.info = info or Info(title="OpenAPI", version="1.0.0") + + # Set security schemes, responses, paths and components self.security_schemes = security_schemes self.responses = responses or {} self.paths: Dict = dict() self.components_schemas: Dict = dict() self.components = Components() + + # Initialize lists for tags and tag names self.tags: List[Tag] = [] self.tag_names: List[str] = [] + + # Set URL prefixes and endpoints self.doc_prefix = doc_prefix self.api_doc_url = api_doc_url self.swagger_url = swagger_url self.redoc_url = redoc_url self.rapidoc_url = rapidoc_url + + # Set OAuth configuration and documentation expansion self.oauth_config = oauth_config self.doc_expansion = doc_expansion + + # Set UI templates for customization self.ui_templates = ui_templates or dict() + + # Set servers and external documentation self.severs = servers self.external_docs = external_docs + + # Set the operation ID callback function self.operation_id_callback: Callable = operation_id_callback + + # Set OpenAPI extensions self.openapi_extensions = openapi_extensions or dict() + + # Initialize the OpenAPI documentation UI if doc_ui: self._init_doc() - # add openapi command + + # Add the OpenAPI command self.cli.add_command(openapi_command) def _init_doc(self) -> None: @@ -114,6 +135,7 @@ def _init_doc(self) -> None: template_folder = os.path.join(_here, "templates") static_folder = os.path.join(template_folder, "static") + # Create the blueprint for OpenAPI documentation blueprint = Blueprint( "openapi", __name__, @@ -121,24 +143,28 @@ def _init_doc(self) -> None: template_folder=template_folder, static_folder=static_folder ) + + # Add the API documentation URL rule blueprint.add_url_rule( rule=self.api_doc_url, endpoint="api_doc", view_func=lambda: self.api_doc ) + + # Combine built-in templates and user-defined templates builtins_templates = { self.swagger_url.strip("/"): swagger_html_string, self.redoc_url.strip("/"): redoc_html_string, - self.rapidoc_url.strip("/"): rapidoc_html_string + self.rapidoc_url.strip("/"): rapidoc_html_string, + **self.ui_templates } - # update builtins_templates - builtins_templates.update(**self.ui_templates) - # iter builtins_templates + + # Add URL rules for the templates for key, value in builtins_templates.items(): blueprint.add_url_rule( rule=f"/{key}", endpoint=key, - # pass default value to source + # Pass default value to source view_func=lambda source=value: render_template_string( source, api_doc_url=self.api_doc_url.lstrip("/"), @@ -147,7 +173,8 @@ def _init_doc(self) -> None: oauth_config=self.oauth_config.dict() if self.oauth_config else None ) ) - # home page + + # Add URL rule for the home page blueprint.add_url_rule( rule="/", endpoint="openapi", @@ -158,36 +185,66 @@ def _init_doc(self) -> None: rapidoc_url=self.rapidoc_url.lstrip("/") ) ) + + # Register the blueprint with the Flask application self.register_blueprint(blueprint) @property def api_doc(self) -> Dict: - """Generate Specification json""" + """ + Generate the OpenAPI specification JSON. + + Returns: + The OpenAPI specification JSON as a dictionary. + + """ spec = APISpec( openapi=self.openapi_version, info=self.info, servers=self.severs, externalDocs=self.external_docs ) + # Set tags spec.tags = self.tags or None + + # Set paths spec.paths = self.paths + + # Set components self.components.schemas = self.components_schemas self.components.securitySchemes = self.security_schemes spec.components = self.components + # Convert spec to JSON spec_json = json.loads(spec.json(by_alias=True, exclude_none=True)) + + # Update with OpenAPI extensions spec_json.update(**self.openapi_extensions) return spec_json def register_api(self, api: APIBlueprint) -> None: - """Register APIBlueprint""" + """ + Register an APIBlueprint. + + Arguments: + api: The APIBlueprint instance to register. + + """ for tag in api.tags: if tag.name not in self.tag_names: + # Append tag to the list of tags self.tags.append(tag) + # Append tag name to the list of tag names self.tag_names.append(tag.name) + + # Update paths with the APIBlueprint's paths self.paths.update(**api.paths) + + # Update components schemas with the APIBlueprint's components schemas self.components_schemas.update(**api.components_schemas) + + # Register the APIBlueprint with the current instance self.register_blueprint(api) def register_api_view(self, api_view: APIView, view_kwargs: Optional[Dict[Any, Any]] = None) -> None: @@ -195,17 +252,27 @@ def register_api_view(self, api_view: APIView, view_kwargs: Optional[Dict[Any, A Register APIView Arguments: - api_view: APIView - view_kwargs: extra view kwargs + api_view: The APIView instance to register. + view_kwargs: Additional keyword arguments to pass to the API views. """ if view_kwargs is None: view_kwargs = {} + + # Iterate through tags of the APIView for tag in api_view.tags: if tag.name not in self.tag_names: + # Append tag to the list of tags self.tags.append(tag) + # Append tag name to the list of tag names self.tag_names.append(tag.name) + + # Update paths with the APIView's paths self.paths.update(**api_view.paths) + + # Update components schemas with the APIView's components schemas self.components_schemas.update(**api_view.components_schemas) + + # Register the APIView with the current instance api_view.register(self, view_kwargs=view_kwargs) def _do_decorator( @@ -220,7 +287,7 @@ def _do_decorator( operation_id: Optional[str] = None, extra_form: Optional[ExtraRequestBody] = None, extra_body: Optional[ExtraRequestBody] = None, - responses: Optional[Dict[str, Optional[Type[BaseModel]]]] = None, + responses: Optional[Dict[str, Union[Type[BaseModel], Dict[Any, Any], None]]] = None, extra_responses: Optional[Dict[str, Dict]] = None, deprecated: Optional[bool] = None, security: Optional[List[Dict[str, List[Any]]]] = None, @@ -230,10 +297,11 @@ def _do_decorator( method: str = HTTPMethod.GET ) -> Tuple[Type[BaseModel], Type[BaseModel], Type[BaseModel], Type[BaseModel], Type[BaseModel], Type[BaseModel]]: """ - Collect openapi specification information + Collects OpenAPI specification information for Flask routes and view functions. + Arguments: - rule: Flask route - func: Flask view_func + rule: Flask route. + func: Flask view_func. tags: Adds metadata to a single tag. summary: A short summary of what the operation does. description: A verbose explanation of the operation behavior. @@ -241,46 +309,47 @@ def _do_decorator( operation_id: Unique string used to identify the operation. extra_form: Extra information describing the request body(application/form). extra_body: Extra information describing the request body(application/json). - responses: response's model must be pydantic BaseModel. + responses: API responses should be either a subclass of BaseModel, a dictionary, or None. extra_responses: Extra information for responses. deprecated: Declares this operation to be deprecated. security: A declaration of which security mechanisms can be used for this operation. servers: An alternative server array to service this operation. openapi_extensions: Allows extensions to the OpenAPI Schema. - doc_ui: Add openapi document UI(swagger, rapidoc and redoc). Defaults to True. + doc_ui: Add OpenAPI document UI (swagger, rapidoc, and redoc). Defaults to True. + method: HTTP method for the operation. Defaults to GET. """ if doc_ui is True: if responses is None: responses = {} if extra_responses is None: extra_responses = {} - # global response combine api responses + # Global response: combine API responses combine_responses = deepcopy(self.responses) combine_responses.update(**responses) - # create operation + # Create operation operation = get_operation( func, summary=summary, description=description, openapi_extensions=openapi_extensions ) - # set external docs + # Set external docs operation.externalDocs = external_docs # Unique string used to identify the operation. operation.operationId = operation_id or self.operation_id_callback( name=func.__name__, path=rule, method=method ) - # only set `deprecated` if True otherwise leave it as None + # Only set `deprecated` if True, otherwise leave it as None operation.deprecated = deprecated - # add security + # Add security operation.security = security - # add servers + # Add servers operation.servers = servers - # store tags + # Store tags if tags is None: tags = [] parse_and_store_tags(tags, self.tags, self.tag_names, operation) - # parse parameters + # Parse parameters header, cookie, path, query, form, body = parse_parameters( func, extra_form=extra_form, @@ -288,14 +357,14 @@ def _do_decorator( components_schemas=self.components_schemas, operation=operation ) - # parse response + # Parse response get_responses(combine_responses, extra_responses, self.components_schemas, operation) - # /pet/ --> /pet/{petId} + # Convert route parameter format from /pet/ to /pet/{petId} uri = re.sub(r"<([^<:]+:)?", "{", rule).replace(">", "}") - # parse method + # Parse method parse_method(uri, method, self.paths, operation) return header, cookie, path, query, form, body else: - # parse parameters + # Parse parameters header, cookie, path, query, form, body = parse_parameters(func, doc_ui=False) return header, cookie, path, query, form, body diff --git a/flask_openapi3/request.py b/flask_openapi3/request.py index ba4023fd..006578d4 100644 --- a/flask_openapi3/request.py +++ b/flask_openapi3/request.py @@ -11,56 +11,56 @@ from pydantic.error_wrappers import ErrorWrapper -def _do_header(header, request_kwargs): +def _do_header(header, func_kwargs): request_headers = dict(request.headers) or {} for key, value in header.schema().get("properties", {}).items(): key_title = key.replace("_", "-").title() - # add original key + # Add original key if key_title in request_headers.keys(): request_headers[key] = request_headers[key_title] - request_kwargs.update({"header": header(**request_headers)}) + func_kwargs.update({"header": header(**request_headers)}) -def _do_cookie(cookie, request_kwargs): +def _do_cookie(cookie, func_kwargs): request_cookies = cookie(**request.cookies or {}) - request_kwargs.update({"cookie": request_cookies}) + func_kwargs.update({"cookie": request_cookies}) -def _do_path(path, kwargs, request_kwargs): - request_path = path(**kwargs) - request_kwargs.update({"path": request_path}) +def _do_path(path, path_kwargs, func_kwargs): + request_path = path(**path_kwargs) + func_kwargs.update({"path": request_path}) -def _do_query(query, request_kwargs): +def _do_query(query, func_kwargs): request_args = request.args query_dict = {} - for k, v in query.schema().get('properties', {}).items(): - if v.get('type') == 'array': + for k, v in query.schema().get("properties", {}).items(): + if v.get("type") == "array": value = request_args.getlist(k) else: value = request_args.get(k) if value is not None: query_dict[k] = value - request_kwargs.update({"query": query(**query_dict)}) + func_kwargs.update({"query": query(**query_dict)}) -def _do_form(form, request_kwargs): +def _do_form(form, func_kwargs): request_form = request.form request_files = request.files form_dict = {} - for k, v in form.schema().get('properties', {}).items(): - if v.get('type') == 'array': - items = v.get('items', {}) - if items.get('type') == 'string' and items.get('format') == 'binary': + for k, v in form.schema().get("properties", {}).items(): + if v.get("type") == "array": + items = v.get("items", {}) + if items.get("type") == "string" and items.get("format") == "binary": # List[FileStorage] - # {'title': 'Files', 'type': 'array', 'items': {'format': 'binary', 'type': 'string'} + # eg: {"title": "Files", "type": "array", "items": {"format": "binary", "type": "string"} value = request_files.getlist(k) else: # List[str], List[int] ... - # {'title': 'Files', 'type': 'array', 'items': {'type': 'string'} + # eg: {"title": "Files", "type": "array", "items": {"type": "string"} value = request_form.getlist(k) else: - if v.get('format') == 'binary': + if v.get("format") == "binary": # FileStorage value = request_files.get(k) else: @@ -68,22 +68,22 @@ def _do_form(form, request_kwargs): value = request_form.get(k) if value is not None: form_dict[k] = value - request_kwargs.update({"form": form(**form_dict)}) + func_kwargs.update({"form": form(**form_dict)}) -def _do_body(body, request_kwargs): +def _do_body(body, func_kwargs): obj = request.get_json(silent=True) or {} if isinstance(obj, str): try: obj = json.loads(obj) except JSONDecodeError as e: - raise ValidationError([ErrorWrapper(e, loc='__root__')], body) + raise ValidationError([ErrorWrapper(e, loc="__root__")], body) if body.__custom_root_type__: - # https://pydantic-docs.helpmanual.io/usage/models/#custom-root-types + # https://docs.pydantic.dev/latest/usage/models/#custom-root-types body_ = body(__root__=obj) else: body_ = body(**obj) - request_kwargs.update({"body": body_}) + func_kwargs.update({"body": body_}) def _do_request( @@ -93,39 +93,49 @@ def _do_request( query: Optional[Type[BaseModel]] = None, form: Optional[Type[BaseModel]] = None, body: Optional[Type[BaseModel]] = None, - **kwargs: Any + path_kwargs: Optional[Dict[Any, Any]] = None ) -> Union[Response, Dict]: """ - Validate requests and responses - :param func: view func - :param responses: response model - :param header: header model - :param cookie: cookie model - :param path: path model - :param query: query model - :param form: form model - :param body: body model - :param kwargs: path parameters - :return: + Validate requests and responses. + + Args: + header: Header model. + cookie: Cookie model. + path: Path model. + query: Query model. + form: Form model. + body: Body model. + path_kwargs: Path parameters. + + Returns: + Union[Response, Dict]: Request kwargs. + + Raises: + ValidationError: If validation fails. """ - # validate header, cookie, path and query - request_kwargs: Dict = dict() + + # Dictionary to store func kwargs + func_kwargs: Dict = dict() + try: + # Validate header, cookie, path, and query parameters if header: - _do_header(header, request_kwargs) + _do_header(header, func_kwargs) if cookie: - _do_cookie(cookie, request_kwargs) + _do_cookie(cookie, func_kwargs) if path: - _do_path(path, kwargs, request_kwargs) + _do_path(path, path_kwargs, func_kwargs) if query: - _do_query(query, request_kwargs) + _do_query(query, func_kwargs) if form: - _do_form(form, request_kwargs) + _do_form(form, func_kwargs) if body: - _do_body(body, request_kwargs) + _do_body(body, func_kwargs) except ValidationError as e: + # Create a JSON response with validation error details response = make_response(e.json()) - response.headers['Content-Type'] = 'application/json' + response.headers["Content-Type"] = "application/json" response.status_code = 422 return response - return request_kwargs + + return func_kwargs diff --git a/flask_openapi3/scaffold.py b/flask_openapi3/scaffold.py index 1177e06e..628047eb 100644 --- a/flask_openapi3/scaffold.py +++ b/flask_openapi3/scaffold.py @@ -4,9 +4,10 @@ import functools import inspect import sys +import warnings from abc import ABC from functools import wraps -from typing import Callable, List, Optional, Dict, Type, Any, Tuple +from typing import Callable, List, Optional, Dict, Type, Any, Tuple, Union from flask.scaffold import Scaffold from flask.wrappers import Response @@ -31,6 +32,8 @@ def iscoroutinefunction(func: Any) -> bool: return inspect.iscoroutinefunction(func) +warnings.simplefilter("once") + class APIScaffold(Scaffold, ABC): def _do_decorator( @@ -45,7 +48,7 @@ def _do_decorator( operation_id: Optional[str] = None, extra_form: Optional[ExtraRequestBody] = None, extra_body: Optional[ExtraRequestBody] = None, - responses: Optional[Dict[str, Optional[Type[BaseModel]]]] = None, + responses: Optional[Dict[str, Union[Type[BaseModel], Dict[Any, Any], None]]] = None, extra_responses: Optional[Dict[str, dict]] = None, deprecated: Optional[bool] = None, security: Optional[List[Dict[str, List[Any]]]] = None, @@ -71,22 +74,40 @@ def create_view_func( view_class=None, view_kwargs=None ): + """ + Create a view function that can be used with Flask to handle API requests. + + Arguments: + func: The original function to be called when handling the API request. + header: The header parameter for the API request. + cookie: The cookie parameter for the API request. + path: The path parameter for the API request. + query: The query parameter for the API request. + form: The form parameter for the API request. + body: The body parameter for the API request. + view_class: The class of the API view (if applicable). + view_kwargs: Additional keyword arguments to pass to the API view. + + Returns: + The view function that can be registered with Flask. + + """ is_coroutine_function = iscoroutinefunction(func) if is_coroutine_function: @wraps(func) async def view_func(**kwargs) -> Response: - result = _do_request( + func_kwargs = _do_request( header=header, cookie=cookie, path=path, query=query, form=form, body=body, - **kwargs + path_kwargs=kwargs ) - if isinstance(result, Response): + if isinstance(func_kwargs, Response): # 422 - return result + return func_kwargs # handle async request if view_class: signature = inspect.signature(view_class.__init__) @@ -95,9 +116,9 @@ async def view_func(**kwargs) -> Response: view_object = view_class(view_kwargs=view_kwargs) else: view_object = view_class() - response = await func(view_object, **result) + response = await func(view_object, **func_kwargs) else: - response = await func(**result) + response = await func(**func_kwargs) return response else: @wraps(func) @@ -109,7 +130,7 @@ def view_func(**kwargs) -> Response: query=query, form=form, body=body, - **kwargs + path_kwargs=kwargs ) if isinstance(result, Response): # 422 @@ -143,7 +164,7 @@ def get( operation_id: Optional[str] = None, extra_form: Optional[ExtraRequestBody] = None, extra_body: Optional[ExtraRequestBody] = None, - responses: Optional[Dict[str, Optional[Type[BaseModel]]]] = None, + responses: Optional[Dict[str, Union[Type[BaseModel], Dict[Any, Any], None]]] = None, extra_responses: Optional[Dict[str, dict]] = None, deprecated: Optional[bool] = None, security: Optional[List[Dict[str, List[Any]]]] = None, @@ -153,7 +174,7 @@ def get( **options: Any ) -> Callable: """ - Decorator for rest api, like: app.route(methods=["GET"]) + Decorator for defining a REST API endpoint with the HTTP GET method. More information goto https://spec.openapis.org/oas/v3.0.3#operation-object Arguments: @@ -165,7 +186,7 @@ def get( operation_id: Unique string used to identify the operation. extra_form: Extra information describing the request body(application/form). extra_body: Extra information describing the request body(application/json). - responses: response's model must be pydantic BaseModel. + responses: API responses should be either a subclass of BaseModel, a dictionary, or None. extra_responses: Extra information for responses. deprecated: Declares this operation to be deprecated. security: A declaration of which security mechanisms can be used for this operation. @@ -174,6 +195,19 @@ def get( doc_ui: Add openapi document UI(swagger, rapidoc and redoc). Defaults to True. """ + if extra_form is not None: + warnings.warn( + """`extra_form` will be deprecated in v3.x, please use `openapi_extra` instead.""", + DeprecationWarning) + if extra_body is not None: + warnings.warn( + """`extra_body` will be deprecated in v3.x, please use `openapi_extra` instead.""", + DeprecationWarning) + if extra_responses is not None: + warnings.warn( + """`extra_responses` will be deprecated in v3.x, please use `responses` instead.""", + DeprecationWarning) + def decorator(func) -> Callable: header, cookie, path, query, form, body = \ self._do_decorator( @@ -215,7 +249,7 @@ def post( operation_id: Optional[str] = None, extra_form: Optional[ExtraRequestBody] = None, extra_body: Optional[ExtraRequestBody] = None, - responses: Optional[Dict[str, Optional[Type[BaseModel]]]] = None, + responses: Optional[Dict[str, Union[Type[BaseModel], Dict[Any, Any], None]]] = None, extra_responses: Optional[Dict[str, dict]] = None, deprecated: Optional[bool] = None, security: Optional[List[Dict[str, List[Any]]]] = None, @@ -225,7 +259,7 @@ def post( **options: Any ) -> Callable: """ - Decorator for rest api, like: app.route(methods=["POST"]) + Decorator for defining a REST API endpoint with the HTTP POST method. More information goto https://spec.openapis.org/oas/v3.0.3#operation-object Arguments: @@ -237,7 +271,7 @@ def post( operation_id: Unique string used to identify the operation. extra_form: Extra information describing the request body(application/form). extra_body: Extra information describing the request body(application/json). - responses: response's model must be pydantic BaseModel. + responses: API responses should be either a subclass of BaseModel, a dictionary, or None. extra_responses: Extra information for responses. deprecated: Declares this operation to be deprecated. security: A declaration of which security mechanisms can be used for this operation. @@ -245,6 +279,18 @@ def post( openapi_extensions: Allows extensions to the OpenAPI Schema. doc_ui: Declares this operation to be shown. """ + if extra_form is not None: + warnings.warn( + """`extra_form` will be deprecated in v3.x, please use `openapi_extra` instead.""", + DeprecationWarning) + if extra_body is not None: + warnings.warn( + """`extra_body` will be deprecated in v3.x, please use `openapi_extra` instead.""", + DeprecationWarning) + if extra_responses is not None: + warnings.warn( + """`extra_responses` will be deprecated in v3.x, please use `responses` instead.""", + DeprecationWarning) def decorator(func) -> Callable: header, cookie, path, query, form, body = \ @@ -287,7 +333,7 @@ def put( operation_id: Optional[str] = None, extra_form: Optional[ExtraRequestBody] = None, extra_body: Optional[ExtraRequestBody] = None, - responses: Optional[Dict[str, Optional[Type[BaseModel]]]] = None, + responses: Optional[Dict[str, Union[Type[BaseModel], Dict[Any, Any], None]]] = None, extra_responses: Optional[Dict[str, dict]] = None, deprecated: Optional[bool] = None, security: Optional[List[Dict[str, List[Any]]]] = None, @@ -297,7 +343,7 @@ def put( **options: Any ) -> Callable: """ - Decorator for rest api, like: app.route(methods=["PUT"]) + Decorator for defining a REST API endpoint with the HTTP PUT method. More information goto https://spec.openapis.org/oas/v3.0.3#operation-object Arguments: @@ -309,7 +355,7 @@ def put( operation_id: Unique string used to identify the operation. extra_form: Extra information describing the request body(application/form). extra_body: Extra information describing the request body(application/json). - responses: response's model must be pydantic BaseModel. + responses: API responses should be either a subclass of BaseModel, a dictionary, or None. extra_responses: Extra information for responses. deprecated: Declares this operation to be deprecated. security: A declaration of which security mechanisms can be used for this operation. @@ -317,6 +363,18 @@ def put( openapi_extensions: Allows extensions to the OpenAPI Schema. doc_ui: Declares this operation to be shown. """ + if extra_form is not None: + warnings.warn( + """`extra_form` will be deprecated in v3.x, please use `openapi_extra` instead.""", + DeprecationWarning) + if extra_body is not None: + warnings.warn( + """`extra_body` will be deprecated in v3.x, please use `openapi_extra` instead.""", + DeprecationWarning) + if extra_responses is not None: + warnings.warn( + """`extra_responses` will be deprecated in v3.x, please use `responses` instead.""", + DeprecationWarning) def decorator(func) -> Callable: header, cookie, path, query, form, body = \ @@ -359,7 +417,7 @@ def delete( operation_id: Optional[str] = None, extra_form: Optional[ExtraRequestBody] = None, extra_body: Optional[ExtraRequestBody] = None, - responses: Optional[Dict[str, Optional[Type[BaseModel]]]] = None, + responses: Optional[Dict[str, Union[Type[BaseModel], Dict[Any, Any], None]]] = None, extra_responses: Optional[Dict[str, dict]] = None, deprecated: Optional[bool] = None, security: Optional[List[Dict[str, List[Any]]]] = None, @@ -369,7 +427,7 @@ def delete( **options: Any ) -> Callable: """ - Decorator for rest api, like: app.route(methods=["DELETE"]) + Decorator for defining a REST API endpoint with the HTTP DELETE method. More information goto https://spec.openapis.org/oas/v3.0.3#operation-object Arguments: @@ -381,7 +439,7 @@ def delete( operation_id: Unique string used to identify the operation. extra_form: Extra information describing the request body(application/form). extra_body: Extra information describing the request body(application/json). - responses: response's model must be pydantic BaseModel. + responses: API responses should be either a subclass of BaseModel, a dictionary, or None. extra_responses: Extra information for responses. deprecated: Declares this operation to be deprecated. security: A declaration of which security mechanisms can be used for this operation. @@ -389,6 +447,18 @@ def delete( openapi_extensions: Allows extensions to the OpenAPI Schema. doc_ui: Declares this operation to be shown. """ + if extra_form is not None: + warnings.warn( + """`extra_form` will be deprecated in v3.x, please use `openapi_extra` instead.""", + DeprecationWarning) + if extra_body is not None: + warnings.warn( + """`extra_body` will be deprecated in v3.x, please use `openapi_extra` instead.""", + DeprecationWarning) + if extra_responses is not None: + warnings.warn( + """`extra_responses` will be deprecated in v3.x, please use `responses` instead.""", + DeprecationWarning) def decorator(func) -> Callable: header, cookie, path, query, form, body = \ @@ -431,7 +501,7 @@ def patch( operation_id: Optional[str] = None, extra_form: Optional[ExtraRequestBody] = None, extra_body: Optional[ExtraRequestBody] = None, - responses: Optional[Dict[str, Optional[Type[BaseModel]]]] = None, + responses: Optional[Dict[str, Union[Type[BaseModel], Dict[Any, Any], None]]] = None, extra_responses: Optional[Dict[str, dict]] = None, deprecated: Optional[bool] = None, security: Optional[List[Dict[str, List[Any]]]] = None, @@ -441,7 +511,7 @@ def patch( **options: Any ) -> Callable: """ - Decorator for rest api, like: app.route(methods=["PATCH"]) + Decorator for defining a REST API endpoint with the HTTP PATCH method. More information goto https://spec.openapis.org/oas/v3.0.3#operation-object Arguments: @@ -453,7 +523,7 @@ def patch( operation_id: Unique string used to identify the operation. extra_form: Extra information describing the request body(application/form). extra_body: Extra information describing the request body(application/json). - responses: response's model must be pydantic BaseModel. + responses: API responses should be either a subclass of BaseModel, a dictionary, or None. extra_responses: Extra information for responses. deprecated: Declares this operation to be deprecated. security: A declaration of which security mechanisms can be used for this operation. @@ -461,6 +531,18 @@ def patch( openapi_extensions: Allows extensions to the OpenAPI Schema. doc_ui: Declares this operation to be shown. """ + if extra_form is not None: + warnings.warn( + """`extra_form` will be deprecated in v3.x, please use `openapi_extra` instead.""", + DeprecationWarning) + if extra_body is not None: + warnings.warn( + """`extra_body` will be deprecated in v3.x, please use `openapi_extra` instead.""", + DeprecationWarning) + if extra_responses is not None: + warnings.warn( + """`extra_responses` will be deprecated in v3.x, please use `responses` instead.""", + DeprecationWarning) def decorator(func) -> Callable: header, cookie, path, query, form, body = \ diff --git a/flask_openapi3/utils.py b/flask_openapi3/utils.py index 9a3faf05..beb94e4f 100644 --- a/flask_openapi3/utils.py +++ b/flask_openapi3/utils.py @@ -4,7 +4,7 @@ import inspect import re -from typing import get_type_hints, Dict, Type, Callable, List, Tuple, Optional, Any +from typing import get_type_hints, Dict, Type, Callable, List, Tuple, Optional, Any, Union from pydantic import BaseModel @@ -22,38 +22,70 @@ def get_operation( description: Optional[str] = None, openapi_extensions: Optional[Dict[str, Any]] = None, ) -> Operation: - """Return an Operation object with summary and description.""" + """ + Return an Operation object with the specified summary and description. + + Arguments: + func: The function or method for which the operation is being defined. + summary: A short summary of what the operation does. + description: A verbose explanation of the operation behavior. + openapi_extensions: Additional extensions to the OpenAPI Schema. + + Returns: + An Operation object representing the operation. + + """ + # Get the docstring of the function doc = inspect.getdoc(func) or "" doc = doc.strip() lines = doc.split("\n") doc_summary = lines[0] or None + + # Determine the summary and description based on provided arguments or docstring if summary is None: doc_description = lines[0] if len(lines) == 0 else "
".join(lines[1:]) or None else: doc_description = "
".join(lines) or None + # Create the operation dictionary with summary and description operation_dict = dict( summary=summary or doc_summary, description=description or doc_description ) - # update openapi_extensions + + # Add any additional openapi_extensions to the operation dictionary openapi_extensions = openapi_extensions or {} operation_dict.update(**openapi_extensions) + # Create and return the Operation object operation = Operation(**operation_dict) return operation def get_operation_id_for_path(*, name: str, path: str, method: str) -> str: + """ + Generate a unique operation ID based on the name, path, and method. + + Arguments: + name: The name or identifier for the operation. + path: The URL path for the operation. + method: The HTTP method for the operation. + + Returns: + A unique operation ID generated based on the provided name, path, and method. + + """ operation_id = name + path + # Replace non-word characters with underscores operation_id = re.sub(r"\W", "_", operation_id) operation_id = operation_id + "_" + method.lower() return operation_id def get_schema(obj: Type[BaseModel]) -> dict: - """Pydantic model conversion to openapi schema""" + """Converts a Pydantic model to an OpenAPI schema.""" + assert inspect.isclass(obj) and issubclass(obj, BaseModel), \ f"{obj} is invalid `pydantic.BaseModel`" @@ -61,7 +93,7 @@ def get_schema(obj: Type[BaseModel]) -> dict: def parse_header(header: Type[BaseModel]) -> Tuple[List[Parameter], dict]: - """Parse header model""" + """Parses a header model and returns a list of parameters and component schemas.""" schema = get_schema(header) parameters = [] components_schemas: Dict = dict() @@ -75,11 +107,11 @@ def parse_header(header: Type[BaseModel]) -> Tuple[List[Parameter], dict]: "required": name in schema.get("required", []), "schema": Schema(**value) } - # parse extra values + # Parse extra values data.update(**value) parameters.append(Parameter(**data)) - # parse definitions + # Parse definitions definitions = schema.get("definitions", {}) for name, value in definitions.items(): components_schemas[name] = Schema(**value) @@ -88,7 +120,7 @@ def parse_header(header: Type[BaseModel]) -> Tuple[List[Parameter], dict]: def parse_cookie(cookie: Type[BaseModel]) -> Tuple[List[Parameter], dict]: - """Parse cookie model""" + """Parses a cookie model and returns a list of parameters and component schemas.""" schema = get_schema(cookie) parameters = [] components_schemas: Dict = dict() @@ -102,11 +134,11 @@ def parse_cookie(cookie: Type[BaseModel]) -> Tuple[List[Parameter], dict]: "required": name in schema.get("required", []), "schema": Schema(**value) } - # parse extra values + # Parse extra values data.update(**value) parameters.append(Parameter(**data)) - # parse definitions + # Parse definitions definitions = schema.get("definitions", {}) for name, value in definitions.items(): components_schemas[name] = Schema(**value) @@ -115,7 +147,7 @@ def parse_cookie(cookie: Type[BaseModel]) -> Tuple[List[Parameter], dict]: def parse_path(path: Type[BaseModel]) -> Tuple[List[Parameter], dict]: - """Parse path model""" + """Parses a path model and returns a list of parameters and component schemas.""" schema = get_schema(path) parameters = [] components_schemas: Dict = dict() @@ -129,11 +161,11 @@ def parse_path(path: Type[BaseModel]) -> Tuple[List[Parameter], dict]: "required": True, "schema": Schema(**value) } - # parse extra values + # Parse extra values data.update(**value) parameters.append(Parameter(**data)) - # parse definitions + # Parse definitions definitions = schema.get("definitions", {}) for name, value in definitions.items(): components_schemas[name] = Schema(**value) @@ -142,7 +174,7 @@ def parse_path(path: Type[BaseModel]) -> Tuple[List[Parameter], dict]: def parse_query(query: Type[BaseModel]) -> Tuple[List[Parameter], dict]: - """Parse query model""" + """Parses a query model and returns a list of parameters and component schemas.""" schema = get_schema(query) parameters = [] components_schemas: Dict = dict() @@ -156,11 +188,11 @@ def parse_query(query: Type[BaseModel]) -> Tuple[List[Parameter], dict]: "required": name in schema.get("required", []), "schema": Schema(**value) } - # parse extra values + # Parse extra values data.update(**value) parameters.append(Parameter(**data)) - # parse definitions + # Parse definitions definitions = schema.get("definitions", {}) for name, value in definitions.items(): components_schemas[name] = Schema(**value) @@ -172,7 +204,7 @@ def parse_form( form: Type[BaseModel], extra_form: Optional[ExtraRequestBody] = None, ) -> Tuple[Dict[str, MediaType], dict]: - """Parse form model""" + """Parses a form model and returns a list of parameters and component schemas.""" schema = get_schema(form) components_schemas = dict() properties = schema.get("properties", {}) @@ -186,7 +218,7 @@ def parse_form( if v.get("type") == "array": encoding[k] = Encoding(style="form") if extra_form: - # update encoding + # Update encoding if extra_form.encoding: encoding.update(**extra_form.encoding) content = { @@ -205,7 +237,7 @@ def parse_form( ) } - # parse definitions + # Parse definitions definitions = schema.get("definitions", {}) for name, value in definitions.items(): components_schemas[name] = Schema(**value) @@ -217,7 +249,7 @@ def parse_body( body: Type[BaseModel], extra_body: Optional[ExtraRequestBody] = None, ) -> Tuple[Dict[str, MediaType], dict]: - """Parse body model""" + """Parses a body model and returns a list of parameters and component schemas.""" schema = get_schema(body) components_schemas = dict() @@ -239,7 +271,7 @@ def parse_body( ) } - # parse definitions + # Parse definitions definitions = schema.get("definitions", {}) for name, value in definitions.items(): components_schemas[name] = Schema(**value) @@ -248,22 +280,17 @@ def parse_body( def get_responses( - responses: dict, + responses: Optional[Dict[str, Union[Type[BaseModel], Dict[Any, Any], None]]], extra_responses: Dict[str, dict], components_schemas: dict, operation: Operation ) -> None: - """ - :param responses: Dict[str, BaseModel] - :param extra_responses: Dict[str, dict] - :param components_schemas: `models.component.py` `Components.schemas` - :param operation: `models.path.py` Operation - """ if responses is None: responses = {} _responses = {} _schemas = {} if not responses.get("422"): + # Handle 422 response for Unprocessable Entity _responses["422"] = Response( description=HTTP_STATUS["422"], content={ @@ -281,8 +308,13 @@ def get_responses( ) _schemas[UnprocessableEntity.__name__] = Schema(**UnprocessableEntity.schema()) for key, response in responses.items(): - # Verify that the response is a class and that class is a subclass of `pydantic.BaseModel` - if inspect.isclass(response) and issubclass(response, BaseModel): + if response is None: + # If the response is None, it means HTTP status code "204" (No Content) + _responses[key] = Response(description=HTTP_STATUS.get(key, "")) + continue + if isinstance(response, dict): + _responses[key] = response # type: ignore + else: schema = response.schema(ref_template=OPENAPI3_REF_TEMPLATE) _responses[key] = Response( description=HTTP_STATUS.get(key, ""), @@ -298,18 +330,27 @@ def get_responses( ) } ) + + model_config = response.Config + if hasattr(model_config, "openapi_extra"): + # Add additional information from model_config to the response + _responses[key].description = model_config.openapi_extra.get("description") + _responses[key].headers = model_config.openapi_extra.get("headers") + _responses[key].links = model_config.openapi_extra.get("links") + _content = _responses[key].content + _content["application/json"].example = model_config.openapi_extra.get("example") # type: ignore + _content["application/json"].examples = model_config.openapi_extra.get("examples") # type: ignore + _content["application/json"].encoding = model_config.openapi_extra.get("encoding") # type: ignore + if model_config.openapi_extra.get("content"): + _responses[key].content.update(model_config.openapi_extra.get("content")) # type: ignore + _schemas[response.__name__] = Schema(**schema) definitions = schema.get("definitions") if definitions: + # Add schema definitions to _schemas for name, value in definitions.items(): _schemas[name] = Schema(**value) - # Verify that if the response is None, because http status code "204" means return "No Content" - elif response is None: - _responses[key] = Response( - description=HTTP_STATUS.get(key, ""), - ) - else: - raise TypeError(f"{response} is invalid `pydantic.BaseModel`.") + # handle extra_responses for key, value in extra_responses.items(): # key "200" value {"content":{"text/csv":{"schema":{"type": "string"}}}} @@ -330,18 +371,30 @@ def parse_and_store_tags( old_tag_names: List[str], operation: Operation ) -> None: - """Store tags - :param new_tags: api tag - :param old_tags: openapi doc tags - :param old_tag_names: openapi doc tag names - :param operation: `models.path.py` Operation + """ + Parses new tags, stores them in old_tags list if they are not already present, + and updates the tags attribute of the operation object. + + Arguments: + new_tags: A list of new Tag objects to be parsed and stored. + old_tags: The list of existing Tag objects. + old_tag_names: The list of names of existing tags. + operation: The operation object whose tags attribute needs to be updated. + + Returns: + None """ if new_tags is None: new_tags = [] + + # Iterate over each tag in new_tags for tag in new_tags: if tag.name not in old_tag_names: old_tag_names.append(tag.name) old_tags.append(tag) + + # Set the tags attribute of the operation object to a list of unique tag names from new_tags + # If the resulting list is empty, set it to None operation.tags = list(set([tag.name for tag in new_tags])) or None @@ -355,45 +408,67 @@ def parse_parameters( doc_ui: bool = True, ) -> Tuple[Type[BaseModel], Type[BaseModel], Type[BaseModel], Type[BaseModel], Type[BaseModel], Type[BaseModel]]: """ - :param func: Flask view func - :param extra_form: Extra information describing the request body(application/form). - :param extra_body: Extra information describing the request body(application/json). - :param components_schemas: `models.component.py` Components.schemas - :param operation: `models.path.py` Operation - :param doc_ui: add openapi document UI(swagger and redoc). Defaults to True. + Parses the parameters of a given function and returns the types for header, cookie, path, + query, form, and body parameters. Also populates the Operation object with the parsed parameters. + + Arguments: + func: The function to parse the parameters from. + extra_form: Additional form data for the request body (default: None). + extra_body: Additional body data for the request body (default: None). + components_schemas: Dictionary to store the parsed components schemas (default: None). + operation: Operation object to populate with parsed parameters (default: None). + doc_ui: Flag indicating whether to return types for documentation UI (default: True). + + Returns: + Tuple[Type[BaseModel], Type[BaseModel], Type[BaseModel], Type[BaseModel], Type[BaseModel], Type[BaseModel]]: + The types for header, cookie, path, query, form, and body parameters respectively. + """ + # Get the type hints from the function annotations = get_type_hints(func) + + # Get the types for header, cookie, path, query, form, and body parameters header = annotations.get("header") cookie = annotations.get("cookie") path = annotations.get("path") query = annotations.get("query") form = annotations.get("form") body = annotations.get("body") + + # If doc_ui is False, return the types without further processing if doc_ui is False: return header, cookie, path, query, form, body # type: ignore + parameters = [] + + # If components_schemas is None, initialize it as an empty dictionary if components_schemas is None: components_schemas = dict() + + # If operation is None, initialize it as an Operation object if operation is None: operation = Operation() + if header: _parameters, _components_schemas = parse_header(header) parameters.extend(_parameters) components_schemas.update(**_components_schemas) + if cookie: _parameters, _components_schemas = parse_cookie(cookie) parameters.extend(_parameters) components_schemas.update(**_components_schemas) + if path: - # get args from a route path _parameters, _components_schemas = parse_path(path) parameters.extend(_parameters) components_schemas.update(**_components_schemas) + if query: - # get args from route query _parameters, _components_schemas = parse_query(query) parameters.extend(_parameters) components_schemas.update(**_components_schemas) + if form: _content, _components_schemas = parse_form(form, extra_form) components_schemas.update(**_components_schemas) @@ -407,7 +482,15 @@ def parse_parameters( request_body = RequestBody(**{ "content": _content, }) + model_config = form.Config + if hasattr(model_config, "openapi_extra"): + request_body.description = model_config.openapi_extra.get("description") + request_body.content["multipart/form-data"].example = model_config.openapi_extra.get("example") + request_body.content["multipart/form-data"].examples = model_config.openapi_extra.get("examples") + if model_config.openapi_extra.get("encoding"): + request_body.content["multipart/form-data"].encoding = model_config.openapi_extra.get("encoding") operation.requestBody = request_body + if body: _content, _components_schemas = parse_body(body, extra_body) components_schemas.update(**_components_schemas) @@ -419,7 +502,15 @@ def parse_parameters( ) else: request_body = RequestBody(content=_content) + model_config = body.Config + if hasattr(model_config, "openapi_extra"): + request_body.description = model_config.openapi_extra.get("description") + request_body.content["application/json"].example = model_config.openapi_extra.get("example") + request_body.content["application/json"].examples = model_config.openapi_extra.get("examples") + request_body.content["application/json"].encoding = model_config.openapi_extra.get("encoding") operation.requestBody = request_body + + # Set the parsed parameters in the operation object operation.parameters = parameters if parameters else None return header, cookie, path, query, form, body # type: ignore @@ -427,11 +518,18 @@ def parse_parameters( def parse_method(uri: str, method: str, paths: dict, operation: Operation) -> None: """ - :param uri: api route path - :param method: get post put delete patch - :param paths: openapi doc paths - :param operation: `models.path.py` Operation + Parses the HTTP method and updates the corresponding PathItem object in the paths' dictionary. + + Arguments: + uri: The URI of the API endpoint. + method: The HTTP method for the API endpoint. + paths: A dictionary containing the API paths and their corresponding PathItem objects. + operation: The Operation object to assign to the PathItem. + + Returns: + None """ + # Check the HTTP method and update the PathItem object in the paths dictionary if method == HTTPMethod.GET: if not paths.get(uri): paths[uri] = PathItem(get=operation) diff --git a/flask_openapi3/view.py b/flask_openapi3/view.py index 496d828e..4807867b 100644 --- a/flask_openapi3/view.py +++ b/flask_openapi3/view.py @@ -3,12 +3,13 @@ # @Time : 2022/10/14 16:09 import re import typing +import warnings if typing.TYPE_CHECKING: from .openapi import OpenAPI from copy import deepcopy -from typing import Optional, List, Dict, Type, Any, Callable +from typing import Optional, List, Dict, Type, Any, Callable, Union from pydantic import BaseModel @@ -19,6 +20,8 @@ from .utils import get_operation, parse_and_store_tags, parse_parameters, get_responses, parse_method, \ get_operation_id_for_path +warnings.simplefilter("once") + class APIView: def __init__( @@ -26,7 +29,7 @@ def __init__( url_prefix: Optional[str] = None, view_tags: Optional[List[Tag]] = None, view_security: Optional[List[Dict[str, List[str]]]] = None, - view_responses: Optional[Dict[str, Optional[Type[BaseModel]]]] = None, + view_responses: Optional[Dict[str, Union[Type[BaseModel], Dict[Any, Any], None]]] = None, doc_ui: bool = True, operation_id_callback: Callable = get_operation_id_for_path, ): @@ -35,13 +38,13 @@ def __init__( Arguments: url_prefix: A path to prepend to all the APIView's urls - view_tags: APIView tags for every api - view_security: APIView security for every api - view_responses: APIView response models - doc_ui: Add openapi document UI(swagger, rapidoc and redoc). Defaults to True. + view_tags: APIView tags for every API. + view_security: APIView security for every API. + view_responses: API responses should be either a subclass of BaseModel, a dictionary, or None. + doc_ui: Enable OpenAPI document UI (Swagger UI and Redoc). Defaults to True. operation_id_callback: Callback function for custom operation_id generation. - Receives name (str), path (str) and method (str) parameters. - Defaults to `get_operation_id_for_path` from utils + Receives name (str), path (str) and method (str) parameters. + Defaults to `get_operation_id_for_path` from utils """ self.url_prefix = url_prefix self.view_tags = view_tags or [] @@ -63,10 +66,10 @@ def wrapper(cls): if self.views.get(rule): raise ValueError(f"malformed url rule: {rule!r}") methods = [] - # /pet/ --> /pet/{petId} + # Convert route parameter format from /pet/ to /pet/{petId} uri = re.sub(r"<([^<:]+:)?", "{", rule).replace(">", "}") trail_slash = uri.endswith("/") - # merge url_prefix and uri + # Merge url_prefix and uri uri = self.url_prefix.rstrip("/") + "/" + uri.lstrip("/") if self.url_prefix else uri if not trail_slash: uri = uri.rstrip("/") @@ -79,16 +82,16 @@ def wrapper(cls): continue if not getattr(cls_method, "operation", None): continue - # parse method + # Parse method parse_method(uri, method, self.paths, cls_method.operation) - # update operation_id + # Update operation_id if not cls_method.operation.operationId: cls_method.operation.operationId = self.operation_id_callback( name=cls_method.__qualname__, path=rule, method=method ) - # /pet/{petId} --> /pet/ + # Convert route parameters from to {param} _rule = uri.replace("{", "<").replace("}", ">") self.views[_rule] = (cls, methods) @@ -106,7 +109,7 @@ def doc( operation_id: Optional[str] = None, extra_form: Optional[ExtraRequestBody] = None, extra_body: Optional[ExtraRequestBody] = None, - responses: Optional[Dict[str, Optional[Type[BaseModel]]]] = None, + responses: Optional[Dict[str, Union[Type[BaseModel], Dict[Any, Any], None]]] = None, extra_responses: Optional[Dict[str, dict]] = None, deprecated: Optional[bool] = None, security: Optional[List[Dict[str, List[Any]]]] = None, @@ -126,7 +129,7 @@ def doc( operation_id: Unique string used to identify the operation. extra_form: Extra information describing the request body(application/form). extra_body: Extra information describing the request body(application/json). - responses: response's model must be pydantic BaseModel. + responses: API responses should be either a subclass of BaseModel, a dictionary, or None. extra_responses: Extra information for responses. deprecated: Declares this operation to be deprecated. security: A declaration of which security mechanisms can be used for this operation. @@ -135,6 +138,19 @@ def doc( doc_ui: Add openapi document UI(swagger, rapidoc and redoc). Defaults to True. """ + if extra_form is not None: + warnings.warn( + """`extra_form` will be deprecated in v3.x, please use `openapi_extra` instead.""", + DeprecationWarning) + if extra_body is not None: + warnings.warn( + """`extra_body` will be deprecated in v3.x, please use `openapi_extra` instead.""", + DeprecationWarning) + if extra_responses is not None: + warnings.warn( + """`extra_responses` will be deprecated in v3.x, please use `responses` instead.""", + DeprecationWarning) + if responses is None: responses = {} if extra_responses is None: @@ -146,29 +162,29 @@ def doc( def decorator(func): if self.doc_ui is False or doc_ui is False: return - # global response combine api responses + # Global response combines API responses combine_responses = deepcopy(self.view_responses) combine_responses.update(**responses) - # create operation + # Create operation operation = get_operation( func, summary=summary, description=description, openapi_extensions=openapi_extensions ) - # set external docs + # Set external docs operation.externalDocs = external_docs # Unique string used to identify the operation. operation.operationId = operation_id - # only set `deprecated` if True otherwise leave it as None + # Only set `deprecated` if True, otherwise leave it as None operation.deprecated = deprecated - # add security + # Add security operation.security = security + self.view_security or None - # add servers + # Add servers operation.servers = servers - # store tags + # Store tags parse_and_store_tags(tags, self.tags, self.tag_names, operation) - # parse parameters + # Parse parameters parse_parameters( func, extra_form=extra_form, @@ -176,7 +192,7 @@ def decorator(func): components_schemas=self.components_schemas, operation=operation ) - # parse response + # Parse response get_responses(combine_responses, extra_responses, self.components_schemas, operation) func.operation = operation @@ -185,6 +201,16 @@ def decorator(func): return decorator def register(self, app: "OpenAPI", view_kwargs: Optional[Dict[Any, Any]] = None): + """ + Register the API views with the given OpenAPI app. + + Args: + app: An instance of the OpenAPI app. + view_kwargs: Additional keyword arguments to pass to the API views. + + Returns: + None + """ if view_kwargs is None: view_kwargs = {} for rule, (cls, methods) in self.views.items(): diff --git a/mkdocs.yml b/mkdocs.yml index 10681601..13ceaae4 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -89,6 +89,7 @@ nav: - Operation: Tutorial/Operation.md - Request: Tutorial/Request.md - Response: Tutorial/Response.md + - OpenAPI Extra: Tutorial/Openapi_extra.md - UI Templates: Tutorial/UI_Templates.md - JSON: Tutorial/JSON.md - Example: Example.md diff --git a/tests/test_openapi.py b/tests/test_openapi.py index c2453537..1d711455 100644 --- a/tests/test_openapi.py +++ b/tests/test_openapi.py @@ -2,11 +2,10 @@ from pydantic import BaseModel -from flask_openapi3 import Example -from flask_openapi3 import OpenAPI, ExtraRequestBody +from flask_openapi3 import OpenAPI -def test_responses_and_extra_responses_are_replicated_in_open_api(request): +def test_responses_are_replicated_in_open_api(request): test_app = OpenAPI(request.node.name) test_app.config["TESTING"] = True @@ -14,11 +13,8 @@ class BaseResponse(BaseModel): """Base description""" test: int - @test_app.get( - "/test", - responses={"201": BaseResponse}, - extra_responses={ - "201": { + class Config: + openapi_extra = { "description": "Custom description", "headers": { "location": { @@ -37,8 +33,8 @@ class BaseResponse(BaseModel): } } } - } - ) + + @test_app.get("/test", responses={"201": BaseResponse}) def endpoint_test(): return b'', 201 @@ -58,7 +54,7 @@ def endpoint_test(): "application/json": { "schema": {"$ref": "#/components/schemas/BaseResponse"} }, - # While this one comes from extra_responses + # While this one comes from responses "text/plain": { "schema": {"type": "string"} } @@ -71,14 +67,13 @@ def endpoint_test(): } -def test_none_responses_and_extra_responses_are_replicated_in_open_api(request): +def test_none_responses_are_replicated_in_open_api(request): test_app = OpenAPI(request.node.name) test_app.config["TESTING"] = True @test_app.get( "/test", - responses={"204": None}, - extra_responses={ + responses={ "204": { "description": "Custom description", "headers": { @@ -127,13 +122,13 @@ def endpoint_test(): } -def test_extra_responses_are_replicated_in_open_api(request): +def test_responses_are_replicated_in_open_api2(request): test_app = OpenAPI(request.node.name) test_app.config["TESTING"] = True @test_app.get( "/test", - extra_responses={ + responses={ "201": { "description": "Custom description", "headers": { @@ -182,13 +177,13 @@ def endpoint_test(): } -def test_extra_responses_without_content_are_replicated_in_open_api(request): +def test_responses_without_content_are_replicated_in_open_api(request): test_app = OpenAPI(request.node.name) test_app.config["TESTING"] = True @test_app.get( "/test", - extra_responses={ + responses={ "201": { "description": "Custom description", "headers": { @@ -237,26 +232,28 @@ def test_body_examples_are_replicated_in_open_api(request): test_app = OpenAPI(request.node.name) test_app.config["TESTING"] = True - @test_app.post( - "/test", - extra_body=ExtraRequestBody( - examples={ - "Example 01": Example( - summary="An example", - value={ + class Config: + openapi_extra = { + "examples": { + "Example 01": { + "summary": "An example", + "value": { "test_int": -1, "test_str": "negative", } - ), - "Example 02": Example( - externalValue="https://example.org/examples/second-example.xml" - ), - "Example 03": Example(**{ + }, + "Example 02": { + "externalValue": "https://example.org/examples/second-example.xml" + }, + "Example 03": { "$ref": "#/components/examples/third-example" - }) + } } - ) - ) + } + + BaseRequest.Config = Config + + @test_app.post("/test") def endpoint_test(body: BaseRequest): return body.json(), 200 @@ -278,54 +275,26 @@ def endpoint_test(body: BaseRequest): } -def test_body_examples_are_not_replicated_with_form(request): +def test_form_examples(request): test_app = OpenAPI(request.node.name) test_app.config["TESTING"] = True - @test_app.post( - "/test", - extra_body=ExtraRequestBody(example={ - "Example 01": Example(**{ - "summary": "An example", - "value": { - "test_int": -1, - "test_str": "negative", - } - }), - }) - ) - def endpoint_test(form: BaseRequest): - return form.json(), 200 - - with test_app.test_client() as client: - resp = client.get("/openapi/openapi.json") - assert resp.status_code == 200 - assert resp.json["paths"]["/test"]["post"]["requestBody"] == { - "content": { - "multipart/form-data": { - "schema": {"$ref": "#/components/schemas/BaseRequest"} + class Config: + openapi_extra = { + "examples": { + "Example 01": { + "summary": "An example", + "value": { + "test_int": -1, + "test_str": "negative", + } } - }, - "required": True + } } + BaseRequest.Config = Config -def test_form_examples(request): - test_app = OpenAPI(request.node.name) - test_app.config["TESTING"] = True - - @test_app.post( - "/test", - extra_form=ExtraRequestBody(examples={ - "Example 01": Example(**{ - "summary": "An example", - "value": { - "test_int": -1, - "test_str": "negative", - } - }), - }) - ) + @test_app.post("/test") def endpoint_test(form: BaseRequest): return form.json(), 200 diff --git a/tests/test_restapi.py b/tests/test_restapi.py index 3f4280fb..fa7132b5 100644 --- a/tests/test_restapi.py +++ b/tests/test_restapi.py @@ -106,7 +106,6 @@ def client(): url="https://www.openapis.org/", description="Something great got better, get excited!"), responses={"200": BookResponse}, - extra_responses={"200": {"content": {"text/csv": {"schema": {"type": "string"}}}}}, security=security ) def get_book(path: BookPath): diff --git a/tests/test_restapi_with_doc_prefix.py b/tests/test_restapi_with_doc_prefix.py index d356ad51..8806f149 100644 --- a/tests/test_restapi_with_doc_prefix.py +++ b/tests/test_restapi_with_doc_prefix.py @@ -70,6 +70,13 @@ class BookResponse(BaseModel): message: str = Field("ok", description="Exception Information") data: BookBodyWithID + class Config: + openapi_extra = { + "content": { + "text/csv": {"schema": {"type": "string"}} + } + } + @pytest.fixture def client(): @@ -82,7 +89,6 @@ def client(): '/book/', tags=[book_tag], responses={"200": BookResponse}, - extra_responses={"200": {"content": {"text/csv": {"schema": {"type": "string"}}}}}, security=security ) def get_book(path: BookPath):