チュートリアル: SQLからMongoDBへのFlaskの移行する

pip install Flask-PyMongo Flask

from datetime import datetimefrom 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 Flaskfrom flask_pymongo import PyMongofrom 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, requestfrom app import app, mongofrom bson.objectid import ObjectIdfrom app.models import Postfrom 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 databasedb_path = 'relative/path/to/site.db'conn = sqlite3.connect(db_path)cursor = conn.cursor()
# Query to get all table namescursor.execute("SELECT name FROM sqlite_master WHERE \               type='table';")tables = cursor.fetchall()
# Close the connectionconn.close()
print(tables)

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

from pymongo import MongoClientimport jsonfrom 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