Create API Model for Dict type data

[muhammad-rafi]

Hi I have been trying to figure out to create a api.model for the following data, I am using in my flask-restx api app. I have tried few options including nested one but nothing is showing up as per I need.

{
    "sandbox-iosxe-recomm-1.cisco.com": {
        "username": "developer", 
        "password": "cisco", 
        "protocol": "ssh",
        "port": 22,
        "os_type": "iosxe"
        },
    "sandbox-iosxr-1.cisco.com": {
        "username": "admin", 
        "password": "cisco", 
        "protocol": "ssh",
        "port": 22,
        "os_type": "iosxr"
}

Please let me know if anyone can help ?

[peter-doggart]

There are two "easy" ways of doing this, depending on if the web domains are changeable or not. If they are fixed as in your example as sandbox-iosxe-recomm-1.cisco.com and sandbox-iosxr-1.cisco.com you can create a model using the example below:

from flask import Flask, Blueprint
from flask_restx import Api, Resource, fields

api_v1 = Blueprint("api", __name__, url_prefix="/api")

api = Api(
    api_v1,
    version="1.0",
    title="Dummy API",
    description="A simple Dummy API",
)

ns = api.namespace("Dummy", description="Dummy operations")

container = api.model(
    "container", {
        "username": fields.String(default='username'),
        "password": fields.String(default='password'),
        "protocol": fields.String(default='ssh'),
        "port": fields.Integer(default=22),
        "os_type": fields.String(default='os_type'),
    }
)

main_model = api.model(
    "main", {
        "sandbox-iosxe-recomm-1.cisco.com": fields.Nested(container),
        "sandbox-iosxr-1.cisco.com": fields.Nested(container)
    }
)

@ns.route("/test")
class TodoList(Resource):
    """Shows a list of all todos, and lets you POST to add new tasks"""

    @api.expect(main_model)
    def post(self):
        """Dummy Method"""
        return "Success", 200

if __name__ == "__main__":
    app = Flask(__name__)
    app.register_blueprint(api_v1)
    app.run(debug=True)

NOTE: For nested models, you need to define the api.expect on the method not on the resource class for it to render correctly.

You should get a Swagger UI that looks like this:

If you don't know the domain names ahead of time, it's a little more complex because flask-restx doesn't support wildcard fields that contain nested models. If that is the case, you are probably better to pass a list of connection models and include the domain like this:

from flask import Flask, Blueprint
from flask_restx import Api, Resource, fields

api_v1 = Blueprint("api", __name__, url_prefix="/api")

api = Api(
    api_v1,
    version="1.0",
    title="Dummy API",
    description="A simple Dummy API",
)

ns = api.namespace("Dummy", description="Dummy operations")

container_2 = api.model(
    "container2", {
        "domain": fields.String(default='*.cisco.com'),
        "username": fields.String(default='username'),
        "password": fields.String(default='password'),
        "protocol": fields.String(default='ssh'),
        "port": fields.Integer(default=22),
        "os_type": fields.String(default='os_type'),
    }
)

main_model_2 = api.model(
    "main_2", {
        "connection_details": fields.List(fields.Nested(container_2))
    }
)

@ns.route("/test")
class TodoList(Resource):
    """Shows a list of all todos, and lets you POST to add new tasks"""

    @api.expect(main_model_2)
    def put(self):
        """Dummy Method"""
        return "Success", 200

if __name__ == "__main__":
    app = Flask(__name__)
    app.register_blueprint(api_v1)
    app.run(debug=True)

Which will give you something like this:

You would then obviously need to verify you got the correct number of connection details manually.

[muhammad-rafi]

Thanks @peter-doggart

Here is what I am doing to get the current dict of devices, notice I am not using list but rather dictionary of dictionaries. I am also using marshal_with to make the consistent output, I will also add post, put and delete once this works OK. Please advise.

from flask import Flask, Blueprint
from flask_restx import Api, Resource, fields

api_v1 = Blueprint("api", __name__, url_prefix="/api/v1")

api = Api(
    api_v1,
    version="1.0",
    title="Cisco Sandbox APIs",
    description="A simple Devices API",
    doc = '/docs'
)

ns = api.namespace("Sandbox Devices APIs", description="REST API Operations")

devices= {
    "sandbox-iosxe-recomm-1.cisco.com": {
        "username": "admin", 
        "password": "cisco", 
        "protocol": "ssh",
        "port": 22,
        "os_type": "iosxe"
        },
    "sandbox-iosxr-1.cisco.com": {
        "username": "admin", 
        "password": "cisco", 
        "protocol": "ssh",
        "port": 22,
        "os_type": "iosxr"
        },
    "sandbox-nxos-1.cisco.com": {
        "username": "admin", 
        "password": "cisco", 
        "protocol": "ssh",
        "port": 22,
        "os_type": "nxos"
        }
      }

device_nested = api.model(
    "device", {
        "username": fields.String(default='username'),
        "password": fields.String(default='password'),
        "protocol": fields.String(default='ssh'),
        "port": fields.Integer(default=22),
        "os_type": fields.String(default='os_type'),
    }
)

device_model = api.model(
    "devices", {
        "device": fields.Nested(device_nested),
    }
)

# @ns.route("/devices")
class Devices(Resource):
    """Shows a list of current devices, and lets you POST to add new device(s)"""

    @api.marshal_with(device_model, code=200, description="List of device")
    def get(self):
        """Get the current devices"""
        return {"devices": devices}

api.add_resource(Devices, '/devices', endpoint='Devices')

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

As you can see from the above output, it doesn't show me the list of but rather model, if you copy this code and run, you will probably better understanding.

[peter-doggart]

I think you are probably best swapping to a list of devices, where each device is a fully self-contained dictionary which allows you to use marshal_list_with like:

from flask import Flask, Blueprint
from flask_restx import Api, Resource, fields

api_v1 = Blueprint("api", __name__, url_prefix="/api/v1")

api = Api(
    api_v1,
    version="1.0",
    title="Cisco Sandbox APIs",
    description="A simple Devices API",
    doc = '/docs'
)

ns = api.namespace("Sandbox Devices APIs", description="REST API Operations")

all_devices = [
    {
        "domain": "sandbox-iosxe-recomm-1.cisco.com",
        "username": "developer", 
        "password": "C1sco12345", 
        "protocol": "ssh",
        "port": 22,
        "os_type": "iosxe",
        "secret_field": "secret"
    },
    {
        "domain": "sandbox-iosxr-1.cisco.com",
        "username": "admin", 
        "password": "C1sco12345", 
        "protocol": "ssh",
        "port": 22,
        "os_type": "iosxr",
        "secret_field": "secret"
    },
    {
        "domain": "sandbox-nxos-1.cisco.com",
        "username": "admin", 
        "password": "Admin_1234!", 
        "protocol": "ssh",
        "port": 22,
        "os_type": "nxos",
        "secret_field": "secret"
    }
]

device_model = api.model(
    "device", {
        "domain": fields.String(default='domain'),
        "username": fields.String(default='username'),
        "password": fields.String(default='password'),
        "protocol": fields.String(default='ssh'),
        "port": fields.Integer(default=22),
        "os_type": fields.String(default='os_type'),
    }
)

# @ns.route("/devices")
class Devices(Resource):
    """Shows a list of current devices, and lets you POST to add new device(s)"""

    @api.marshal_list_with(device_model)
    def get(self):
        """Get the current devices"""
        return all_devices

api.add_resource(Devices, '/devices', endpoint='Devices')

if __name__ == "__main__":
    app = Flask(__name__)
    app.register_blueprint(api_v1)
    app.run(debug=True)

Obviously, if you can't change your internal workings, switching from dictionaries to a list like this (and reverse) is fairly trivial by iterating over the keys/list.

There is no easy way to marshal a dictionary of dictionaries where the keys change over time because when you specify the API models and therefore the swagger.json it has to be in a static way. This means the field names have to be known in advance.

[muhammad-rafi]

@peter-doggart , I listened to your advise and changed it to the list of devices but I have few other issues, I am not sure if I raise new request or you can help here.

Here is my code, if you like to try

from flask import Flask, Blueprint, jsonify
from flask_restx import Resource, Api, fields, reqparse, marshal, model
from devices import devices_list
import json

blueprint = Blueprint("api", __name__, url_prefix="/api/v1")

api = Api(
    blueprint,
    version="1.0",
    title="Cisco Sandbox APIs",
    description="A simple Devices API",
    doc = '/docs'
)

device_model = api.model(
    "device", {
        "id": fields.Integer(required=False, description='id', default='0'),
        "host": fields.String(required=True, description='HostName/IP', default='cml-core01.example.com'),
        'username': fields.String(required=True, description='username', default='admin'),
        'password': fields.String(required=True, description='password', default='mYsEcrEt'),
        'protocol': fields.String(required=True, description='protocol', default='ssh'),
        'port': fields.Integer(required=True, description='port', default=22),
        'os_type': fields.String(required=True, description='os_type', default='iosxe'),
    },
)

response_model = api.model(
    "response",
    {
        "message": fields.String(required=True),
        "status_code": fields.Integer(required=True)
    },
)

def marshal_resp(message, response_code, response_model):
    """ Marshal the api response."""
    return (
        marshal (
            {
            "message": message, 
            "status_code": response_code
             }, 
            response_model
        )
    )

class Devices(Resource):
    """Shows a list of current devices, and lets you POST to add new device(s)"""

    @api.marshal_list_with(device_model, code=200, description="List of device", envelope='devices')
    def get(self):
        """Get the current devices"""
        # return {"devices": devices}
        return devices_list

    @api.marshal_with(response_model)
    @api.expect(device_model, validate=True)
    def post(self):
        """Add a new device"""
        new_device = api.payload
        # del new_device['id']

        for d in devices_list:
            if d.get('host') == new_device['host']:
                return marshal_resp("Conflict, device already exists", 409, response_model)

        new_device['id'] = len(devices_list) + 1
        devices_list.append(new_device)
        return marshal_resp("Device(s) added successfully", 201, response_model)

api.add_resource(Devices, '/devices', endpoint='Devices')

if __name__ == "__main__":
    app = Flask(__name__)
    app.register_blueprint(blueprint)
    app.config['SWAGGER_UI_JSONEDITOR'] = True
    app.run(host='0.0.0.0', port=8000, debug=True)

You can I am using app.config['SWAGGER_UI_JSONEDITOR'] = True, so I was expecting the boxes in Swagger UI but I see the payload box only, see the picture below;

and the other issue is my api.mode, I added a new key called 'id' and I want this ID only shows up on the response of the GET request not in the POST request payload as you can see in my POST function, I am increasing the ID based on the length of the devices in the list.

Please let me know if this make sense and can help out.

Thanks

[peter-doggart]

@muhammad-rafi In the case where you want an ID field in one, but not the other you will need to define another model. What I would suggest is define the model without the id field then use model.clone() functionality to create an additional model with the extra field without having to duplicate everything.

Unfortunately, the SWAGGER_UI_JSONEDITOR was not maintained and was removed in Swagger UI >3 (which flask-restx uses).

[muhammad-rafi]

@peter-doggart , thanks for your advise, I followed your instruction and clone the mode with additional key 'id' but obviously shows at the las which is not a big deal.

For SWAGGER_UI_JSONEDITOR, do we have any alternative for that to achieve same results what we get with this env variable ?

Other issue I notice that, when I try to add multiple devices, it doesn't do anything no error but got 200 but no changes in the devices output

{ "host": "cml-core02.example.com", "username": "admin", "password": "mYsEcrEt", "protocol": "ssh", "port": 22, "os_type": "iosxe" }, { "host": "cml-core03.example.com", "username": "admin", "password": "mYsEcrEt", "protocol": "ssh", "port": 22, "os_type": "iosxe" }

[muhammad-rafi]

the reason I was asking about the alternate for SWAGGER_UI_JSONEDITOR, because i want to hide the password value, if we are using the swagger UI, not sure how to achieve this either. but from API call, it will need to be in the payload, which also doesnt sound good practice

[peter-doggart]

@muhammad-rafi Can you provide your full code if you are having issues with devices being added? In your last block above, devices_list doesn't appear to be defined anywhere.

In terms of passing the password value, doing it in the payload is fine, but you need to send your request using HTTPS if this API is going to be on the public internet. If you don't, regardless of where you put it in the request (even if it's hidden in the SwaggerUI), it will be readable by anyone who intercepts it. Configuring that is outside the scope of just flask-restx though.

