Database: mongoDB
If you prefer NoSQL databases, mongoDB is definitely a great one that you don't want to miss.
Try mongoDB
- Install mongoDB on your machine and start it.
Note: Remember to create indexes on collections according to your needs in a production project.
- Add mongo dependency:
pip3 install pymongo
Update requirements.txt:
pip3 freeze > requirements.txt
- 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!