Integrating Swagger UI with Your Python Flask

Swagger is an open-source toolset for generating RESTful API documentation. Additionally, it offers utility tools for developers, such as the automatic generation of client libraries, server stubs, and more.

Swagger consists of the following key components:

1. Swagger Editor is a browser-based API designer tool that allows developers to manually create API documentation with validation and autocomplete features in YAML or JSON formats.
2. Swagger UI is a visualization tool that allows developers to build an interactive API console for examining documentation and testing RESTful APIs.
3. Swagger Codegen is an automated tool that generates client libraries, server stubs, and API documentation, helping developers in software system development.

OpenAPI Specification (OAS) is a standard of RESTful API specification used by Swagger to build API documentation. It offers the management of multiple versions of API documentation and can be embedded directly within the source code.

In this post, I demonstrate how to integrate Swagger UI into a Python Flask application. Here is a step-by-step guide to help you get started:

1. Create a Flask application
If you don’t have a Flask application, create one by installing Flask using the following command:

pip install flask

Next, establish a Flask application instance by creating a Python file named “app.py” to expose RESTful API methods. However, I already have an existing Flask application named “flask-kafka.” The flask-kafka project was developed to publish and subscribe from Kafka topics by exposing services as REST-API. This API enables two REST controllers, including the consumer and producer. The consumer contains a consume method that allows the retrieval of messages from a specific Kafka topic, whereas the producer contains a produce method that allows the dispatch of messages to a specific Kafka topic. The content of my app.py is as follows:

import sys
from flask import Flask
from pathlib import Path
from api.bootstrap import Bootstrap

from api.controller.consumer_controller import consumer
from api.controller.producer_controller import producer

"""
Author          : Neda Peyrone
Create Date     : 30-08-2023
File            : app.py
"""

def create_app():
  path = Path().absolute()
  Bootstrap(path)

  app = Flask(__name__)

  app.register_blueprint(consumer)
  app.register_blueprint(producer)

  return app


if __name__ == '__main__':
  try:
    port = int(sys.argv[1])
  except Exception:
    port = 8081

  create_app().run(host='0.0.0.0', port=port, debug=True, use_reloader=True)

The code below is an example of how to create the consumer_controller.py file.

from flask import request, Blueprint
from marshmallow import ValidationError
from api.domain.consumer import Consumer
from api.util import exception_util, service_util
from api.constant.message_code import MessageCode
from api.schema.consumer_schema import ConsumerSchema
from api.exception.service_exception import ServiceException
from api.business_logic.consumer_service import ConsumerService

consumer = Blueprint('consumer', __name__)

"""
Author          : Neda Peyrone
Create Date     : 30-08-2023
File            : consumer_controller.py
"""

@consumer.route("/consume", methods=['POST'])
def consume():
  payload = request.get_json()

  try:
    validated_data = ConsumerSchema().load(payload)
    consumer = Consumer(**validated_data)

    msg = ConsumerService().consume(consumer)
    return service_util.build_server_response(MessageCode.SUCCESS, msg)
  except (ValidationError, ServiceException) as err:
    return exception_util.handler(err)

2. Install a Flasgger library
The Flasgger library facilitates the incorporation of Swagger UI into the flask-kafka project, which can be installed using the command below:

pip install flasgger

3. Configure Swagger UI in the flask-kafka project
To begin with, configure my existing app.py file to introduce Swagger UI into the application according to the code below.

import sys
from flask import Flask
from pathlib import Path
from flasgger import Swagger
from api.bootstrap import Bootstrap

from api.controller.consumer_controller import consumer
from api.controller.producer_controller import producer

"""
Author          : Neda Peyrone
Create Date     : 30-08-2023
File            : app.py
"""

def create_app():
  path = Path().absolute()
  Bootstrap(path)

  app = Flask(__name__)

  template = {
    "swagger": "2.0",
    "info": {
      "title": "Flask Kafka API",
      "description": "This API was developed using Python Flask, which provides an interface for producing and consuming messages with Apache Kafka topics via HTTP endpoints.",
      "version": "1.0"
    }
  }
  app.config['SWAGGER'] = {
    'title': 'Flask Kafka API',
    'uiversion': 2,
    'template': './resources/flasgger/swagger_ui.html'
  }
  Swagger(app, template=template)

  app.register_blueprint(consumer)
  app.register_blueprint(producer)

  return app


if __name__ == '__main__':
  try:
    port = int(sys.argv[1])
  except Exception:
    port = 8081

  create_app().run(host='0.0.0.0', port=port, debug=True, use_reloader=True)

4. Create the “consume.yml” OAS file in YAML format for the consume method, following the specified guidelines

tags:
  - Consumer
parameters:
  - in: "body"
    name: "body"
    description: "Consume a message from a specific Kafka topic"
    required: true
    schema:
      $ref: "#/definitions/ConsumerRequest"
definitions:
  ConsumerRequest:
    type: "object"
    properties:
      group_id:
        type: "string"
        example: "mygroup"
      topic:
        type: "string"
        example: "hello-kafka"
      config:
        type: "object"
        additionalProperties: {}
        example: { "auto.offset.reset": "earliest" }
  ConsumerResponse:
    type: "object"
    properties:
      body:
        type: "object"
        example: "Hello World!"
      statusResponse:
        type: "object"
        properties:
          code:
            type: "string"
          desc:
            type: "string"
        example:
          code: SUCCESS
          desc: Service Success.
responses:
  200:
    description: "Successful response"
    schema:
      $ref: "#/definitions/ConsumerResponse"

5. Add the consume.yml file to the consume method’s route, following the code below

from flasgger import swag_from
from flask import request, Blueprint
from marshmallow import ValidationError
from api.domain.consumer import Consumer
from api.util import exception_util, service_util
from api.constant.message_code import MessageCode
from api.schema.consumer_schema import ConsumerSchema
from api.exception.service_exception import ServiceException
from api.business_logic.consumer_service import ConsumerService

consumer = Blueprint('consumer', __name__)

"""
Author          : Neda Peyrone
Create Date     : 30-08-2023
File            : consumer_controller.py
"""

@swag_from('../../docs/consume.yml')
@consumer.route("/consume", methods=['POST'])
def consume():
  payload = request.get_json()

  try:
    validated_data = ConsumerSchema().load(payload)
    consumer = Consumer(**validated_data)

    msg = ConsumerService().consume(consumer)
    return service_util.build_server_response(MessageCode.SUCCESS, msg)
  except (ValidationError, ServiceException) as err:
    return exception_util.handler(err)

6. Start the flask-kafka project using the following command:

python app.py

7. Access the Swagger UI by navigating to the following URL in a web browser:

http://localhost:8081/apidocs/

Now, you should be able to view the Swagger UI, which uses the YAML file to document the method routes in the Flask application.

The available RESTful API methods of the flask-kafka project as depicted in Swagger UI.

I hope the above steps are helpful in setting up Swagger UI in Python Flask.