medium-crub-app-create.py

@app.route("/create", methods=["POST"])
def create_book():
    """"
    Create a new book
    """
    payload = request.get_json()
    if "id" in payload:    # user cannot pass id when creating a new book
        payload.pop("id")
    status = BookModel.Schema().validate(payload, partial=("id",))   # no validation to id
    if status:
        return jsonify(status), 400
    book = BookModel.from_dict(payload)
    db_create_book(book)
    return jsonify(data=book.to_dict()), 201

def db_create_book(book: BookModel) -> bool:
    TBook.insert_one(book.__dict__) # 

medium-crub-app-delete.py

@app.route("/delete", methods=["DELETE"])
def delete_book():
    """"
    Delete a book
    """
    book_id = request.args.get("id")
    if not book_id:
        return jsonify({"message": "Book id is required"}), 400
    success = db_delete_book(book_id)
    if not success:
        return jsonify({"message": "Book Delete failed"}), 404
    return jsonify({"message": "Book Deleted"}), 200
  
 def db_delete_book(book_id: str) -> bool:
    res = TBook.delete_one({"id": book_id})
    return res.deleted_count > 0

medium-crub-app-list.py

@app.route("/list", methods=["GET"])
def list_book():
    """"
    Retrieve all books
    """
    books = db_list_books()
    res = {"list": [book.to_dict() for book in books], "count": len(books)}
    return jsonify(data=res), 200
  
def db_list_books() -> List[BookModel]:
    return [BookModel.from_dict(r) for r in TBook.find()]

medium-crub-app-retrieve.py

@app.route("/retrieve", methods=["GET"])
def retrieve_book():
    """"
    Retrieve a book
    """
    book_id = request.args.get("id")
    book = db_retrieve_book(book_id)
    if book:
        return jsonify(data=book.to_dict()), 200
    return jsonify({"message": "Book not found"}), 404

def db_create_book(book: BookModel) -> bool:
    TBook.insert_one(book.__dict__)

medium-crub-app-update.py

@app.route("/update", methods=["POST"])
def update_book():
    """"
    Update a book
    """
    payload = request.get_json()
    status = BookModel.Schema().validate(payload)
    if status:
        return jsonify(status), 400
    book_id = payload.get("id")
    if not db_retrieve_book(book_id):
        return jsonify({"message": "Book not found"}), 404

    success = db_update_book(book_id, payload)
    if not success:
        return jsonify({"message": "Book Update failed"}), 404
    book_db = db_retrieve_book(book_id)
    return jsonify(data=book_db.to_dict()), 200
    
 def db_update_book(book_id: str, book: Dict[str, Any]) -> bool:
    res = TBook.update_one({"id": book_id}, {"$set": book})
    return res.modified_count > 0

medium-crub-dataclasses-model2.py

from dataclasses import field, dataclass
from datetime import datetime

from typing import Optional

@dataclass
class BookModel:
    id: int
    author: Optional[str]
    title: str
    year: Optional[int]
    genre: Optional[str]
    description: Optional[str]
    image: Optional[str]
    rating: Optional[float]

    created_at: datetime = field(default_factory=datetime.utcnow)

medium-crub-dataclasses-model3.py

@dataclass_json(undefined=Undefined.EXCLUDE)
@mm_dataclass(frozen=True)
class BookModel:
    author: Optional[str]
    title: str = field(metadata={"validate": validate.Length(min=1, max=256)})
    year: Optional[int]
    genre: Optional[int]
    description: Optional[str]
    image: Optional[str]
    rating: Optional[float]

    id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
    created_at: datetime = field(metadata={
        'dataclasses_json': {
            'encoder': lambda x: datetime.timestamp(x),
        }
    }, default_factory=datetime.utcnow)

medium-crub-db.py

import pymongo
from model import *
from typing import List, Dict, Any

client = pymongo.MongoClient("mongodb://localhost:27017/")
TDb = client["data"]
TBook = TDb['book']