Operasi CRUD dengan Python dan MongoDB

Subscribe Dengan Account Google Untuk Membaca Artikel Tanpa Iklan
Operasi CRUD dengan Python dan MongoDB

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:


  • Ganti string koneksi, nama database, dan nama koleksi dengan milik Anda.

  • Ini adalah contoh sederhana, tentunya Anda perlu menyesuaikan dengan case yang ingin Anda selesaikan.

  • Lihat dokumentasi pymongo untuk fungsi lainnya


Baca artikel lainya :