当サイトは、アフィリエイト広告を利用しています

【Docker × Flask】FlaskAPIでMongoDBを使う

作成日：2024月07月20日

更新日：2025年04月12日

pythonのFlask製のREST APIからMongoDBを使う方法を
をまとめる。

FlaskでMongoDBにアクセスする場合は

PyMongo
Flask-PyMongo

のどちらかのライブラリを使う。

当記事では両方のパターンの使い方をまとめる。
サンプルとしてFlask製の簡易なREST APIを実装して動作を確認する

もしその前にMongoDBがそもそもわからないという場合は下記記事でまとめていますので
見て頂ければ助かります。

PyMongoとFlask-PyMongoの違い

PyMongoとFlask-PyMongoはどちらもMongoDBに接続するためのライブラリだが
使用する場面や目的が少し異なるので違いをまとめる

汎用性

汎用性としてはPyMongoの方が高い。 PyMongoはどのようなPythonのアプリケーションでも使用可能だが
Flask-PyMongoはFlaskアプリケーション専用のため他では使えない

設定方法

設定方法としては、使うのがFlaskであった場合は
Flask-PyMongoの方が簡単にできる

PyMongoは手動で設定と接続管理を行う必要があるが
Flask-PyMongo: Flask の設定ファイルを使って簡単に設定可能。

拡張性

pymongo自体はFlaskに依存していないため、カスタマイズなどを
行う場合などの拡張性はPyMongoの方が高い

違いまとめ

まとめるとFlaskを使う場合はFlask-PyMongoを使うでいいと思う。
Flask-PyMongoはPyMongoの機能をラップしているため、基本的にPyMongoと同じ機能にアクセス可能。

余程、複雑な処理をする。または別のフレームワークに切り替える可能性がある
などの場合はフレームワークに依存しないPyMongoを使ったほうがいい

環境

開発環境としては、dockerを使ったコンテナ環境をつくり
動作を確認するので環境条件は下記のようにする

Windows10
Docker version 24.0.2(Docker for Windows)
Flask 3.0.2
pymongo 4.8.0 ※pymongoを使う場合
flask_pymongo 2.3.0　※flask_pymongoを使う場合

PyMongoを使う

まずはPyMongoを使うパターンでやってみる。

構成

プロジェクト構成は下記にする

bash


.
|-- Dockerfile
|-- api
|   |-- __init__.py
|   |-- api.py
|   |-- config.py
|   |-- database.py
|   |-- endpoints.py
|   `-- wsgi.py
|-- docker-compose.yml
|-- .env
|-- mongo
|   |-- db
|   `-- init
|       `-- init-db.js
|-- requirements.txt
`-- vscode_ex_install.sh

コンテナ作成

REST APIからMongoDBを使うので

flask REST APIコンテナ
MongoDBコンテナ

は必須で作成する

今回はこれに加えてMongoDBのデータをブラウザで
見れるように

mongo_expressコンテナ

も作る

mongo_expressの使い方については下記記事で紹介しています

MongoDBコンテナを直接みる場合は「mongosh」を使うこともできる

コンテナ作成関係のファイルを解説する

Dockerfile


FROM python:3.12

# workspaceディレクトリ作成、移動
WORKDIR /workspace

# プロジェクトディレクトリにコピー
COPY requirements.txt /workspace

# 必要モジュールのインストール
RUN pip install --upgrade pip
RUN pip install -r requirements.txt

python:3.12から必要なライブラリをrequirements.txtを
読み込んでインストールしたimageを作る

requirements.txt


Flask==3.0.2
pymongo==4.8.0

必要なライブラリを記載しているファイル。

docker-compose.yml


