Operasi CRUD dengan Python dan MongoDB

Publish on Mar 02, 2024 by Alfin F

Kode berikut menunjukkan operasi CRUD (Create, Read, Update, Delete) menggunakan Python dan library pymongo, dilengkapi dengan penanganan kesalahan, dan validasi.

1. Dependensi:

pip install pymongo

Selanjutnya import library pymongo

import pymongo
from bson.objectid import ObjectId  # Untuk menangani ObjectIds

2. Fungsi Penanganan Kesalahan:

def handle_error(e):
    print(f"Terjadi kesalahan: {e}")
    exit(1)

3. Koneksi dan Database:

try:
    # Ganti dengan string koneksi dan nama database Anda
    client = pymongo.MongoClient("mongodb://localhost:27017/")
    db = client["database_saya"]
except pymongo.errors.ConnectionFailure as e:
    handle_error(e)

4. Collection

koleksi = db["koleksi_saya"]

5. Create (Simpan Data):

def buat_dokumen(data):
    try:
        # Validasi data sebelum disimpan
        if not isinstance(data, dict):
            raise ValueError("Data harus berupa kamus")
        result = koleksi.insert_one(data)
        print(f"Dokumen dimasukkan dengan ID: {result.inserted_id}")
    except (pymongo.errors.PyMongoError, ValueError) as e:
        handle_error(e)

# Contoh penggunaan
dokumen_baru = {"nama": "John Doe", "umur": 30}
buat_dokumen(dokumen_baru)

6. Read (Membaca Data):

def cari_dokumen(filter=None):
    try:
        if filter:
            # Gunakan filter untuk kriteria tertentu
            kursor = koleksi.find(filter)
        else:
            # Cari semua dokumen
            kursor = koleksi.find({})
        dokumen = list(kursor)
        return dokumen
    except pymongo.errors.PyMongoError as e:
        handle_error(e)

# Contoh penggunaan
semua_dokumen = cari_dokumen()
print("Semua dokumen:")
for doc in semua_dokumen:
    print(doc)

dokumen_terfilter = cari_dokumen({"umur": 30})
print("\nDokumen dengan umur 30:")
for doc in dokumen_terfilter:
    print(doc)

7. Update (Memperbarui Data):

def perbarui_dokumen(id_dokumen, data_update):
    try:
        if not isinstance(id_dokumen, ObjectId):
            raise ValueError("id_dokumen harus berupa ObjectId yang valid")
        result = koleksi.update_one({"_id": id_dokumen}, data_update)
        if result.matched_count == 0:
            print(f"Tidak ditemukan dokumen dengan ID: {id_dokumen}")
        else:
            print(f"{result.modified_count} dokumen diperbarui")
    except (pymongo.errors.PyMongoError, ValueError) as e:
        handle_error(e)

# Contoh penggunaan
id_dokumen = ObjectId("...")  # Ganti dengan ID sebenarnya
data_update = {"$set": {"umur": 35}}  # Memperbarui umur

perbarui_dokumen(id_dokumen, data_update)

8. Delete (Menghapus Data):

def hapus_dokumen(id_dokumen):
    try:
        if not isinstance(id_dokumen, ObjectId):
            raise ValueError("id_dokumen harus berupa ObjectId yang valid")
        result = koleksi.delete_one({"_id": id_dokumen})
        if result.deleted_count == 0:
            print(f"Tidak ditemukan dokumen dengan ID: {id_dokumen}")
        else:
            print(f"{result.deleted_count} dokumen dihapus")
    except (pymongo.errors.PyMongoError, ValueError) as e:
        handle_error(e)

# Contoh penggunaan
id_dokumen = ObjectId("...")  # Ganti dengan ID sebenarnya

hapus_dokumen(id_dokumen)

Catatan: