Operasi CRUD dengan Python dan MongoDB

Publish on Mar 02, 2024 by Alfin F

SSD ADVAN X HIKSEMI G3000 M.2 NVMe 256GB

Solid State Drive (SSD) adalah perangkat penyimpanan data yang menggunakan teknologi IC (integrated Circuit) sehingga lebih cepat, memiliki kinerja tinggi, dan mengkonsumsi daya lebih rendah. SSD semakin populer dan banyak digunakan dalam berbagai perangkat termasuk laptop, PC, server, dan lainnya.

Free Klik Disini !

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: