Access to MongoDB using Flask

Flask is a microframework written in Python, that allows you to handle RESTful requests to interact with your system. If you are reading this post you probably already know what I am talking about.

We will see how you can connect Flask to a MongoDB. If you want to have your own MongoDB up and running to follow the rest of the tutorial, you might interested to read this old blog post: MongoDB quickstart using Docker and Robo 3T. If you already have an up and running instance of MongoDB, you are ready to start!

Flask hello world

Create your own virtual environment, and then install the needed packages:

pip install flask
pip install flask_pymongo
pip install flasgger

Create your own file hello_world.py and insert the following content:

from flask import Flask

app = Flask(__name__)

@app.route('/')
def hello():
  return 'Hello World!'

if __name__=='__main__':
  app.run()

Run it using:

python hello_world.py

Then connect to the URL http://127.0.0.1:5000/ with your favourite browser and you will receive the message Hello World! as output.

We have instantiated the Flask class, run it throgh the main function, and defined a route to the root /. Then, we defined a function that will be called as soon as someone goes to the root path on the IP and port specified, in that case the default ones.

Good! We are ready for a full example, using Swagger and MongoDB.

Customer registry management

In order to plan our first complex example, we will need nothing but the setup we did during the first chapter and a MongoDB already configured. If you need help to configure MongoDB, you can check one of my older posts where I explained how to use MongoDB and configure it using Docker.

From the theorical point of view, I should spend a few words about Swagger. It is a system to document and test your REST APIs, it is a system often called an “APIs playground”. Instead of using external tools like REST client, postman, so on and so forth, you can directly test your APIs using this tools, that is also quite useful as documentation.

The newest API framework FastAPI generates automatically all the documentation you need (that is a really cool thing!). Using Flask, unfortunately, you have to write the documentation on your own. This is not a huge cons in my opinion, because it allows you to have a more deep control of your code.

Fortunately, Swagger follows the OpenAPI specifications, making really easy to create a documentation in a really quick way. You can decide to use the format you want, for example JSON. In this context I am going to use YAML, because it is more synthetic for configuration purpose. For learning more about how to create the OpenAPI specifications, Swagger created a really awesome guide for you to follow.

Create your own MongoDB, it should be accessible at localhost:27017 (that is the default). It should also have a database named flask and a collection called user. Now you are good to go, and if you copy-paste the following code and OpenAPI specifications, everything should go well. Please remember to put the specifications in a specifications directory that is near the Python script.

In this example, I show you some examples telling you how to interact with MongoDB using three methods:

  • GET: allows you to get an user you inserted previously, specifying its ID.
  • POST: allows you to insert a new user to your document store (that is MongoDB).
  • DELETE: allows you to delete an user that you have previously inserted in your document store.

flask_user.py

from flasgger import Swagger, swag_from
from flask import Flask, request, jsonify, abort
from flask_pymongo import PyMongo, DESCENDING

# Initialization

app = Flask(__name__)
app.config['MONGO_URI'] = 'mongodb://localhost:27017/flask'
app.config['MONGO_DBNAME'] = 'flask'
mongo = PyMongo(app)
swagger = Swagger(app)

# REST APIs

@app.route('/user/<int:id>', methods=['GET'])
@swag_from('specifications/user_get.yml', methods=['GET'])
def user_get(id):
    payload = {}

    for user in mongo.db.user.find({'id': id}):
        if payload:
            abort(409)
        payload = {
            'id': user['id'],
            'name': user['name'],
            'surname': user['surname'],
            'age': user['age']
        }

    if not payload:
        abort(404)

    return jsonify(payload)


@app.route('/user', methods=['POST'])
@swag_from('specifications/user_post.yml', methods=['POST'])
def user_post():
    before = mongo.db.user.count()

    try:
        id = mongo.db.user.find_one(sort=[('id', DESCENDING)]).get('id')
    except AttributeError:
        id = None

    mongo.db.user.insert({
        'id': id + 1 if id else 1,
        'name': request.json.get('name'),
        'surname': request.json.get('surname'),
        'age': request.json.get('age')
    })

    return jsonify({'insert_count': mongo.db.user.count() - before})


@app.route('/user/<int:id>', methods=['DELETE'])
@swag_from('specifications/user_delete.yml', methods=['DELETE'])
def user_delete(id):
    before = mongo.db.user.count()
    mongo.db.user.delete_one({'id': int(id)})
    return jsonify({'delete_count': before - mongo.db.user.count()})


if __name__=='__main__':
    app.run(debug=True)

specifications/user_get.yml

Returns a single user
---
summary: Returns a single user
parameters:
  - in: path
    name: id
    required: true
    description: ID of the user
    schema:
      type: integer
definitions:
  User:
    type: object
    properties:
      id:
        type: integer
      name:
        type: string
      surname:
        type: string
      age:
        type: integer
responses:
  200:
    description: Requested user
    schema:
      $ref: '#/definitions/User'

specifications/user_post.yml

Insert a new user
---
summary: Insert a new user
parameters:
  - in: body
    name: user
    description: User data
    schema:
      type: object
      required:
        - name
        - surname
        - age
      properties:
        name:
          type: string
          description: Name of the user
        surname:
          type: string
          description: Surname of the user
        age:
          type: integer
          description: Age of the user
responses:
  200:
    description: Total number of inserted users
    schema:
      type: object
      properties:
        insert_count:
          type: integer

specifications/user_delete.yml

Delete the user with the specified ID
---
summary: Delete a user
parameters:
  - name: id
    in: path
    schema:
      type: integer
    required: true
    description: ID of the user to delete
responses:
  200:
    description: Total number of deleted users
    schema:
      type: object
      properties:
        delete_count:
          type: integer

Now run the application:

python flask_user.py

Open your favourite browser (this is mine), and connect to http://127.0.0.1:5000/. There should appear the Swagger API playground, something like that:

Screenshot showing an example of the Swagger API interface

Here you go! You can try your APIs using the Swagger interface. Click on the HTTP method, press the Try it out button, and interact with your MongoDB instance.

Conclusions

I hope you learned the basics about how to interact with MongoDB using Python, and specifically Flask. You have also learned what it is a Swagger, and how can be used to test your REST APIs.

Leave a Reply

Fill in your details below or click an icon to log in:

WordPress.com Logo

You are commenting using your WordPress.com account. Log Out /  Change )

Google photo

You are commenting using your Google account. Log Out /  Change )

Twitter picture

You are commenting using your Twitter account. Log Out /  Change )

Facebook photo

You are commenting using your Facebook account. Log Out /  Change )

Connecting to %s

Website Built with WordPress.com.

Up ↑

%d bloggers like this: