What You’ll Learn

• How to integrate OpenAPI into a Flask application with the Python Connexion library
• How to test your OpenAPI Specification with Swagger UI

What You’ll Need

• Python 3.6+, with Flask and Connexion (with swagger-ui extra) libraries installed. This tutorial was created using Flask version 2.2.2 and Connexion version 2.14.1, which can be installed with the commands pip install Flask==2.2.2 and pip install "connexion[swagger ui]==2.14.1". You might have to use a slightly different command, such as replacing pip with pip3, depending on your environment setup. Although more recent versions of both Flask and Connexion exist, it is advised that you use the same versions that were used to create the tutorial; using different versions could lead to incompatibilities.
• A code editor, such as Visual Studio Code

In the tutorial Flask and SQLite: Consuming API Endpoints with Python Requests, you learned how to create a Flask web application, store its data in an SQLite database, and interact with the application using the Python Requests library. If you have not viewed it yet, it might be helpful to do so before continuing.

Although Python Requests is a well-known tool for consuming application programming interface (API) endpoints, it is not so well suited for determining all the capabilities of the API. It is standard practice to provide documentation when building an API, and the OpenAPI Specification (OAS) is a popular method for doing just that. The tools that are used to configure OpenAPI are often referred to as “Swagger” tools. The reason for this is that they were developed by the team behind the original Swagger Specification, upon which OpenAPI is built.

Although there are a number of Python libraries that support OpenAPI usage in Flask, like Flask-RESTX, this tutorial uses the Connexion library. Connexion features Swagger UI console with live documentation and “try it out” features. Make sure that you include the swagger-ui extra when you install Connexion so that Swagger UI is available for your API.

As a starting point, you will run the Flask application from the tutorial mentioned above. Then, you will modify the application so that it supports the OAS. Although the application contains several API endpoints that can be consumed and tested, the focus of this tutorial is the integration of OpenAPI with the application using Swagger UI. If you require more detailed information about the application, it is recommended that you view the previous tutorial.

The code for the initial application setup is included in the following steps.

In this step, you are going to initialize the SQLite database, then verify its contents using sqlite3.

Create the Schema

Create a file named db_schema.sql in a folder in your Python 3.6+ environment, and add the following code:

DROP TABLE IF EXISTS records;

CREATE TABLE records (
    id INTEGER PRIMARY KEY,
    fname TEXT NOT NULL,
    lname TEXT NOT NULL,
    role TEXT NOT NULL
);

Create the Database

Create a script called new_db.py in the same folder, and add the following code:

import sqlite3

con = sqlite3.connect('employees.db')

with open('db_schema.sql') as f:
    con.executescript(f.read())

cur = con.cursor()

cur.execute("INSERT INTO records (fname, lname, role) VALUES (?, ?, ?)",
            ('Aisha', 'Kumar', 'Product Manager')
            )

cur.execute("INSERT INTO records (fname, lname, role) VALUES (?, ?, ?)",
            ('Qian', 'Chen', 'Cloud Engineer')
            )

cur.execute("INSERT INTO records (fname, lname, role) VALUES (?, ?, ?)",
            ('Kamali', 'Keita', 'Data Analyst')
            )

con.commit()
con.close()

Open an OS command prompt and run the script:

python new_db.py

When running the script, you may have to type python3 or something similar, depending on your environment setup. For simplicity, this tutorial will continue to refer to python when running scripts. Once you have run the script, verify that the database file employees.db has been created in the same folder.

Query the Database

Next, you are going to examine the contents of the database using sqlite3.

Start a Python Read Evaluate Print Loop (REPL) session from the same folder that you created the database in, and enter the following code:

import sqlite3

con = sqlite3.connect('employees.db')
cur = con.cursor()
records = cur.execute('select * from records').fetchall()

for record in records:
    print(f'Record {record[0]}\n\tFirst name: {record[1]}\n\tLast name: {record[2]}\n\tRole: {record[3]}\n')

The output should look like this:

In this step, you are going to deploy the web application, then verify that it works.

Create the Application

Create a script called employee.py, and add the following code:

import sqlite3
from flask import Flask, jsonify, request

app = Flask(__name__)


def db_connect():
    con = sqlite3.connect('employees.db')
    return con


