チュートリアル: SQLからMongoDBへのフラグメント移行

pip install Flask-PyMongo Flask

from datetime import datetime
from bson.objectid import ObjectId
class Post:
    def __init__(self, title, content, date_posted=None, 
                 _id=None):
        self.title = title
        self.content = content
        self.date_posted = date_posted if date_posted \
                           else datetime.utcnow()
        self._id = _id if _id else ObjectId()
    def to_dict(self):
        return {
            "title": self.title,
            "content": self.content,
            "date_posted": self.date_posted,
            "_id": self._id
        }
    @staticmethod
    def from_dict(data):
        return Post(
            title=data.get('title'),
            content=data.get('content'),
            date_posted=data.get('date_posted'),
            _id=data.get('_id')
        )
    def __repr__(self):
        return f"Post('{self.title}', \
                '{self.date_posted}')"

from flask import Flask
from flask_pymongo import PyMongo
from config import Config
app = Flask(__name__)
app.config.from_object(Config)
mongo = PyMongo(app)
from app import routes

import os
class Config:
    SECRET_KEY = os.urandom(24)
    MONGO_URI = 'mongodb://localhost:27017/blogdb'

from flask import render_template, url_for, flash, \
                  redirect, request
from app import app, mongo
from bson.objectid import ObjectId
from app.models import Post
from app.forms import PostForm
@app.route("/")
@app.route("/home")
def home():
    posts_data = mongo.db.posts.find()
    posts = [Post.from_dict(post) for post in posts_data]
    return render_template("index.html", posts=posts)
@app.route("/post/new", methods=["GET", "POST"])
def new_post():
    form = PostForm()
    if form.validate_on_submit():
        post = Post(title=form.title.data, 
                   content=form.content.data)
        mongo.db.posts.insert_one(post.to_dict())
        flash("Your post has been created!", "success")
        return redirect(url_for("home"))
    return render_template("post.html", title="New Post", 
                         form=form)
@app.route("/post/<post_id>")
def post(post_id):
    post_data = mongo.db.posts.find_one_or_404(
        {"_id": ObjectId(post_id)})
    post = Post.from_dict(post_data)
    return render_template("detail.html", post=post)
@app.route("/post/<post_id>/edit", methods=["GET", "POST"])
def edit_post(post_id):
    post_data = mongo.db.posts.find_one_or_404(
        {"_id": ObjectId(post_id)})
    post = Post.from_dict(post_data)
    form = PostForm()
    if form.validate_on_submit():
        updated_post = {
            "$set": {"title": form.title.data, 
                    "content": form.content.data}
        }
        mongo.db.posts.update_one(
            {"_id": ObjectId(post_id)}, updated_post)
        flash("Your post has been updated!", "success")
        return redirect(url_for("post", post_id=post_id))
    elif request.method == "GET":
        form.title.data = post.title
        form.content.data = post.content
    return render_template("edit_post.html", 
                         title="Edit Post", form=form)
@app.route("/post/<post_id>/delete", methods=["POST"])
def delete_post(post_id):
    mongo.db.posts.delete_one({"_id": ObjectId(post_id)})
    flash("Your post has been deleted!", "success")
    return redirect(url_for("home"))
@app.route("/search", methods=["GET", "POST"])
def search():
    query = request.args.get("query")
    posts_data = mongo.db.posts.find(
        {
            "$or": [
                {"title": {"$regex": query, 
                          "$options": "i"}},
                {"content": {"$regex": query, 
                            "$options": "i"}},
            ]
        }
    )
    posts = [Post.from_dict(post) for post in posts_data]
    return render_template("index.html", posts=posts)

import sqlite3
# Connect to the SQLite database
db_path = 'relative/path/to/site.db'
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
# Query to get all table names
cursor.execute("SELECT name FROM sqlite_master WHERE \
               type='table';")
tables = cursor.fetchall()
# Close the connection
conn.close()
print(tables)

import sqlite3
import json
# Connect to the SQLite database
db_path = 'relative/path/to/site.db'
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
# Export data from the 'post' table
cursor.execute("SELECT * FROM post")
rows = cursor.fetchall()
# Get the column names
column_names = [description[0] for description in \
                cursor.description]
# Convert to list of dictionaries
data = [dict(zip(column_names, row)) for row in rows]
# Save to a JSON file
with open('post.json', 'w') as f:
    json.dump(data, f, indent=4)
# Close the connection
conn.close()

from pymongo import MongoClient
import json
from datetime import datetime
def import_json_to_mongo(db_name, collection_name, 
                        json_path):
    client = MongoClient("mongodb://localhost:27017/")
    db = client[db_name]
    def convert_date(data):
        # Update 'date_posted' field to datetime object
        for item in data:
            if 'date_posted' in item:
                item['date_posted'] = datetime.strptime(
                    item['date_posted'], 
                    '%Y-%m-%d %H:%M:%S.%f')
        return data
    with open(json_path, 'r') as f:
        data = json.load(f)
        data = convert_date(data)
        db[collection_name].insert_many(data)
    client.close()
if __name__ == "__main__":
    db_name = 'blogdb'
    collection_name = 'posts'
    json_path = 'flask-mongo/post.json'
    import_json_to_mongo(db_name, collection_name, 
                        json_path)

python run.py