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:
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.
Andrea Romagnoli 
Leave a Reply