@app.route('/')
def welcome():
    return '<h1>Welcome to the Employee Records database</h1>'


@app.route('/employee', methods=['GET'])
def get_emps():
    con = db_connect()
    cur = con.cursor()
    rows = cur.execute('select * from records').fetchall()
    con.close()
    return jsonify(rows)


@app.route('/employee/<empid>', methods=['GET'])
def get_emp(empid):
    con = db_connect()
    cur = con.cursor()
    row = cur.execute('select * from records WHERE id = ?', empid).fetchone()
    con.close()
    return jsonify(row)


@app.route('/employee/', methods=['POST'])
def add_emp():
    con = db_connect()
    cur = con.cursor()
    new_emp = request.get_json()
    id = new_emp['id']
    fname = new_emp['fname']
    lname = new_emp['lname']
    role = new_emp['role']
    cur.execute('INSERT INTO records (id, fname, lname, role) VALUES (?, ?, ?, ?)', (id, fname, lname, role))
    con.commit()
    con.close()
    return new_emp


@app.route('/employee/<empid>', methods=['PUT'])
def put_emp(empid):
    con = db_connect()
    cur = con.cursor()
    emp = cur.execute('SELECT * FROM records WHERE id = ?', (empid,)).fetchone()
    fname = emp[1]
    lname = emp[2]
    role = emp[3]
    update_emp = request.get_json()
    if 'fname' in update_emp:
        fname = update_emp['fname']
    if 'lname' in update_emp:
        lname = update_emp['lname']
    if 'role' in update_emp:
        role = update_emp['role']

    cur.execute('UPDATE records SET fname = ?, lname = ?, role = ?'' WHERE id = ?', (fname, lname, role, empid))
    con.commit()
    con.close()
    return f'Record {empid} was successfully updated!'


@app.route('/employee/<empid>', methods=['PATCH'])
def mod_emp(empid):
    con = db_connect()
    cur = con.cursor()
    emp = cur.execute('SELECT * FROM records WHERE id = ?', (empid,)).fetchone()
    fname = emp[1]
    lname = emp[2]
    role = emp[3]
    update_emp = request.get_json()
    if 'fname' in update_emp:
        fname = update_emp['fname']
    if 'lname' in update_emp:
        lname = update_emp['lname']
    if 'role' in update_emp:
        role = update_emp['role']

    cur.execute('UPDATE records SET fname = ?, lname = ?, role = ?'' WHERE id = ?', (fname, lname, role, empid))
    con.commit()
    con.close()
    return f'Record {empid} was successfully updated!'


@app.route('/employee/<empid>', methods=['DELETE'])
def del_emp(empid):
    con = db_connect()
    cur = con.cursor()
    cur.execute('DELETE FROM records WHERE id = ?', (empid,))
    con.commit()
    con.close()
    return f'Record {empid} was successfully deleted!'


@app.route('/employee/find', methods=['GET'])
def find_emp():
    con = db_connect()
    cur = con.cursor()
    if request.args.get('id'):
        emp_id = request.args.get('id')
        row = cur.execute('select * from records WHERE id = ?', (emp_id,)).fetchall()
    elif request.args.get('fname'):
        emp_fname = request.args.get('fname')
        row = cur.execute('select * from records WHERE fname = ?', (emp_fname,)).fetchall()
    elif request.args.get('lname'):
        emp_lname = request.args.get('lname')
        row = cur.execute('select * from records WHERE lname = ?', (emp_lname,)).fetchall()
    elif request.args.get('role'):
        emp_role = request.args.get('role')
        row = cur.execute('select * from records WHERE role = ?', (emp_role,)).fetchall()
    con.close()
    return jsonify(row)


if __name__ == "__main__":
    app.run(debug=True, port=8000)

There is a slight difference between this script and the one that it originates from. There are separate routes for the patch and put methods. The reason is that a combined route for put and patch can lead to operational errors with an OAS, so it is simpler to keep them separate.

Start the Script

Open an OS command prompt and run the script:

python employee.py

The application runs on a simple, built-in web server that is sufficient for testing but not suitable for use in production. The server listens on port 5000 by default, although it has been changed to port 8000 in this tutorial. You may need to change it if port 8000 is already in use in your environment. The server output looks like this:

