» Python: Build a REST API with Flask » 2. Development » 2.8 Database: MongoDB

Database: mongoDB

If you prefer NoSQL databases, mongoDB is definitely a great one that you don't want to miss.

Try mongoDB

  1. Install mongoDB on your machine and start it.

Note: Remember to create indexes on collections according to your needs in a production project.

  1. Add mongo dependency:
pip3 install pymongo

Update requirements.txt:

pip3 freeze > requirements.txt
  1. Update code.

Use mongoDB for CRUD operations

Add a new domain entity Review in domain/model/review.py:

from dataclasses import dataclass
from datetime import datetime


@dataclass
class Review:
    id: str
    book_id: int
    author: str
    title: str
    content: str
    created_at: datetime
    updated_at: datetime

Declare its business capabilities in domain/gateway/review_manager.py:

from abc import ABC, abstractmethod
from typing import List, Optional

from ..model import Review


class ReviewManager(ABC):
    @abstractmethod
    def create_review(self, r: Review) -> str:
        pass

    @abstractmethod
    def update_review(self, id: str, r: Review) -> None:
        pass

    @abstractmethod
    def delete_review(self, id: str) -> None:
        pass

    @abstractmethod
    def get_review(self, id: str) -> Optional[Review]:
        pass

    @abstractmethod
    def get_reviews_of_book(self, book_id: int) -> List[Review]:
        pass

Implement these methods in infrastructure/database/mongo.py:

from bson.objectid import ObjectId
import dataclasses
from pymongo import MongoClient
from typing import Any, Dict, List, Optional

from ...domain.gateway import ReviewManager
from ...domain.model import Review

COLL_REVIEW = "reviews"


class MongoPersistence(ReviewManager):
    def __init__(self, uri: str, db_name: str):
        self.client = MongoClient(uri)
        self.db = self.client[db_name]
        self.coll = self.db[COLL_REVIEW]

    def create_review(self, r: Review) -> str:
        result = self.coll.insert_one(dataclasses.asdict(r))
        return str(result.inserted_id)

    def update_review(self, id: str, r: Review) -> None:
        new_data = {"title": r.title, "content": r.content,
                    "updated_at": r.updated_at}
        self.coll.update_one({"_id": ObjectId(id)}, {"$set": new_data})

    def delete_review(self, id: str) -> None:
        self.coll.delete_one({"_id": ObjectId(id)})

    def get_review(self, id: str) -> Optional[Review]:
        review_data = self.coll.find_one({"_id": ObjectId(id)})
        if review_data is None:
            return None
        return Review(**_polish(review_data))

    def get_reviews_of_book(self, book_id: int) -> List[Review]:
        reviews_data = self.coll.find({"book_id": book_id})
        return [Review(**_polish(r)) for r in reviews_data]


def _polish(r: Dict[str, Any]):
    r['id'] = str(r['_id'])
    del r['_id']
    return r

Add config items for mongodb in infrastructure/config/config.py:

@@ -10,6 +10,8 @@ class DBConfig:
     user: str
     password: str
     database: str
+    mongo_uri: str
+    mongo_db_name: str
 
 
 @dataclass

Add config values in config.yml:

@@ -7,3 +7,5 @@ db:
   user: "test_user"
   password: "test_pass"
   database: "lr_book"
+  mongo_uri: "mongodb://localhost:27017"
+  mongo_db_name: "lr_book"

Add review_operator in Application layer, application/executor/review_operator.py:

from datetime import datetime
from typing import List, Optional

from ...domain.model import Review
from ...domain.gateway import ReviewManager


class ReviewOperator():

    def __init__(self, review_manager: ReviewManager):
        self.review_manager = review_manager

    def create_review(self, r: Review) -> Review:
        now = datetime.now()
        r.created_at = now
        r.updated_at = now
        id = self.review_manager.create_review(r)
        r.id = id
        return r

    def get_review(self, id: str) -> Optional[Review]:
        return self.review_manager.get_review(id)

    def get_reviews_of_book(self, review_id: int) -> List[Review]:
        return self.review_manager.get_reviews_of_book(review_id)

    def update_review(self, id: str, r: Review) -> Review:
        r.updated_at = datetime.now()
        self.review_manager.update_review(id, r)
        return r

    def delete_review(self, id: str) -> None:
        return self.review_manager.delete_review(id)

Tune application/wire_helper.py to wire in mongodb connections:

@@ -1,16 +1,21 @@
-from ..domain.gateway import BookManager
+from books.domain.gateway import BookManager, ReviewManager
 from ..infrastructure.config import Config
-from ..infrastructure.database import MySQLPersistence
+from ..infrastructure.database import MySQLPersistence, MongoPersistence
 
 
 class WireHelper:
-    def __init__(self, persistence: MySQLPersistence):
-        self.persistence = persistence
+    def __init__(self, sqlPersistence: MySQLPersistence, noSQLPersistence: MongoPersistence):
+        self.sqlPersistence = sqlPersistence
+        self.noSQLPersistence = noSQLPersistence
 
     @classmethod
     def new(cls, c: Config):
         db = MySQLPersistence(c.db)
-        return cls(db)
+        mdb = MongoPersistence(c.db.mongo_uri, c.db.mongo_db_name)
+        return cls(db, mdb)
 
     def book_manager(self) -> BookManager:
-        return self.persistence
+        return self.sqlPersistence
+
+    def review_manager(self) -> ReviewManager:
+        return self.noSQLPersistence

Add review routes in adaptor/router.py:

@@ -1,16 +1,17 @@
 import logging
 from flask import Flask, request, jsonify
 
-from ..application.executor import BookOperator
+from ..application.executor import BookOperator, ReviewOperator
 from ..application import WireHelper
-from ..domain.model import Book
+from ..domain.model import Book, Review
 from .util import dataclass_from_dict
 
 
 class RestHandler:
-    def __init__(self, logger: logging.Logger, book_operator: BookOperator):
+    def __init__(self, logger: logging.Logger, book_operator: BookOperator, review_operator: ReviewOperator):
         self._logger = logger
         self.book_operator = book_operator
+        self.review_operator = review_operator
 
     def get_books(self):
         try:
@@ -56,6 +57,50 @@ class RestHandler:
             self._logger.error(f"Failed to delete: {e}")
             return jsonify({"error": "Failed to delete"}), 404
 
+    def get_reviews_of_book(self, book_id: int):
+        try:
+            reviews = self.review_operator.get_reviews_of_book(book_id)
+            return jsonify(reviews), 200
+        except Exception as e:
+            self._logger.error(f"Failed to get reviews of book: {e}")
+            return jsonify({"error": "Failed to get reviews of book"}), 404
+
+    def get_review(self, id: str):
+        try:
+            review = self.review_operator.get_review(id)
+            if not review:
+                return jsonify({"error": f"The review with id {id} does not exist"}), 404
+            return jsonify(review), 200
+        except Exception as e:
+            self._logger.error(f"Failed to get the review with {id}: {e}")
+            return jsonify({"error": "Failed to get the review"}), 404
+
+    def create_review(self):
+        try:
+            b = dataclass_from_dict(Review, request.json)
+            review = self.review_operator.create_review(b)
+            return jsonify(review), 201
+        except Exception as e:
+            self._logger.error(f"Failed to create: {e}")
+            return jsonify({"error": "Failed to create"}), 400
+
+    def update_review(self, id: str):
+        try:
+            b = dataclass_from_dict(Review, request.json)
+            review = self.review_operator.update_review(id, b)
+            return jsonify(review), 200
+        except Exception as e:
+            self._logger.error(f"Failed to update: {e}")
+            return jsonify({"error": "Failed to update"}), 404
+
+    def delete_review(self, id: str):
+        try:
+            self.review_operator.delete_review(id)
+            return "", 204
+        except Exception as e:
+            self._logger.error(f"Failed to delete: {e}")
+            return jsonify({"error": "Failed to delete"}), 404
+
 
 def health():
     return jsonify({"status": "ok"})
@@ -63,7 +108,7 @@ def health():
 
 def make_router(app: Flask, wire_helper: WireHelper):
     rest_handler = RestHandler(
-        app.logger, BookOperator(wire_helper.book_manager()))
+        app.logger, BookOperator(wire_helper.book_manager()), ReviewOperator(wire_helper.review_manager()))
     app.add_url_rule('/', view_func=health)
     app.add_url_rule('/books', view_func=rest_handler.get_books)
     app.add_url_rule('/books/<int:id>', view_func=rest_handler.get_book)
@@ -73,3 +118,12 @@ def make_router(app: Flask, wire_helper: WireHelper):
                      methods=['PUT'])
     app.add_url_rule('/books/<int:id>', view_func=rest_handler.delete_book,
                      methods=['DELETE'])
+    app.add_url_rule('/books/<int:book_id>/reviews',
+                     view_func=rest_handler.get_reviews_of_book)
+    app.add_url_rule('/reviews/<id>', view_func=rest_handler.get_review)
+    app.add_url_rule('/reviews', view_func=rest_handler.create_review,
+                     methods=['POST'])
+    app.add_url_rule('/reviews/<id>', view_func=rest_handler.update_review,
+                     methods=['PUT'])
+    app.add_url_rule('/reviews/<id>', view_func=rest_handler.delete_review,
+                     methods=['DELETE'])

All the changes are applied now. Let's try these endpoints with curl.

Try with curl

Create a new review:

curl -X POST \
  -H "Content-Type: application/json" \
  -d '{
        "book_id": 1,
        "author": "John Doe",
        "title": "Great Book",
        "content": "This is a great book!"
      }' \
  http://localhost:5000/reviews

It should respond with this:

{
  "author": "John Doe",
  "book_id": 1,
  "content": "This is a great book!",
  "created_at": "Thu, 07 Mar 2024 23:49:04 GMT",
  "id": "65e9e1f020162bf22e98b4a1",
  "title": "Great Book",
  "updated_at": "Thu, 07 Mar 2024 23:49:04 GMT"
}

Fetch a single review by ID:

curl -X GET http://localhost:5000/reviews/65e9e1f020162bf22e98b4a1

Result:

{
  "author": "John Doe",
  "book_id": 1,
  "content": "This is a great book!",
  "created_at": "Thu, 07 Mar 2024 23:49:04 GMT",
  "id": "65e9e1f020162bf22e98b4a1",
  "title": "Great Book",
  "updated_at": "Thu, 07 Mar 2024 23:49:04 GMT"
}

List all reviews of a book:

curl -X GET http://localhost:5000/books/1/reviews

Result list:

[
  {
    "author": "Carl Smith",
    "book_id": 1,
    "content": "This is a great book!",
    "created_at": "Fri, 01 Mar 2024 15:18:24 GMT",
    "id": "65e1f1c0f1c5f50b36b2ce61",
    "title": "Best best Book",
    "updated_at": "Fri, 01 Mar 2024 15:18:24 GMT"
  },
  {
    "author": "John Doe",
    "book_id": 1,
    "content": "This is a great book!",
    "created_at": "Thu, 07 Mar 2024 23:49:04 GMT",
    "id": "65e9e1f020162bf22e98b4a1",
    "title": "Great Book",
    "updated_at": "Thu, 07 Mar 2024 23:49:04 GMT"
  }
]

Update an existing review

curl -X PUT \
  -H "Content-Type: application/json" \
  -d '{
        "content": "I prefer Robert Smith new book",
        "title": "Not that good"
      }' \
  http://localhost:5000/reviews/65e9e1f020162bf22e98b4a1

Result:

{
  "author": null,
  "book_id": null,
  "content": "I prefer Robert Smith new book",
  "created_at": null,
  "id": null,
  "title": "Not that good",
  "updated_at": "Fri, 08 Mar 2024 00:05:12 GMT"
}

Delete an existing review:

curl -X DELETE http://localhost:5000/reviews/65e9e1f020162bf22e98b4a1

It returns code 204 for a sucessful deletion.

Voila! Your api server is using mongoDB as well now!

PrevNext