version: "3"
services:
  flask-api:
    container_name: "flask-api"
    build:
      context: .
      dockerfile: Dockerfile
    ports:
      - "5000:5000"
    volumes:
      # バインドマウント
      - .:/workspace
    # 環境変数読み込み
    env_file:
      - .env
    tty: true
    networks:
      flaskmongo_network:
        ipv4_address: 172.25.2.2

  # MongoDBコンテナ
  mongodb:
    container_name: "mongodb"
    restart: always
    image: mongo:6.0.13
    ports:
      - "27017:27017"
    volumes:
      - ./mongo/init:/docker-entrypoint-initdb.d # 初期化スクリプト
      - ./mongo/db:/data/db
    env_file:
      - .env
    networks:
      flaskmongo_network:
        ipv4_address: 172.25.2.3

  # mongo_expressコンテナ
  mongo_express:
    container_name: "mongo_express"
    image: mongo-express:1.0.2
    restart: always
    ports:
      - "8081:8081"
    env_file:
      - .env
    depends_on:
      mongodb:
        condition: service_started # mongoコンテナが起動してから起動させる
    networks:
      flaskmongo_network:
        ipv4_address: 172.25.2.4

# ネットワーク設定
networks:
  # ネットワーク名（ファイル内での使う名称）
  flaskmongo_network:
    # ネットワークドライバーを指定
    driver: bridge

    #ネットワークの名前を指定（dockerネットワークとしての名称）
    name: flaskmongo_network

    # IPアドレス管理（IPAM）の設定
    ipam:
      # IPAMドライバーを指定
      driver: default

      # ネットワークのサブネット設定
      config:
        - subnet: 172.25.2.0/24

flask REST APIコンテナ
MongoDBコンテナ
mongo_expressコンテナ

の3つのコンテナを作成する。

.env

コンテナに設定する環境変数を指定するファイル

.env


# flask-REST APIコンテナ用
# Flaskがどのアプリケーションを実行するか指定(pyファイル:インスタンス)
FLASK_APP=api.wsgi:api

# Flaskがどの環境で動作するか指定
FLASK_ENV=development

#デバッグモードの指定
FLASK_DEBUG=true

# MongoDBコンテナ用
# MongoDBの管理者のユーザー名
MONGO_INITDB_ROOT_USERNAME=admin

# MongoDBの管理者のパスワード
MONGO_INITDB_ROOT_PASSWORD=admin123

# 初期DBの作成
MONGO_INITDB_DATABASE=mongodb 

# MongoDB接続用URL
MONGO_URI=mongodb://admin:admin123@172.25.2.3:27017/

# Mongo Expressコンテナ用
# Mongo Expressが接続するMongoDBのサーバーを指定(MongoDBのサービス名)
ME_CONFIG_MONGODB_SERVER=mongodb

# MongoDBの管理者（rootユーザー）のユーザー名(Mongo ExpressがMongoDBに接続する際に使用)
ME_CONFIG_MONGODB_ADMINUSERNAME=admin

# MongoDBの管理者（rootユーザー）のパスワード(Mongo ExpressがMongoDBに接続する際に使用)
ME_CONFIG_MONGODB_ADMINPASSWORD=admin123

# Mongo Expressへの基本認証のユーザー名
ME_CONFIG_BASICAUTH_USERNAME=admin

# Mongo Expressへの基本認証のパスワード
ME_CONFIG_BASICAUTH_PASSWORD=password

それぞれの意味はコメントの通り。

docker-compose.ymlのenv_fileセクションで読み込ませれば
各コンテナの環境変数として設定できる

詳細は下記記事で紹介しています

mongo/db

MongoDBのデータの保存場所。
バインドマウントさせる

バインドマウント等のdockerのvolume関係については下記記事で詳細を書いてます

mongo/init/init-db.js

DBに初期データを入れるためのスクリプト※javascript コンテナ起動時に実行される

init-db.js


// init-db.js

// 管理者ユーザーでログイン
db = db.getSiblingDB("admin");
db.auth("admin", "admin123");

// mongotable データベースに接続する
db = db.getSiblingDB("SAMPLE_DB");
// コレクションを作成する
db.createCollection("USERS_COLLECTION");
// サンプルドキュメントを挿入する
db.USERS_COLLECTION.insertOne({
  user_id: "id",
  name: "name",
  age: 0,
});

print("Initialization completed.");

MongoDB内に

SAMPLE_DBデータベース

を作り、その中に

USERS_COLLECTIONコレクション

を作り、ドキュメントを1件挿入している
※MongoDBはデータベースに少なくも1つのコレクションをがないとデータベースとして認識しないので注意

コンテナ作成実行

プロジェクト直下でdocker-compose upを実行すると
3つのコンテナが作成される

コンテナ作成


$ docker-compose up -d

$ docker ps
ONTAINER ID   IMAGE                                   COMMAND                   CREATED       STATUS             PORTS                      NAMES
5c98965f5ddc   mongo-express:1.0.2                     "/sbin/tini -- /dock…"   2 hours ago   Up About an hour   0.0.0.0:8081->8081/tcp     mongo_express
38e8a9b37878   mongo:6.0.13                            "docker-entrypoint.s…"   2 hours ago   Up 2 hours         0.0.0.0:27017->27017/tcp   mongodb
ac65b03a5476   flask-rest-api-microservice-flask-api   "python3"                 2 hours ago   Up 2 hours         0.0.0.0:5000->5000/tcp     flask-api

簡易REST API作成

MongoDBにアクセスするFlaskの簡易REST APIを実装していく。

MongoDBに接続してデータのやり取るする手順については
下記のようになる※実行しているモジュールも記載

MongoDBに接続し、クライアントインスタンスを取得する
- api.py → database.pyで実行
クライアントインスタンスから使用するデータベースを選択する
- endpoints.pyで実行
データベースから使用するコレクションを選択する
- endpoints.pyで実行
コレクションに対してクエリを発行する
- endpoints.pyで実行

2,3のデータベースの指定とコレクションの指定については
使用するデータベースやコレクションが決まっている場合は1のところでもできる。

MongoDBのデータ取得だけ確認する

次からFlaskのREST APIでMongoDBにアクセスする方法をまとめるが
REST APIを作らずMongoDBからデータを取得するだけなら、コンテナを起動後に
ホストから下記スクリプトを実行すればデータは取得できる

データ取得確認pythonスクリプト

testScript


from pymongo import MongoClient

def main():
    # MongoDB接続先
    mongo_uri = 'mongodb://admin:admin123@localhost:27017/'

    # クライアント取得
    client = MongoClient(mongo_uri)

    # データベース指定
    db = client.SAMPLE_DB

    # コレクション指定
    collection = db.USERS_COLLECTION

    # クエリ発行
    users = list(collection.find({}, {'_id': 0}))
    print(users)

if __name__ == "__main__":
    main()

# 実行結果
# [{'user_id': 'id', 'name': 'name', 'age': 0}]

ホスト側にpymongoがない場合はpipでインストールする
MongoDB接続先はホストからアクセスするので「localhost」を設定する

pymongo自体はFlaskに依存していないのでPythonスクリプトでも使うことができる

スクリプトでホストからの接続が失敗するの場合は、ちゃんとmongodbコンテナが起動していることを
確認してから実行する。

それでも失敗する場合は、docker-compose.ymlのmongodbに

yml


command: ["mongod", "--bind_ip", "0.0.0.0"]

を追加して明示的にbindIp を 0.0.0.0 に設定する。
※すべての外部インターフェースからの接続を許可するようになる

api.py


from flask import Flask

from api.database import Database
from api.endpoints import configure_endpoints

def create_app():
    # アプリケーションインスタンスの作成
    app = Flask(__name__)

    # 日本語文字化け対応
    app.json.ensure_ascii = False

    # データベース接続
    Database.init_db()

    # ルーティング設定
    configure_endpoints(app)

    return app

Flaskアプリケーションの初期設定を行う。
初期設定時にMongoDBへの接続関数を呼び、クライアントインスタンスを初期化する

config.py


import os

class Config:
    MONGO_URI = os.environ.get('MONGO_URI')

コンテナ作成時に環境変数に設定した

MONGO_URI=mongodb://admin:admin123@172.25.2.3:27017/

の値を読み込み、保持させておく。

database.py


from flask import Flask
from pymongo import MongoClient
from api.config import Config

class Database:

    client = None

    @classmethod
    def init_db(cls):
        # mogogoDBサーバーに接続
        cls.client = MongoClient(Config.MONGO_URI)

MongoDBサーバーに接続し、クライアントインスタンスを初期化する。
クラス変数としてクライアントインスタンスを保持させることで
この「Database.client」を使ってどこからでもMongoDBに接続できるようにする

endpoints.py

MongoDBに対してのCRUD処理を行うモジュール

endpoints.py


from flask import Flask, jsonify, make_response, request
from pymongo import MongoClient
from api.config import Config
from api.database import Database

class Endpoints:
    def __init__(self, app: Flask):
        self.app = app
        self.configure_routes()

    def configure_routes(self):
        # 参照
        @self.app.route('/', methods=['GET'])
        def get_users():
            db = Database.client.SAMPLE_DB
            collection = db.USERS_COLLECTION
            users = list(collection.find({}, {'_id': 0}))
            return make_response(jsonify(users), 200)

        # 指定参照
        @self.app.route('/<user_id>', methods=['GET'])
        def get_user_by_id(user_id):
            db = Database.client.SAMPLE_DB
            collection = db.USERS_COLLECTION
            user = collection.find_one({"user_id": user_id}, {'_id': 0})

            if not user:
                return make_response(jsonify({"error": "ユーザーが見つかりません"}), 404)

            return make_response(jsonify(user), 200)

        # 登録
        @self.app.route('/register', methods=['POST'])
        def register_users():
            data = request.get_json()

            if not data or any(key not in data for key in ("user_id", "name", "age")):
                return make_response(jsonify({"error": "無効な入力データです"}), 400)

            db = Database.client.SAMPLE_DB
            collection = db.USERS_COLLECTION
            collection.insert_one({
                "user_id": data["user_id"],
                "name": data["name"],
                "age": data["age"]
            })

            return make_response(jsonify({"message": f"ユーザー登録が成功しました。user_id: {data['user_id']}"}), 201)

        # 削除
        @self.app.route('/<user_id>', methods=['DELETE'])
        def delete_user(user_id):
            db = Database.client.SAMPLE_DB
            collection = db.USERS_COLLECTION
            result = collection.delete_one({"user_id": user_id})

            if result.deleted_count == 0:
                return make_response(jsonify({"error": "ユーザーが見つかりません"}), 404)

            return make_response(jsonify({"message": f"ユーザーが削除されました。user_id: {user_id}"}), 200)

        # 更新
        @self.app.route('/<user_id>', methods=['PUT'])
        def update_user(user_id):
            data = request.get_json()

            if not data or any(key not in data for key in ("name", "age")):
                return make_response(jsonify({"error": "無効な入力データです"}), 400)

            db = Database.client.SAMPLE_DB
            collection = db.USERS_COLLECTION
            result = collection.update_one(
                {"user_id": user_id},
                {"$set": {"name": data["name"], "age": data["age"]}}
            )

            if result.matched_count == 0:
                return make_response(jsonify({"error": "ユーザーが見つかりません"}), 404)

            return make_response(jsonify({"message": f"ユーザー情報が更新されました。user_id: {user_id}"}), 200)

# エンドポイントの初期化処理
def configure_endpoints(app: Flask):
    Endpoints(app)

MongoDBに対してクエリを発行しているのは下記の部分になる

MongoDBアクセス部分


def configure_routes(self):
        # 参照
        @self.app.route('/', methods=['GET'])
        def get_users():
            # クライアントインスタンスから使用するデータベースを指定
            db = Database.client.SAMPLE_DB

            # データベースから使用するコレクションを選択する
            collection = db.USERS_COLLECTION

            # コレクションに対してクエリを発行する
            users = list(collection.find({}, {'_id': 0}))
            return make_response(jsonify(users), 200)

他の更新、削除等の処理はクエリーの部分が変わるだけ。

wsgi.py

WSGIエントリーポイント

wsgi.py


from api.api import create_app

api = create_app()

動作確認

念のため正常に動くかサーバー起動して確認する
起動に必要な設定は.envファイルで指定し、コンテナに環境変数に設定してあるので

起動コマンド


flask run --host=0.0.0.0

で起動できる。

REST APIの基本について詳しく知りたい方は下記記事参照

サーバー起動後にホストからHTTPリクエストを実行して確認する

参照


$ curl -X GET http://localhost:5000/
[
  {
    "age": 0,
    "name": "name",
    "user_id": "id"
  }
]

指定参照


$ curl -X GET http://localhost:5000/id
{
  "age": 0,
  "name": "name",
  "user_id": "id"
}