Open a browser and point it to http://localhost:8000 to view one of the resources. It should look like this:

Point your browser to another resource: http://localhost:8000/employee; this returns the same three user records that were returned when querying the database in the previous step. Of course, you could interact with the other resources using Python Requests, or an alternative REST client, but the focus of this tutorial is to develop documentation for the API.

To stop the server, press CTRL+C at the prompt where you started it.

In this step, you are going to create an API configuration file for your web application.

Connexion is a modern Python web framework that facilitates spec-first and API-first development. You describe your API in an OAS (or Swagger Specification), and Connexion will guarantee that it works as you specified. Connexion allows you to create a user interface (Swagger UI) to provide live documentation and “try it out” features for the API.

With Connexion, you write your API specification first and automatically get a lot of other functionality. With many other Python web frameworks, you write your functionality first and automatically get your specification.

Create the API Configuration File

The Swagger configuration file can be written in YAML or JSON, and it contains your OpenAPI definitions. It allows you to describe your entire API, including the available endpoints and operation parameters.

Create a file called swagger.yml, and begin adding metadata to it:

openapi: 3.0.0
info:
    title: "Employee Records REST API"
    description: "An API for consuming employee records"
    version: "1.0.0"

servers:
    - url: "/api"

Every API definition must include the version of the OAS that it is based on.

The info section contains your API name (title), optional description and also its version (not to be confused with the openapi version). The servers section specifies the API server and base URL. The default value for the server is localhost, and all API paths are relative to it.

Add an Endpoint to the API Configuration File

Add the following metadata to the end of your swagger.yml file:

components:
  schemas:
    Person:
      type: "object"
      #required:
      #  - id
      properties:
        id:
          type: "integer"
        fname:
          type: "string"
        lname:
          type: "string"
        role:
          type: "string"
  parameters:
    id:
      name: "id"
      description: "Employee identity"
      in: path
      required: True
      schema:
        type: "integer"

paths:
    /employees:
        get:
            operationId: "employee.get_emps"
            tags:
                - "Employees"
            summary: "View employee records"
            responses:
                "200":
                    description: "Successfully viewed employee records!" 

The global components and schemas section lets you define common data structures that are used in your API, such as the id, fname, lname, and role of the records that are stored in the employees database.

The paths section defines individual endpoints (paths) in your API and the HTTP methods (operations) that are supported by these endpoints. In the code above, you added a single endpoint—GET /employees. You will add more endpoints in a later step.

Currently, your Flask application does not know about your swagger.yml file. You will configure this integration in the next step.

Now that you have created the swagger.yml configuration file, the next step is to integrate it with your Flask application.

Integrate Swagger UI with the Flask Application

Create a file called conn_app.py, and add the following code:

import connexion

app = connexion.App(__name__, specification_dir="./")
app.add_api("swagger.yml")

@app.route('/')
def welcome():
    return "<h1>Welcome to the new Employee Records database</h1>"


if __name__ == "__main__":
   app.run(host="0.0.0.0", port =8000, debug=True)

The connexion module is imported, and then the application is created as an instance of Connexion rather than Flask. The API that is defined in swagger.yml is registered to the application, providing additional functionality.

Open an OS command prompt and run the script:

python con_app.py

Open a browser and point it to http://localhost:8000/api/ui. It should look like this:

Examine the API User Interface

Notice the name Swagger in the top left-hand corner of the screen. Just below it is the API name that you defined in swagger.yml. Change it to My Employee Record REST API, and refresh the browser to view the change. Click the GET /employees endpoint to expand it. The screen should look like this:

Select Try it out, followed by Execute. (You may have to scroll down to view the server response.) This returns the same three employee records that we viewed in earlier steps. In the bottom right-hand corner of the response, there is an option to download the results in JSON format. The screen also includes an equivalent command for retrieving results with the popular curl tool.

Add the Remaining Endpoints

For convenience, the completed swagger.yml file is shown in the following code. You can either replace the entire contents of your swagger.yml or copy the remaining endpoints, whichever works best for you.

openapi: 3.0.0
info:
  title: "My Employee Records REST API"
  description: "An API for consuming employee records"
  version: "1.0.0"

