Penggunaan try-except di dalam blok transaction.atomic() pada Django sering kali memicu TransactionManagementError ketika berhadapan dengan basis data PostgreSQL. Gejala ini ditandai dengan kegagalan beruntun pada query berikutnya, disertai pesan error current transaction is aborted, commands ignored until end of transaction block.

Gejala: Transaksi Gagal Beruntun pada PostgreSQL

Saat menjalankan operasi penulisan atau pembaruan database di dalam blok transaksi Django, developer sering menangkap IntegrityError atau error basis data lainnya menggunakan try-except Python biasa dengan asumsi eksekusi dapat dilanjutkan. Namun, query berikutnya di dalam blok transaksi yang sama justru melempar exception:

django.db.transaction.TransactionManagementError: An error occurred in the current transaction. You can't execute queries until the end of the 'atomic' block.

Pada level PostgreSQL driver (psycopg2 atau psycopg3), error aslinya berbunyi:

psycopg2.errors.InFailedSqlTransaction: current transaction is aborted, commands ignored until end of transaction block

Anti-Pattern: Menangkap Exception di Dalam Atomic Tunggal

Pola di bawah ini memperlihatkan penanganan error yang salah. Operasi database pertama memicu pelanggaran constraint, ditangkap oleh Python, lalu query kedua dieksekusi dalam scope transaksi yang sudah rusak.

from django.db import transaction, IntegrityError
from myapp.models import Customer, AuditLog

def register_customer(email, name):
    with transaction.atomic():
        try:
            # Gagal jika email sudah ada (Unique Constraint Violation)
            customer = Customer.objects.create(email=email, name=name)
        except IntegrityError:
            # Python menangkap exception, tetapi koneksi PostgreSQL berstatus ABORTED
            customer = None

        # Query berikut DIPASTIKAN melempar TransactionManagementError
        AuditLog.objects.create(
            event='CUSTOMER_REGISTRATION_FAILED' if customer is None else 'CUSTOMER_CREATED',
            detail=email
        )
        return customer

Root Cause: Desinkronisasi State PostgreSQL dan Python Runtime

Akar masalah terjadi karena perbedaan state antara level database server dan runtime Python:

  • PostgreSQL Transaction State: Ketika query menghasilkan error level SQL (seperti unique_violation atau foreign_key_violation), PostgreSQL langsung mengubah status transaksi koneksi menjadi ERROR / ABORTED. Pada status ini, PostgreSQL menolak semua perintah SQL berikutnya selain ROLLBACK.
  • Python Exception Handling: Blok try-except hanya menangkap objek exception pada memori runtime Python. Blok tersebut tidak mengirimkan instruksi pemulihan ke server PostgreSQL.
  • Django Safety Mechanism: Django menandai koneksi database dengan flag internal needs_rollback = True begitu database driver melaporkan error. Jika kode aplikasi mencoba menjalankan query lain sebelum transaksi di-rollback, Django sengaja melempar TransactionManagementError guna mencegah inkonsistensi data.

Solusi Refactoring

Terdapat dua pola baku untuk menyelesaikan masalah ini tergantung kebutuhan atomisitas proses bisnis.

Solusi 1: Isolasi Operasi Rentan Menggunakan Nested Atomic (Savepoint)

Django mendukung sub-transaksi menggunakan savepoint. Ketika transaction.atomic() dipanggil di dalam blok atomic lain, Django membuat savepoint di PostgreSQL (SAVEPOINT savepoint_name). Jika blok bersarang mengalami exception dan ditangkap di luar blok bersarang tersebut, Django melakukan ROLLBACK TO SAVEPOINT, membersihkan status error pada koneksi, dan mengizinkan query lanjutan berjalan dalam transaksi utama.

from django.db import transaction, IntegrityError
from myapp.models import Customer, AuditLog

def register_customer_safe(email, name):
    with transaction.atomic():
        customer = None
        try:
            # Blok nested atomic membuka savepoint database
            with transaction.atomic():
                customer = Customer.objects.create(email=email, name=name)
        except IntegrityError:
            # Rollback otomatis dilakukan ke batas savepoint
            pass

        # Sukses dieksekusi karena status transaksi koneksi kembali valid
        AuditLog.objects.create(
            event='CUSTOMER_REGISTRATION_FAILED' if customer is None else 'CUSTOMER_CREATED',
            detail=email
        )
        return customer

Solusi 2: Pindahkan Blok Try-Except ke Luar Transaksi

Jika operasi yang gagal membatalkan seluruh proses utama, jangan masukkan penanganan error di dalam transaksi. Jalankan transaksi di dalam blok try, lalu tangkap exception di luar konteks atomic.

from django.db import transaction, IntegrityError
from myapp.models import Customer, AuditLog

def register_customer_outer(email, name):
    try:
        with transaction.atomic():
            customer = Customer.objects.create(email=email, name=name)
            AuditLog.objects.create(event='CUSTOMER_CREATED', detail=email)
            return customer
    except IntegrityError:
        # Transaksi sudah di-rollback penuh oleh atomic keluar
        AuditLog.objects.create(event='CUSTOMER_REGISTRATION_FAILED', detail=email)
        return None

Verifikasi Pengujian Menggunakan TransactionTestCase

Kelas TestCase standar Django membungkus setiap unit test ke dalam satu transaksi tunggal yang di-rollback pada akhir method. Untuk menguji perilaku commit, rollback nyata, dan savepoint database secara akurat, gunakan TransactionTestCase.

from django.test import TransactionTestCase
from myapp.models import Customer, AuditLog
from myapp.services import register_customer_safe

class CustomerRegistrationTransactionTests(TransactionTestCase):
    def setUp(self):
        # Buat data awal yang memicu unique constraint
        Customer.objects.create(email='[email protected]', name='User A')

    def test_savepoint_handles_duplicate_and_creates_audit_log(self):
        # Eksekusi service yang menggunakan nested atomic
        customer = register_customer_safe(email='[email protected]', name='User B')

        # Verifikasi bahwa operasi tidak melempar TransactionManagementError
        self.assertIsNone(customer)
        self.assertEqual(Customer.objects.filter(email='[email protected]').count(), 1)
        self.assertTrue(
            AuditLog.objects.filter(
                event='CUSTOMER_REGISTRATION_FAILED',
                detail='[email protected]'
            ).exists()
        )

Aturan Penerapan

  • Jangan pernah mendiamkan error database menggunakan try-except di dalam blok transaction.atomic() tunggal tanpa nested block.
  • Gunakan nested with transaction.atomic(): untuk membungkus operasi berisiko tinggi jika query sesudahnya masih perlu disimpan dalam transaksi yang sama.
  • Pertimbangkan overhead savepoint pada database engine; hindari loop bersarang ribuan iterasi yang masing-masing membuka savepoint baru.