登録


$ curl -X POST http://localhost:5000/register -H "Content-Type: application/json" -d '{  "user_id": "1",  "name": "tujimura",  "age": 11}'
{
  "message": "ユーザー登録が成功しました。user_id: 1"
}

更新


$ curl -X PUT http://localhost:5000/1 -H "Content-Type: application/json" -d '{  "user_id": "11",  "name": "tujimura",  "age": 11111}'
{
  "message": "ユーザー情報が更新されました。user_id: 1"
}

削除


$ curl -X DELETE http://localhost:5000/id
{
  "message": "ユーザーが削除されました。user_id: id"
}

Flask-PyMongoを使う

Flask-PyMongoでMongoDBに接続してデータのやり取るする手順については
下記のようになる※実行しているモジュールも記載

FlaskアプリケーションインスタンスにMongoDBのURIを設定する
- api.pyで実行
Flask-PyMongoにFlaskアプリケーションインスタンスを渡してクライアントインスタンスを取得する
- api.py → database.pyで実行
クライアントインスタンスから使用するデータベースを選択する
- endpoints.pyで実行
データベースから使用するコレクションを選択する
- endpoints.pyで実行
コレクションに対してクエリを発行する
- endpoints.pyで実行

Flask-PyMongoを使う場合も基本的には同じコンテナ群と構成で作れる。

変更としては

requirements.txt
api.py
database.py
endpoints.py

の4つのファイルを変更するだけでできる。
requirements.txtが変わるためコンテナ群自体は作りなおす必要がある。

requirements.txt


Flask==3.0.2
Flask-PyMongo==2.3.0

Flask-PyMongoに変更する

api.py


from flask import Flask

from api.database import Database
from api.endpoints import configure_endpoints
from api.config import Config

def create_app():
    # アプリケーションインスタンスの作成
    app = Flask(__name__)

    # アプリケーションの設定
    app.config.from_object(Config)

    # 日本語文字化け対応
    app.json.ensure_ascii = False

    # データベース接続
    Database.init_db(app)

    # ルーティング設定
    configure_endpoints(app)

    return app

アプリケーション設定をconfigから読み込みFlaskアプリケーションインスタンスに設定する。
※configにはMongoDBのURIを設定しておく

Database.init_dbにはFlaskアプリケーションインスタンスを渡すようにする

database.py


from flask import Flask
from flask_pymongo import PyMongo

class Database:

    client = None

    @classmethod
    def init_db(cls, app):
        # mogogoDBサーバーに接続
        cls.client = PyMongo(app)

Flask-PyMongoにFlaskアプリケーションインスタンスを渡してクライアントインスタンスを取得する。
クライアントインスタンスはクラス変数として保持させ、どこからでもアクセス可能にする

endpoints.py


from flask import Flask, jsonify, make_response, request
from pymongo import MongoClient
from api.config import Config
from api.database import Database

