docker run -d -p 27017:27017 -e MONGO_INITDB_ROOT_USERNAME=root -e MONGO_INITDB_ROOT_PASSWORD=root -e MONGO_INITDB_DATABASE=test --name mongodb mongo:4.2-bionic

確認容器是否正常運行

docker ps -a

Docker 容器清單
容器狀態應有 UP 字串，代表容器正在運行

資料庫操作

建立連線並寫入資料

import datetime

import loguru
import pymongo

def main():
    # 建立連線用戶端
    client = pymongo.MongoClient('mongodb://%s:%s@%s:%s/' % (
        'root',         # 資料庫帳號
        'root',         # 資料庫密碼
        'localhost',    # 資料庫位址
        '27017'         # 資料庫埠號
    ))

    # 取得資料庫
    db = client.test

    # 建立第一筆資料
    db.stocks.insert_one({
        'code': '1101',
        'name': '台泥'
    })

    # 建立第二筆資料
    db.stocks.insert_one({
        'code': '1102',
        'name': '亞泥'
    })

    # 建立第三筆資料
    db.stocks.insert_one({
        'code': '1103',
        'name': '嘉泥'
    })

    # 建立第四筆資料
    db.stocks.insert_one({
        'code': '1201',
        'name': '味全'
    })

    # 關閉用戶端
    client.close()

if __name__ == '__main__':
    loguru.logger.add(f'{datetime.date.today():%Y%m%d}.log', rotation='1 day', retention='7 days', level='DEBUG')
    main()

資料表格式

讀取資料

import datetime

import loguru
import pymongo

def main():
    client = pymongo.MongoClient('mongodb://%s:%s@%s:%s/' % (
        'root',
        'root',
        'localhost',
        '27017'
    ))

    db = client.test

    try:
        db.stocks.drop()
    except Exception as e:
        loguru.logger.error('清空資料失敗')
        loguru.logger.error(e)

    loguru.logger.info('取出資料表所有資料')
    stocks = db.stocks.find()
    for stock in stocks:
        loguru.logger.info(f'{stock["code"]} {stock["name"]}')

    loguru.logger.info('使用過濾條件取出資料表資料')
    stocks = db.stocks.find({
        'code': '1102'
    })
    for stock in stocks:
        loguru.logger.info(f'{stock["code"]} {stock["name"]}')

    loguru.logger.info('使用多重過濾條件 OR 取出資料表資料')
    stocks = db.stocks.find({
        '$or': [{
            'code': {
                '$regex': '02$'
            }
        }, {
            'code': {
                '$regex': '^12'
            }
        }]
    })
    for stock in stocks:
        loguru.logger.info(f'{stock["code"]} {stock["name"]}')

    loguru.logger.info('使用多重過濾條件 AND 取出資料表資料')
    stocks = db.stocks.find({
        '$and': [{
            'code': {
                '$regex': '2$'
            }
        }, {
            'name': {
                '$regex': '泥$'
            }
        }]
    })
    for stock in stocks:
        loguru.logger.info(f'{stock["code"]} {stock["name"]}')

    loguru.logger.info('使用排序取出資料表資料')
    stocks = db.stocks.find().sort('code', pymongo.DESCENDING)
    for stock in stocks:
        loguru.logger.info(f'{stock["code"]} {stock["name"]}')

    client.close()

if __name__ == '__main__':
    loguru.logger.add(f'{datetime.date.today():%Y%m%d}.log', rotation='1 day', retention='7 days', level='DEBUG')
    main()

更新資料

import datetime

import loguru
import pymongo

def main():
    client = pymongo.MongoClient('mongodb://%s:%s@%s:%s/' % (
        'root',
        'root',
        'localhost',
        '27017'
    ))

    db = client.test

    loguru.logger.info('----- 更新單筆資料 -----')
    try:
        db.stocks.update_one({
            'code': '1102'
        }, [{
            '$set': {
                'name': {
                    '$concat': ['$name', '（水泥產業）']
                }
            }
        }])
    except Exception as e:
        loguru.logger.error('更新資料失敗')
        loguru.logger.error(e)

    loguru.logger.info('取出資料表所有資料')
    stocks = db.stocks.find()
    for stock in stocks:
        loguru.logger.info(f'{stock["code"]} {stock["name"]}')

    loguru.logger.info('----- 更新多筆資料 -----')
    try:
        session.query(Stock).filter(
            sqlalchemy.and_(
                Stock.code.like('11%'),
                Stock.code != '1102'
            )
        ).update({
            Stock.name: Stock.name + '（水泥產業）'
        }, synchronize_session=False)

        # 寫入資料庫
        session.commit()
    except Exception as e:
        # 發生例外錯誤，還原交易
        session.rollback()
        loguru.logger.error('更新資料失敗')
        loguru.logger.error(e)

    loguru.logger.info('取出資料表所有資料')
    results = session.query(Stock).all()
    for stock in results:
        loguru.logger.info(f'{stock.code} {stock.name}')

    client.close()

if __name__ == '__main__':
    loguru.logger.add(f'{datetime.date.today():%Y%m%d}.log', rotation='1 day', retention='7 days', level='DEBUG')
    main()

刪除資料

import datetime

import loguru
import pymongo

def main():
    client = pymongo.MongoClient('mongodb://%s:%s@%s:%s/' % (
        'root',
        'root',
        'localhost',
        '27017'
    ))

    db = client.test
    
    loguru.logger.info('----- 刪除單筆資料 -----')
    try:
        db.stocks.delete_one({
            'code': '1102'
        })
    except Exception as e:
        loguru.logger.error('刪除資料失敗')
        loguru.logger.error(e)

    loguru.logger.info('取出資料表所有資料')
    stocks = db.stocks.find()
    for stock in stocks:
        loguru.logger.info(f'{stock["code"]} {stock["name"]}')
    
    loguru.logger.info('----- 刪除多筆資料 -----')
    try:
        db.stocks.delete_many({
            'code': {
                '$regex': '^11'
            }
        })
    except Exception as e:
        loguru.logger.error('刪除資料失敗')
        loguru.logger.error(e)

    loguru.logger.info('取出資料表所有資料')
    stocks = db.stocks.find()
    for stock in stocks:
        loguru.logger.info(f'{stock["code"]} {stock["name"]}')

    session.close()

if __name__ == '__main__':
    loguru.logger.add(f'{datetime.date.today():%Y%m%d}.log', rotation='1 day', retention='7 days', level='DEBUG')
    main()