[muhammad-rafi]

Here is my app.py

from flask import Flask, Blueprint, jsonify
from flask_restx import Resource, Api, fields, reqparse, marshal, model
from devices import devices_list
import json

blueprint = Blueprint("api", __name__, url_prefix="/api/v1")

api = Api(
    blueprint,
    version="1.0",
    title="Cisco Sandbox APIs",
    description="A simple Devices API",
    doc = '/docs'
)

device_model = api.model(
    "device", {
        "host": fields.String(required=True, description='HostName/IP', default='cml-core01.example.com'),
        'username': fields.String(required=True, description='username', default='admin'),
        'password': fields.String(required=True, description='password', default='mYsEcrEt'),
        'protocol': fields.String(required=True, description='protocol', default='ssh'),
        'port': fields.Integer(required=True, description='port', default=22),
        'os_type': fields.String(required=True, description='os_type', default='iosxe'),
    },
)

device_resp_model = device_model.clone(
            'devices', {
                'id': fields.Integer(required=True, description='id'),
                },
            )

api.models[device_resp_model.name] = device_resp_model

response_model = api.model(
    "response",
    {
        "message": fields.String(required=True),
        "status_code": fields.Integer(required=True)
    },
)

def marshal_resp(message, response_code, response_model):
    """ Marshal the api response."""
    return (
        marshal (
            {
            "message": message, 
            "status_code": response_code
             }, 
            response_model
        )
    )