class Endpoints:
    def __init__(self, app: Flask):
        self.app = app
        self.configure_routes()

    def configure_routes(self):
        @self.app.route('/', methods=['GET'])
        def get_users():
            db = Database.client.cx.SAMPLE_DB
            collection = db.USERS_COLLECTION
            users = list(collection.find({}, {'_id': 0}))
            return make_response(jsonify(users), 200)

        @self.app.route('/<user_id>', methods=['GET'])
        def get_user_by_id(user_id):
            db = Database.client.cx.SAMPLE_DB
            collection = db.USERS_COLLECTION
            user = collection.find_one({"user_id": user_id}, {'_id': 0})

            if not user:
                return make_response(jsonify({"error": "ユーザーが見つかりません"}), 404)

            return make_response(jsonify(user), 200)

        @self.app.route('/register', methods=['POST'])
        def register_users():
            data = request.get_json()

            if not data or any(key not in data for key in ("user_id", "name", "age")):
                return make_response(jsonify({"error": "無効な入力データです"}), 400)

            db = Database.client.cx.SAMPLE_DB
            collection = db.USERS_COLLECTION
            collection.insert_one({
                "user_id": data["user_id"],
                "name": data["name"],
                "age": data["age"]
            })

            return make_response(jsonify({"message": f"ユーザー登録が成功しました。user_id: {data['user_id']}"}), 201)

        @self.app.route('/<user_id>', methods=['DELETE'])
        def delete_user(user_id):
            db = Database.client.cx.SAMPLE_DB
            collection = db.USERS_COLLECTION
            result = collection.delete_one({"user_id": user_id})

            if result.deleted_count == 0:
                return make_response(jsonify({"error": "ユーザーが見つかりません"}), 404)

            return make_response(jsonify({"message": f"ユーザーが削除されました。user_id: {user_id}"}), 200)

        @self.app.route('/<user_id>', methods=['PUT'])
        def update_user(user_id):
            data = request.get_json()

            if not data or any(key not in data for key in ("name", "age")):
                return make_response(jsonify({"error": "無効な入力データです"}), 400)

            db = Database.client.cx.SAMPLE_DB
            collection = db.USERS_COLLECTION
            result = collection.update_one(
                {"user_id": user_id},
                {"$set": {"name": data["name"], "age": data["age"]}}
            )

            if result.matched_count == 0:
                return make_response(jsonify({"error": "ユーザーが見つかりません"}), 404)

            return make_response(jsonify({"message": f"ユーザー情報が更新されました。user_id: {user_id}"}), 200)

def configure_endpoints(app: Flask):
    """
    ユーザー関連のルートを設定します。

    Args:
        app (Flask): Flaskアプリケーションインスタンス
    """
    Endpoints(app)

基本的にはpymongoと同じだが、下記の部分が異なる

MongoDBアクセス部分


def configure_routes(self):
        @self.app.route('/', methods=['GET'])
        def get_users():
            db = Database.client.cx.SAMPLE_DB
            collection = db.USERS_COLLECTION
            users = list(collection.find({}, {'_id': 0}))
            return make_response(jsonify(users), 200)

データベースを指定する部分が「cx」が必要になる。

他の更新、削除等の処理はクエリーの部分が変わるだけ。

まとめ

FlaskでMongoDBを扱う場合の方法として

PyMongo
Flask-PyMongo

の2つの方法をまとめた。

個人的な感想としては、正直どっちを使ってもあまり変わらないが
普通にFlaskで作るなら「Flask-PyMongo」を使うかな。

ただ「Flask-PyMongo」はFlaskに依存しているため
他のフレームワークに変える可能性がある場合は、「PyMongo」で
つくっておいたほうがいいとおもう。

当記事ではMongDBの使用部分のみに焦点をあてて実装したが
全体的なプロジェクト構成構成を考えた上でMongoDBを使ったREST APIを実装した記事を
下記で書いているのでよければ参考にしてください

余談

当記事ではPyMongoでMongoDBにアクセスするためのクライアントインスタンス取得時に

URIで設定


# MongoDB接続先
    mongo_uri = 'mongodb://admin:admin123@localhost:27017/'

    # クライアント取得
    client = MongoClient(mongo_uri)

のようにURIを設定していたが

それぞれを設定


host = 'localhost'
    port = 27017
    username = 'admin'
    password = 'admin123'
    auth_source = 'admin'  # 認証データベース名

    # MongoClientを作成
    client = MongoClient(
        host=host,
        port=port,
        username=username,
        password=password,
        authSource=auth_source
    )

のように各々で設定することもできる。

Flask-PyMongoの場合はMongoDBの設定をURI形式で一括指定する必要があるので
MONGO_URIで設定しないとエラーになる

【Docker × Flask】FlaskAPIでMongoDBを使う

PyMongoとFlask-PyMongoの違い

汎用性

設定方法

拡張性

違いまとめ

環境

PyMongoを使う

構成

コンテナ作成

Dockerfile

requirements.txt

docker-compose.yml

.env

mongo/db

mongo/init/init-db.js

コンテナ作成実行

簡易REST API作成

MongoDBのデータ取得だけ確認する

データ取得確認pythonスクリプト

api.py

config.py

database.py

endpoints.py

wsgi.py

動作確認

参照

指定参照

登録

更新

削除

Flask-PyMongoを使う

requirements.txt

api.py

database.py

endpoints.py

まとめ

余談

新着記事