tenacity
tenacity 9.0.0
Silakan merujuk ke dokumentasi keuletan untuk pengalaman yang lebih baik.
Tenacity adalah pustaka percobaan ulang tujuan umum berlisensi Apache 2.0, yang ditulis dengan Python, untuk menyederhanakan tugas menambahkan perilaku percobaan ulang ke apa saja. Ini berasal dari percobaan ulang yang sayangnya tidak lagi dipertahankan. Tenacity bukanlah api yang kompatibel dengan percobaan ulang tetapi menambahkan fungsionalitas baru yang signifikan dan memperbaiki sejumlah bug yang sudah lama ada.
Kasus penggunaan yang paling sederhana adalah mencoba kembali fungsi yang tidak stabil setiap kali terjadi Pengecualian hingga suatu nilai dikembalikan.
..kode tes::
impor acak
dari percobaan ulang impor keuletan
@mencoba kembali
def do_something_unreliable():
jika acak.randint(0, 10) > 1:
raise IOError("Saus pecah, semuanya tersiram!!!111satu")
kalau tidak:
kembali "Saus yang enak!"
mencetak(melakukan_sesuatu_tidak dapat diandalkan())
.. hasil uji:: :bersembunyi: Saus yang luar biasa!
.. toctree::
:tersembunyi:
:kedalaman maksimal: 2
log perubahan
api
Untuk menginstal tenacity , cukup:
$ pip install tenacity .. pengaturan pengujian::
impor pencatatan
#
# Perhatikan impor berikut digunakan untuk kenyamanan demonstrasi saja.
# Kode produksi harus selalu mengimpor nama yang dibutuhkan secara eksplisit.
#
dari impor keuletan *
kelas MyException(Pengecualian):
lulus
Seperti yang Anda lihat di atas, perilaku defaultnya adalah mencoba lagi selamanya tanpa menunggu saat pengecualian dimunculkan.
..kode tes::
@mencoba kembali
def never_gonna_give_you_up():
print("Coba lagi selamanya dengan mengabaikan Pengecualian, jangan menunggu di antara percobaan ulang")
menaikkan Pengecualian
Mari kita tidak terlalu gigih dan menetapkan batasan, seperti jumlah upaya sebelum menyerah.
..kode tes::
@coba lagi(stop=stop_after_attempt(7))
def stop_after_7_attempts():
print("Berhenti setelah 7 kali percobaan")
menaikkan Pengecualian
Kita tidak punya waktu seharian, jadi mari kita tetapkan batasan berapa lama kita harus mencoba kembali sesuatu.
..kode tes::
@coba lagi(stop=stop_after_delay(10))
def stop_after_10_s():
print("Berhenti setelah 10 detik")
menaikkan Pengecualian
Jika Anda berada pada tenggat waktu yang ketat, dan melebihi waktu tunda bukanlah hal yang baik, maka Anda dapat berhenti mencoba lagi satu kali sebelum Anda dapat melampaui penundaan tersebut.
..kode tes::
@coba lagi(stop=stop_before_delay(10))
def stop_before_10_s():
print("Menghentikan 1 percobaan sebelum 10 detik")
menaikkan Pengecualian
Anda dapat menggabungkan beberapa kondisi stop dengan menggunakan | operator:
..kode tes::
@retry(stop=(stop_after_delay(10) | stop_after_attempt(5)))
def stop_after_10_s_or_5_retries():
print("Berhenti setelah 10 detik atau 5 kali percobaan")
menaikkan Pengecualian
Kebanyakan hal tidak suka disurvei secepat mungkin, jadi tunggu saja 2 detik di antara percobaan ulang.
..kode tes::
@coba lagi(tunggu=tunggu_diperbaiki(2))
def tunggu_2_s():
print("Tunggu 2 detik antara percobaan ulang")
menaikkan Pengecualian
Beberapa hal memiliki kinerja terbaik dengan sedikit keacakan.
..kode tes::
@coba lagi(tunggu=tunggu_acak(min=1, maks=2))
def tunggu_acak_1_ke_2_s():
print("Tunggu secara acak 1 hingga 2 detik di antara percobaan ulang")
menaikkan Pengecualian
Selain itu, sulit untuk mengalahkan kemunduran eksponensial ketika mencoba kembali layanan terdistribusi dan titik akhir jarak jauh lainnya.
..kode tes::
@coba lagi(tunggu=tunggu_eksponensial(pengganda=1, min=4, maks=10))
def tunggu_eksponensial_1():
print("Tunggu 2^x * 1 detik antara setiap percobaan ulang dimulai dengan 4 detik, lalu hingga 10 detik, lalu 10 detik setelahnya")
menaikkan Pengecualian
Selain itu, sulit untuk mengalahkan kombinasi waktu tunggu dan jitter yang tetap (untuk membantu menghindari kawanan yang bergemuruh) ketika mencoba kembali layanan terdistribusi dan titik akhir jarak jauh lainnya.
..kode tes::
@coba lagi(tunggu=tunggu_diperbaiki(3) + tunggu_acak(0, 2))
def tunggu_fixed_jitter():
print("Tunggu minimal 3 detik, dan tambahkan penundaan acak hingga 2 detik")
menaikkan Pengecualian
Ketika beberapa proses bersaing untuk mendapatkan sumber daya bersama, peningkatan jitter secara eksponensial membantu meminimalkan tabrakan.
..kode tes::
@retry(tunggu=tunggu_random_exponential(pengganda=1, maks=60))
def tunggu_exponential_jitter():
print("Tunggu secara acak hingga 2^x * 1 detik antara setiap percobaan ulang hingga rentang mencapai 60 detik, lalu secara acak hingga 60 detik setelahnya")
menaikkan Pengecualian
Terkadang kita perlu membangun rantai kemunduran.
..kode tes::
@retry(wait=wait_chain(*[wait_fixed(3) untuk saya dalam rentang(3)] +
[tunggu_diperbaiki(7) untuk saya dalam rentang(2)] +
[tunggu_diperbaiki(9)]))
def tunggu_fixed_chained():
print("Tunggu 3 detik untuk 3 percobaan, 7 detik untuk 2 percobaan berikutnya, dan 9 detik untuk semua percobaan setelahnya")
menaikkan Pengecualian
Kami memiliki beberapa opsi untuk menangani percobaan ulang yang memunculkan pengecualian khusus atau umum, seperti dalam kasus di sini.
..kode tes::
kelas ClientError (Pengecualian):
"""Beberapa jenis kesalahan klien."""
@retry(retry=retry_if_Exception_type(IOError))
def might_io_error():
print("Coba lagi selamanya tanpa menunggu jika terjadi IOError, munculkan error lainnya")
menaikkan Pengecualian
@retry(retry=retry_if_not_Exception_type(ClientError))
def mungkin_klien_kesalahan():
print("Coba lagi selamanya tanpa menunggu jika terjadi kesalahan selain ClientError. Segera naikkan ClientError.")
menaikkan Pengecualian
Kita juga dapat menggunakan hasil dari fungsi tersebut untuk mengubah perilaku percobaan ulang.
..kode tes::
def is_none_p(nilai):
"""Kembalikan Benar jika nilainya Tidak Ada"""
nilai pengembaliannya adalah Tidak Ada
@retry(retry=retry_if_result(is_none_p))
def mungkin_return_none():
print("Coba lagi tanpa menunggu jika nilai yang dikembalikan adalah Tidak Ada")
Lihat juga metode ini:
..kode tes::
coba lagi_jika_pengecualian
coba lagi_jika_pengecualian_tipe
coba lagi_jika_bukan_pengecualian_tipe
coba lagi_kecuali_pengecualian_tipe
coba lagi_jika_hasil
coba lagi_jika_tidak_hasil
coba lagi_jika_pengecualian_pesan
coba lagi_jika_bukan_pengecualian_pesan
coba lagi_apa saja
coba lagi_semua
Kami juga dapat menggabungkan beberapa kondisi:
..kode tes::
def is_none_p(nilai):
"""Kembalikan Benar jika nilainya Tidak Ada"""
nilai pengembaliannya adalah Tidak Ada
@retry(retry=(retry_if_result(is_none_p) | retry_if_Exception_type()))
def mungkin_return_none():
print("Coba lagi selamanya mengabaikan Pengecualian tanpa menunggu jika nilai yang dikembalikan adalah Tidak Ada")
Kombinasi berhenti, menunggu, dll. juga didukung untuk memberi Anda kebebasan untuk memadupadankan.
Anda juga dapat mencoba ulang secara eksplisit kapan saja dengan memunculkan pengecualian TryAgain:
..kode tes::
@mencoba kembali
def lakukan_sesuatu():
hasil = sesuatu_else()
jika hasil == 23:
naikkan Coba Lagi
Biasanya ketika fungsi Anda gagal untuk terakhir kalinya (dan tidak akan dicoba lagi berdasarkan pengaturan Anda), RetryError akan muncul. Pengecualian yang ditemui kode Anda akan ditampilkan di suatu tempat di tengah-tengah jejak tumpukan.
Jika Anda lebih suka melihat pengecualian yang ditemukan kode Anda di akhir pelacakan tumpukan (tempat yang paling terlihat), Anda dapat mengatur reraise=True.
..kode tes::
@retry(reraise=Benar, stop=stop_after_attempt(3))
def naikkan_pengecualian_saya():
naikkan MyException("Gagal")
mencoba:
naikkan_pengecualian_saya()
kecuali Pengecualian Saya:
# waktu percobaan ulang habis
lulus
Anda dapat mengeksekusi suatu tindakan sebelum ada upaya memanggil fungsi tersebut dengan menggunakan fungsi before callback:
..kode tes::
impor pencatatan
sistem impor
logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
logger = logging.getLogger(__name__)
@retry(stop=stop_after_attempt(3), sebelum=before_log(logger, logging.DEBUG))
def naikkan_pengecualian_saya():
naikkan MyException("Gagal")
Dengan semangat yang sama, dimungkinkan untuk mengeksekusi setelah panggilan gagal:
..kode tes::
impor pencatatan
sistem impor
logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
logger = logging.getLogger(__name__)
@retry(stop=stop_after_attempt(3), after=after_log(logger, logging.DEBUG))
def naikkan_pengecualian_saya():
naikkan MyException("Gagal")
Dimungkinkan juga untuk hanya mencatat kegagalan yang akan dicoba ulang. Biasanya percobaan ulang terjadi setelah interval tunggu, sehingga argumen kata kunci disebut before_sleep :
..kode tes::
impor pencatatan
sistem impor
logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
logger = logging.getLogger(__name__)
@coba lagi(stop=stop_after_attempt(3),
before_sleep=before_sleep_log(logger, logging.DEBUG))
def naikkan_pengecualian_saya():
naikkan MyException("Gagal")
Anda dapat mengakses statistik tentang percobaan ulang yang dilakukan pada suatu fungsi dengan menggunakan atribut statistik yang dilampirkan pada fungsi tersebut:
..kode tes::
@coba lagi(stop=stop_after_attempt(3))
def naikkan_pengecualian_saya():
naikkan MyException("Gagal")
mencoba:
naikkan_pengecualian_saya()
kecuali Pengecualian:
lulus
cetak(naikkan_pengecualian_saya.statistik)
.. hasil uji:: :bersembunyi: ...
Anda juga dapat menentukan callback Anda sendiri. Callback harus menerima satu parameter bernama retry_state yang berisi semua informasi tentang pemanggilan percobaan ulang saat ini.
Misalnya, Anda dapat memanggil fungsi panggilan balik khusus setelah semua percobaan ulang gagal, tanpa memunculkan pengecualian (atau Anda dapat menaikkan kembali atau melakukan apa saja)
..kode tes::
def return_last_value(coba lagi):
"""mengembalikan hasil percobaan panggilan terakhir"""
kembalikan retry_state.outcome.result()
def is_false(nilai):
"""Kembalikan Benar jika nilainya Salah"""
nilai pengembaliannya adalah False
# akan mengembalikan False setelah mencoba 3 kali untuk mendapatkan hasil yang berbeda
@coba lagi(stop=stop_after_attempt(3),
coba lagi_error_callback=return_last_value,
coba lagi=coba lagi_jika_hasil(adalah_false))
def akhirnya_return_false():
kembali Salah
Argumen retry_state adalah objek kelas :class:`~tenacity.RetryCallState` .
Dimungkinkan juga untuk menentukan callback khusus untuk argumen kata kunci lainnya.
.. fungsi:: my_stop(retry_state) :param RetryCallState retry_state: info tentang pemanggilan percobaan ulang saat ini :return: apakah percobaan ulang harus dihentikan atau tidak :rtype: bodoh
.. fungsi:: my_wait(retry_state) :param RetryCallState retry_state: info tentang pemanggilan percobaan ulang saat ini :return: jumlah detik untuk menunggu sebelum mencoba lagi berikutnya :rtype: mengapung
.. fungsi:: my_retry(retry_state) :param RetryCallState retry_state: info tentang pemanggilan percobaan ulang saat ini :return: apakah percobaan ulang harus dilanjutkan atau tidak :rtype: bodoh
.. fungsi:: my_before(retry_state) :param RetryCallState retry_state: info tentang pemanggilan percobaan ulang saat ini
.. fungsi:: my_after(retry_state) :param RetryCallState retry_state: info tentang pemanggilan percobaan ulang saat ini
.. fungsi:: my_before_sleep(retry_state) :param RetryCallState retry_state: info tentang pemanggilan percobaan ulang saat ini
Berikut ini contoh dengan fungsi before_sleep khusus:
..kode tes::
impor pencatatan
logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
logger = logging.getLogger(__name__)
def my_before_sleep(coba lagi):
jika retry_state.attempt_number < 1:
loglevel = logging.INFO
kalau tidak:
loglevel = logging.PERINGATAN
logger.log(
loglevel, 'Mencoba ulang %s: percobaan %s diakhiri dengan: %s',
retry_state.fn, retry_state.attempt_number, retry_state.outcome)
@coba lagi(stop=stop_after_attempt(3), before_sleep=my_before_sleep)
def naikkan_pengecualian_saya():
naikkan MyException("Gagal")
mencoba:
naikkan_pengecualian_saya()
kecuali RetryError:
lulus
Anda dapat mengubah argumen dekorator percobaan ulang sesuai kebutuhan saat memanggilnya dengan menggunakan fungsi retry_with yang dilampirkan pada fungsi yang dibungkus:
..kode tes::
@coba lagi(stop=stop_after_attempt(3))
def naikkan_pengecualian_saya():
naikkan MyException("Gagal")
mencoba:
naikkan_pengecualian_saya.coba lagi_dengan(stop=stop_after_attempt(4))()
kecuali Pengecualian:
lulus
cetak(naikkan_pengecualian_saya.statistik)
.. hasil uji:: :bersembunyi: ...
Jika Anda ingin menggunakan variabel untuk menyiapkan parameter percobaan ulang, Anda tidak harus menggunakan dekorator percobaan ulang - Anda dapat menggunakan Coba Ulang secara langsung:
..kode tes::
def tidak pernah_cukup_baik(arg1):
naikkan Pengecualian('Argumen tidak valid: {}'.format(arg1))
def coba_never_good_enough(max_attempts=3):
retryer = Mencoba lagi(stop=stop_after_attempt(max_attempts), reraise=True)
retryer(never_good_enough, 'Saya benar-benar mencoba')
Anda mungkin juga ingin mengubah perilaku fungsi yang didekorasi untuk sementara, seperti dalam pengujian untuk menghindari waktu tunggu yang tidak perlu. Anda dapat memodifikasi/menambal atribut retry yang melekat pada fungsi. Ingatlah bahwa ini adalah atribut tulis saja, statistik harus dibaca dari atribut statistik fungsi.
..kode tes::
@coba lagi(stop=stop_after_attempt(3), tunggu=tunggu_diperbaiki(3))
def naikkan_pengecualian_saya():
naikkan MyException("Gagal")
dari tiruan impor yang paling unittest
dengan mock.patch.object(raise_my_Exception.retry, "wait", wait_fixed(0)):
mencoba:
naikkan_pengecualian_saya()
kecuali Pengecualian:
lulus
cetak(naikkan_pengecualian_saya.statistik)
.. hasil uji:: :bersembunyi: ...
Tenacity memungkinkan Anda mencoba kembali blok kode tanpa perlu membungkusnya dalam fungsi yang terisolasi. Hal ini memudahkan untuk mengisolasi blok yang gagal sambil berbagi konteks. Caranya adalah dengan menggabungkan perulangan for dan pengelola konteks.
..kode tes::
dari kegigihan impor Mencoba lagi, RetryError, stop_after_attempt
mencoba:
untuk mencoba Mencoba Kembali(stop=stop_after_attempt(3)):
dengan upaya:
naikkan Pengecualian('Kode saya gagal!')
kecuali RetryError:
lulus
Anda dapat mengonfigurasi setiap detail kebijakan percobaan ulang dengan mengonfigurasi objek Mencoba Kembali.
Dengan kode async Anda dapat menggunakan AsyncRetrying.
..kode tes::
dari kegigihan impor AsyncRetrying, RetryError, stop_after_attempt
fungsi def async():
mencoba:
async untuk upaya di AsyncRetrying(stop=stop_after_attempt(3)):
dengan upaya:
naikkan Pengecualian('Kode saya gagal!')
kecuali RetryError:
lulus
Dalam kedua kasus tersebut, Anda mungkin ingin mengatur hasil percobaan sehingga tersedia dalam strategi percobaan ulang seperti retry_if_result . Ini dapat dilakukan dengan mengakses properti retry_state :
..kode tes::
dari kegigihan impor AsyncRetrying, retry_if_result
fungsi def async():
async untuk upaya di AsyncRetrying(retry=retry_if_result(lambda x: x < 3)):
dengan upaya:
hasil = 1 # Beberapa perhitungan rumit, pemanggilan fungsi, dll.
jika tidak mencoba.retry_state.outcome.failed:
percobaan.retry_state.set_result(hasil)
hasil kembali
Terakhir, retry juga berfungsi pada coroutine asyncio, Trio, dan Tornado (>= 4.5). Tidur juga dilakukan secara asinkron.
@ retry
async def my_asyncio_function ( loop ):
await loop . getaddrinfo ( '8.8.8.8' , 53 ) @ retry
async def my_async_trio_function ():
await trio . socket . getaddrinfo ( '8.8.8.8' , 53 ) @ retry
@ tornado . gen . coroutine
def my_async_tornado_function ( http_client , url ):
yield http_client . fetch ( url )Anda bahkan dapat menggunakan event loop alternatif seperti curio dengan meneruskan fungsi sleep yang benar:
@ retry ( sleep = curio . sleep )
async def my_async_curio_function ():
await asks . get ( 'https://example.org' )reno digunakan untuk mengelola log perubahan. Lihatlah dokumen penggunaannya.
Pembuatan dokumen akan secara otomatis mengkompilasi log perubahan. Anda hanya perlu menambahkannya.
# Opens a template file in an editor
tox -e reno -- new some-slug-for-my-change --edit
