Modelul Cancellation Token în Python

  • 2 minute de citit
  • 4 exemple de cod în Python

Disponibil și în English

Salut! 👋

Articolul despre modelul Cancellation Token este un model inspirat de structura CancellationToken din C# și de pachetul context din Golang.

Ideea principală a modelului este să permită utilizatorului unui API să anuleze operațiile acestuia, dar pentru ca modelul să funcționeze, API-ul trebuie scris având în vedere modelul cancellation token.

Ca să folosești modelul, ai nevoie de o clasă care reprezintă un cancellation token. Iată versiunea mea simplă:

class CancellationToken:    def __init__(self):        self._is_canceled = False    def is_canceled(self):        return self._is_canceled    def cancel(self):        self._is_canceled = True

Deoarece cancel() este o operație idempotentă, clasa CancellationToken poate fi folosită în siguranță într-o aplicație multithreaded.

Codul unui API trebuie scris având în vedere CancellationToken. De exemplu, să considerăm următoarea bucată de cod, care nu este scrisă având în vedere modelul:

while True:    time.sleep(1)    print("Hello world.")

Rulează la nesfârșit, doarme o secundă și afișează șirul „Hello World.” Îl putem rescrie folosind modelul cancellation token astfel:

ct = CancellationToken()while not ct.is_canceled():    time.sleep(1)    print('Hello World!')

Codul verifică acum dacă token-ul este anulat și, dacă este, se oprește din rulare.

Utilizatorul API-ului are mai mult control asupra momentului în care anulează execuția operației/codului.

Să luăm acest exemplu, care anulează token-ul când primește semnalul SIGINT. Semnalul SIGINT este trimis de sistemul de operare programului când apeși CTRL+C în terminal sau apeși butonul de stop din IDE-ul IntelliJ:

import functoolsimport signalimport timeclass CancellationToken:    def __init__(self):        self._is_canceled = False    def is_canceled(self):        return self._is_canceled    def cancel(self):        self._is_canceled = Truedef main():    # Configurează cancellation token    ct = CancellationToken()    # Configurează tratarea semnalului SIGINT    def signal_handler(cancellation_token: CancellationToken, sig, frame):        print("Handling sigint")        cancellation_token.cancel()    signal.signal(signal.SIGINT, functools.partial(signal_handler, ct))    # Blochează până când este primit semnalul    while not ct.is_canceled():        time.sleep(1)        print('Hello World!')    print("Done")if __name__ == '__main__':    main()

Dacă un API are un flux de control mai complex, de exemplu rulează mai multe funcții în paralel, putem trece cancellation token-ul inițial către fiecare funcție și le putem anula pe toate deodată.

Golang are WithTimeout, care creează un context ce este anulat după timpul de expirare specificat. Pentru a implementa o funcționalitate similară în CancellationToken-ul nostru, îl putem extinde printr-o subclasă în implementarea noastră Python și putem folosi un obiect Timer pentru a anula token-ul. Alternativ, putem adăuga timpul de expirare la momentul curent și să-l numim timeout_time, iar când se apelează is_canceled() putem verifica dacă momentul curent este mai mare sau egal (>=) decât timeout_time.

Sper că ți-a plăcut acest articol și mulțumesc că ai citit! 🙏

Dacă ai gânduri despre acest articol, nu ezita să lași un comentariu.