class Devices(Resource):
    """Shows list of current devices in the database, 
    and lets you POST to add new device(s)"""

    @api.marshal_list_with(device_resp_model, code=200, description="List of device", envelope='devices')
    def get(self):
        """Get the current devices"""
        return devices_list

    @api.marshal_with(response_model)
    @api.expect(device_model, validate=True)
    def post(self):
        """Add a new device"""
        new_device = api.payload

        for d in devices_list:
            if d.get('host') == new_device['host']:
                return marshal_resp("Conflict, device already exists", 409, response_model)

        new_device['id'] = len(devices_list) + 1
        devices_list.append(new_device)
        return marshal_resp("Device(s) added successfully", 201, response_model)

api.add_resource(Devices, '/devices', endpoint='Devices')

if __name__ == "__main__":
    app = Flask(__name__)
    app.register_blueprint(blueprint)
    app.config['SWAGGER_UI_JSONEDITOR'] = True

    app.run(host='0.0.0.0', port=8000, debug=True)

Here is my devices.py

# List of Devices
devices_list = [
    {
        "id": 1,
        "host": "cml-core-rtr01",
        "username": "admin", 
        "password": "admin", 
        "protocol": "ssh",
        "port": 22,
        "os_type": "iosxe",
    },
    {
        "id": 2,
        "host": "cml-core-rtr02",
        "username": "admin", 
        "password": "admin", 
        "protocol": "ssh",
        "port": 22,
        "os_type": "iosxr",
    },
    {
        "id": 3,
        "host": "cml-dist-sw01",
        "username": "admin", 
        "password": "admin!", 
        "protocol": "ssh",
        "port": 22,
        "os_type": "nxos",
    },
    {
        "id": 4,
        "host": "cml-dist-sw02",
        "username": "admin", 
        "password": "admin!", 
        "protocol": "ssh",
        "port": 22,
        "os_type": "nxos",
    }
]

I put the devices_list in different file above in devices.py and import in my app.py from devices import devices_list Please let me know if you need more info ?

[muhammad-rafi]

curl -X 'POST' \
  'http://127.0.0.1:8000/api/v1/devices' \
  -H 'accept: application/json' \
  -H 'Content-Type: application/json' \
  -d '{
  "devices": [
    {
      "host": "cml-core01.example.com",
      "username": "admin",
      "password": "mYsEcrEt",
      "protocol": "ssh",
      "port": 22,
      "os_type": "iosxe"
    },
    {
      "host": "cml-core02.example.com",
      "username": "admin",
      "password": "mYsEcrEt",
      "protocol": "ssh",
      "port": 22,
      "os_type": "iosxe"
    }
  ]
}'

This what I want to do to post multiple devices to be added.