servers:
  - url: "/api"

components:
  schemas:
    Person:
      type: "object"
      #required:
      #  - id
      properties:
        id:
          type: "integer"
        fname:
          type: "string"
        lname:
          type: "string"
        role:
          type: "string"
  parameters:
    id:
      name: "id"
      description: "Employee identity"
      in: path
      required: True
      schema:
        type: "integer"

paths:
  /employees:
    get:
      operationId: "employee.get_emps"
      tags:
        - "Employees"
      summary: "View employee records"
      responses:
        "200":
          description: "Successfully viewed employee records!"
    post:
      operationId: "employee.add_emp"
      tags:
        - Employees
      summary: "Add employee record"
      requestBody:
        description: "Enter values for id, fname, lname and role - id must be unique"
        required: True
        content:
          application/json:
            schema:
              x-body-name: "person"
              $ref: "#/components/schemas/Person"
      responses:
        "201":
          description: "Successfully added employee!"

  /employees/{id}:
    get:
      operationId: "employee.get_emp"
      tags:
        - Employees
      summary: "View employee record"
      parameters:
        - $ref: "#/components/parameters/id"
      responses:
        "200":
          description: "Successfully viewed employee!"
    patch:
      operationId: "employee.mod_emp"
      tags:
        - Employees
      summary: "Update employee record"
      parameters:
        - $ref: "#/components/parameters/id"
      requestBody:
        description: "At least one <key:value> pair is needed - remove <id:value> & other unneeded <key:value> pairs."
        required: True
        content:
          application/json:
            schema:
              x-body-name: 'person'
              $ref: "#/components/schemas/Person"
      responses:
        "200":
          description: Successfully updated employee!"
    put:
      operationId: "employee.put_emp"
      tags:
        - Employees
      summary: "Update employee record"
      parameters:
        - $ref: "#/components/parameters/id"
      requestBody:
        description: "At least one <key:value> pair is needed - remove <id:value> & other unneeded <key:value> pairs."
        required: True
        content:
          application/json:
            schema:
              x-body-name: 'person'
              $ref: "#/components/schemas/Person"
      responses:
        "200":
          description: Successfully updated employee!"
    delete:
      operationId: "employee.del_emp"
      tags:
        - Employees
      summary: "Delete employee record"
      parameters:
        - $ref: "#/components/parameters/id"
      responses:
        "200":
          description: "Successfully deleted employee!"

  /employees/find:
    get:
      operationId: "employee.find_emp"
      tags:
        - Employees
      summary: "Find employee record"
      parameters:
        - in: query
          name: id
          schema:
            type: integer
          description: Identity ( if you enter a value, leave the other 3 empty )
        - in: query
          name: fname
          schema:
            type: string
          description: First name ( if you enter a value, leave the other 3 empty )
        - in: query
          name: lname
          schema:
            type: string
          description: Last name ( if you enter a value, leave the other 3 empty )
        - in: query
          name: role
          schema:
            type: string
          description: Role ( if you enter a value, leave the other 3 empty )
      responses:
        "200":
          description: "Successfully found employee!"

Review the Other Endpoints

Refresh your browser; you should still be pointing it to http://localhost:8000/api/ui. It should look like this:

Examine the other endpoints as you see fit. As mentioned earlier, the focus of this tutorial is the integration of OpenAPI with your Flask application through the use of Swagger UI. Further details on endpoint usage are covered in the previous tutorial.

Congratulations! You have successfully completed this tutorial.

You learned:

• How to deploy a Flask application that integrates with an SQLite database.
• How to reconfigure the application so that it supports the OAS
• How to document and test the endpoints using Swagger UI

Learn More

• Connexion
• Flask
• Swagger UI

Continue Your Education

• DEVCOR Learning Path
• Understanding Exceptions and Exception Handling
• Working with Data Structures in Network Automation Scripts

Further Learning Resources

• Inside Cisco U.: Step-by-Step Guide to Learning Paths, Courses, Labs, and Tutorials

Need Help or Want to Engage?

• To ask questions and share ideas, join our Cisco Learning Community.

• For technical issues, feedback, or more resources, visit our Cisco U. Support page.

Finishing Up

Don’t forget to click Exit Tutorial to log your completed content.