跳转至

OpenAPI

OpenAPI ⚓︎

Bases: APIScaffold, Flask

Source code in flask_openapi3/openapi.py
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
class OpenAPI(APIScaffold, Flask):
    def __init__(
            self,
            import_name: str,
            *,
            info: Optional[Info] = None,
            security_schemes: Optional[SecuritySchemesDict] = None,
            responses: Optional[ResponseDict] = None,
            servers: Optional[List[Server]] = None,
            external_docs: Optional[ExternalDocumentation] = None,
            operation_id_callback: Callable = get_operation_id_for_path,
            openapi_extensions: Optional[Dict[str, Any]] = None,
            validation_error_status: Union[str, int] = 422,
            validation_error_model: Type[BaseModel] = ValidationErrorModel,
            validation_error_callback: Callable = make_validation_error_response,
            doc_ui: bool = True,
            doc_prefix: str = "/openapi",
            doc_url: str = "/openapi.json",
            **kwargs: Any
    ) -> None:
        """
        OpenAPI class that provides REST API functionality along with Swagger UI and Redoc.

        Args:
            import_name: The import name for the Flask application.
            info: Information about the API (title, version, etc.).
                See https://spec.openapis.org/oas/v3.1.0#info-object.
            security_schemes: Security schemes for the API.
                See https://spec.openapis.org/oas/v3.1.0#security-scheme-object.
            responses: API responses should be either a subclass of BaseModel, a dictionary, or None.
            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.1.0#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.1.0#specification-extensions.
            validation_error_status:
                HTTP Status of the response given when a validation error is detected by pydantic.
                Defaults to 422.
            validation_error_model: Validation error response model for OpenAPI Specification.
            validation_error_callback: Validation error response callback, the return format corresponds to
                the validation_error_model.
            doc_ui: Enable OpenAPI document UI (Swagger UI and Redoc).
                Defaults to True.
            doc_prefix: URL prefix used for OpenAPI document and UI.
                Defaults to "/openapi".
            doc_url: URL for accessing the OpenAPI specification document in JSON format.
                Defaults to "/openapi.json".
            **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.1.0"
        self.info = info or Info(title="OpenAPI", version="1.0.0")

        # Set security schemes, responses, paths and components
        self.security_schemes = security_schemes

        # Convert key to string
        self.responses = convert_responses_key_to_string(responses or {})

        # Initialize instance variables
        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.doc_url = doc_url

        # 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 {}

        # Set HTTP Response of validation errors within OpenAPI
        self.validation_error_status = str(validation_error_status)
        self.validation_error_model = validation_error_model
        self.validation_error_callback = validation_error_callback

        # Initialize the OpenAPI documentation UI
        if doc_ui:
            self._init_doc()

        # Add the OpenAPI command
        self.cli.add_command(openapi_command)  # type: ignore

        # Initialize specification JSON
        self.spec_json: Dict = {}

    def _init_doc(self) -> None:
        """
        Provide Swagger UI, Redoc, and Rapidoc
        """
        _here = os.path.dirname(__file__)
        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__,
            url_prefix=self.doc_prefix,
            template_folder=template_folder,
            static_folder=static_folder
        )

        # Add the API documentation URL rule
        blueprint.add_url_rule(
            rule=self.doc_url,
            endpoint="doc_url",
            view_func=lambda: self.api_doc
        )

        ui_templates = []
        # Iterate over all entry points in the "flask_openapi3.plugins" group
        for entry_point in entry_points(group="flask_openapi3.plugins"):
            try:
                module_path = entry_point.value
                module_name, class_name = module_path.rsplit(".", 1)
                module = import_module(module_name)
                plugin_class = getattr(module, class_name)
                plugin_register = plugin_class.register
                plugin_name = plugin_class.name
                plugin_display_name = plugin_class.display_name
                bp = plugin_register(doc_url=self.doc_url.lstrip("/"))
                self.register_blueprint(bp, url_prefix=self.doc_prefix)
                ui_templates.append({"name": plugin_name, "display_name": plugin_display_name})
            except (ModuleNotFoundError, AttributeError):  # pragma: no cover
                import traceback
                print(f"Warning: plugin '{entry_point.value}' registration failed.")
                traceback.print_exc()

        # Add URL rule for the home page
        blueprint.add_url_rule(
            rule="/",
            endpoint="openapi",
            view_func=lambda: render_template_string(
                self.config.get("OPENAPI_HTML_STRING") or openapi_html_string,
                ui_templates=ui_templates
            )
        )

        # Register the blueprint with the Flask application
        self.register_blueprint(blueprint)

    @property
    def api_doc(self) -> Dict:
        """
        Generate the OpenAPI specification JSON.

        Returns:
            The OpenAPI specification JSON as a dictionary.

        """
        if self.spec_json:
            return self.spec_json

        spec = APISpec(
            openapi=self.openapi_version,
            info=self.info,
            paths=self.paths
        )

        if self.severs:
            spec.servers = self.severs

        if self.external_docs:
            spec.externalDocs = self.external_docs

        # Set tags
        if self.tags:
            spec.tags = self.tags

        # Add ValidationErrorModel to components schemas
        schema = get_model_schema(self.validation_error_model)
        self.components_schemas[self.validation_error_model.__name__] = Schema(**schema)

        # Parse definitions
        definitions = schema.get("$defs", {})
        for name, value in definitions.items():
            self.components_schemas[name] = Schema(**value)

        # Set components
        self.components.schemas = self.components_schemas
        self.components.securitySchemes = self.security_schemes
        spec.components = self.components

        # Convert spec to JSON
        self.spec_json = spec.model_dump(mode="json", by_alias=True, exclude_unset=True, warnings=False)

        # Update with OpenAPI extensions
        self.spec_json.update(**self.openapi_extensions)

        # Handle validation error response
        for rule, path_item in self.spec_json["paths"].items():
            for http_method, operation in path_item.items():
                if operation.get("parameters") is None and operation.get("requestBody") is None:
                    continue
                if not operation.get("responses"):
                    operation["responses"] = {}
                if operation["responses"].get(self.validation_error_status):
                    continue
                operation["responses"][self.validation_error_status] = {
                    "description": HTTP_STATUS[self.validation_error_status],
                    "content": {
                        "application/json": {
                            "schema": {
                                "type": "array",
                                "items": {"$ref": f"{OPENAPI3_REF_PREFIX}/{self.validation_error_model.__name__}"}
                            }
                        }
                    }
                }

        return self.spec_json

    def register_api(self, api: APIBlueprint) -> None:
        """
        Register an APIBlueprint.

        Args:
            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 component schemas with the APIBlueprint's component 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:
        """
        Register APIView

        Args:
            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 component schemas with the APIView's component 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 _add_url_rule(
            self,
            rule,
            endpoint=None,
            view_func=None,
            provide_automatic_options=None,
            **options,
    ) -> None:
        self.add_url_rule(rule, endpoint, view_func, provide_automatic_options, **options)

    def _collect_openapi_info(
            self,
            rule: str,
            func: Callable,
            *,
            tags: Optional[List[Tag]] = None,
            summary: Optional[str] = None,
            description: Optional[str] = None,
            external_docs: Optional[ExternalDocumentation] = None,
            operation_id: Optional[str] = None,
            responses: Optional[ResponseDict] = None,
            deprecated: Optional[bool] = None,
            security: Optional[List[Dict[str, List[Any]]]] = None,
            servers: Optional[List[Server]] = None,
            openapi_extensions: Optional[Dict[str, Any]] = None,
            doc_ui: bool = True,
            method: str = HTTPMethod.GET
    ) -> ParametersTuple:
        """
        Collects OpenAPI specification information for Flask routes and view functions.

        Args:
            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.
            external_docs: Additional external documentation for this operation.
            operation_id: Unique string used to identify the operation.
            responses: API responses should be either a subclass of BaseModel, a dictionary, or None.
            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: Declares this operation to be shown. Default to True.
            method: HTTP method for the operation. Defaults to GET.
        """
        if doc_ui is True:
            # Convert key to string
            new_responses = convert_responses_key_to_string(responses or {})

            # Global response: combine API responses
            combine_responses = {**self.responses, **new_responses}

            # Create operation
            operation = get_operation(
                func,
                summary=summary,
                description=description,
                openapi_extensions=openapi_extensions
            )
            # Set external docs
            if 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
            if deprecated is not None:
                operation.deprecated = deprecated

            # Add security
            if security:
                operation.security = security

            # Add servers
            if servers:
                operation.servers = servers

            # Store tags
            parse_and_store_tags(tags or [], self.tags, self.tag_names, operation)

            # Parse response
            get_responses(combine_responses, self.components_schemas, operation)

            # Convert a route parameter format from /pet/<petId> to /pet/{petId}
            uri = re.sub(r"<([^<:]+:)?", "{", rule).replace(">", "}")

            # Parse method
            parse_method(uri, method, self.paths, operation)

            # Parse parameters
            return parse_parameters(func, components_schemas=self.components_schemas, operation=operation)
        else:
            return parse_parameters(func, doc_ui=False)

api_doc: Dict property ⚓︎

Generate the OpenAPI specification JSON.

Returns:

Type Description
Dict

The OpenAPI specification JSON as a dictionary.

__init__(import_name, *, info=None, security_schemes=None, responses=None, servers=None, external_docs=None, operation_id_callback=get_operation_id_for_path, openapi_extensions=None, validation_error_status=422, validation_error_model=ValidationErrorModel, validation_error_callback=make_validation_error_response, doc_ui=True, doc_prefix='/openapi', doc_url='/openapi.json', **kwargs) ⚓︎

OpenAPI class that provides REST API functionality along with Swagger UI and Redoc.

Parameters:

Name Type Description Default
import_name str

The import name for the Flask application.

required
info Optional[Info]

Information about the API (title, version, etc.). See https://spec.openapis.org/oas/v3.1.0#info-object.

None
security_schemes Optional[SecuritySchemesDict] None
responses Optional[ResponseDict]

API responses should be either a subclass of BaseModel, a dictionary, or None.

None
servers Optional[List[Server]]

An array of Server objects providing connectivity information to a target server.

None
external_docs Optional[ExternalDocumentation]

External documentation for the API. See: https://spec.openapis.org/oas/v3.1.0#external-documentation-object.

None
operation_id_callback Callable

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.

get_operation_id_for_path
openapi_extensions Optional[Dict[str, Any]]

Extensions to the OpenAPI Schema. See https://spec.openapis.org/oas/v3.1.0#specification-extensions.

None
validation_error_status Union[str, int]

HTTP Status of the response given when a validation error is detected by pydantic. Defaults to 422.

422
validation_error_model Type[BaseModel]

Validation error response model for OpenAPI Specification.

ValidationErrorModel
validation_error_callback Callable

Validation error response callback, the return format corresponds to the validation_error_model.

make_validation_error_response
doc_ui bool

Enable OpenAPI document UI (Swagger UI and Redoc). Defaults to True.

True
doc_prefix str

URL prefix used for OpenAPI document and UI. Defaults to "/openapi".

'/openapi'
doc_url str

URL for accessing the OpenAPI specification document in JSON format. Defaults to "/openapi.json".

'/openapi.json'
**kwargs Any

Additional kwargs to be passed to Flask.

{}
Source code in flask_openapi3/openapi.py
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def __init__(
        self,
        import_name: str,
        *,
        info: Optional[Info] = None,
        security_schemes: Optional[SecuritySchemesDict] = None,
        responses: Optional[ResponseDict] = None,
        servers: Optional[List[Server]] = None,
        external_docs: Optional[ExternalDocumentation] = None,
        operation_id_callback: Callable = get_operation_id_for_path,
        openapi_extensions: Optional[Dict[str, Any]] = None,
        validation_error_status: Union[str, int] = 422,
        validation_error_model: Type[BaseModel] = ValidationErrorModel,
        validation_error_callback: Callable = make_validation_error_response,
        doc_ui: bool = True,
        doc_prefix: str = "/openapi",
        doc_url: str = "/openapi.json",
        **kwargs: Any
) -> None:
    """
    OpenAPI class that provides REST API functionality along with Swagger UI and Redoc.

    Args:
        import_name: The import name for the Flask application.
        info: Information about the API (title, version, etc.).
            See https://spec.openapis.org/oas/v3.1.0#info-object.
        security_schemes: Security schemes for the API.
            See https://spec.openapis.org/oas/v3.1.0#security-scheme-object.
        responses: API responses should be either a subclass of BaseModel, a dictionary, or None.
        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.1.0#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.1.0#specification-extensions.
        validation_error_status:
            HTTP Status of the response given when a validation error is detected by pydantic.
            Defaults to 422.
        validation_error_model: Validation error response model for OpenAPI Specification.
        validation_error_callback: Validation error response callback, the return format corresponds to
            the validation_error_model.
        doc_ui: Enable OpenAPI document UI (Swagger UI and Redoc).
            Defaults to True.
        doc_prefix: URL prefix used for OpenAPI document and UI.
            Defaults to "/openapi".
        doc_url: URL for accessing the OpenAPI specification document in JSON format.
            Defaults to "/openapi.json".
        **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.1.0"
    self.info = info or Info(title="OpenAPI", version="1.0.0")

    # Set security schemes, responses, paths and components
    self.security_schemes = security_schemes

    # Convert key to string
    self.responses = convert_responses_key_to_string(responses or {})

    # Initialize instance variables
    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.doc_url = doc_url

    # 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 {}

    # Set HTTP Response of validation errors within OpenAPI
    self.validation_error_status = str(validation_error_status)
    self.validation_error_model = validation_error_model
    self.validation_error_callback = validation_error_callback

    # Initialize the OpenAPI documentation UI
    if doc_ui:
        self._init_doc()

    # Add the OpenAPI command
    self.cli.add_command(openapi_command)  # type: ignore

    # Initialize specification JSON
    self.spec_json: Dict = {}

register_api(api) ⚓︎

Register an APIBlueprint.

Parameters:

Name Type Description Default
api APIBlueprint

The APIBlueprint instance to register.

required
Source code in flask_openapi3/openapi.py
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
def register_api(self, api: APIBlueprint) -> None:
    """
    Register an APIBlueprint.

    Args:
        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 component schemas with the APIBlueprint's component schemas
    self.components_schemas.update(**api.components_schemas)

    # Register the APIBlueprint with the current instance
    self.register_blueprint(api)

register_api_view(api_view, view_kwargs=None) ⚓︎

Register APIView

Parameters:

Name Type Description Default
api_view APIView

The APIView instance to register.

required
view_kwargs Optional[Dict[Any, Any]]

Additional keyword arguments to pass to the API views.

None
Source code in flask_openapi3/openapi.py
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
def register_api_view(self, api_view: APIView, view_kwargs: Optional[Dict[Any, Any]] = None) -> None:
    """
    Register APIView

    Args:
        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 component schemas with the APIView's component schemas
    self.components_schemas.update(**api_view.components_schemas)

    # Register the APIView with the current instance
    api_view.register(self, view_kwargs=view_kwargs)