Idempotentne narzędzia i efekty uboczne

Dlaczego idempotencja ma znaczenie

Czym jest idempotencja

Klucze idempotencji

import uuid

def create_payment(amount, idempotency_key):
    # Stripe accepts this header — returns the existing payment if key was used
    return stripe.Charge.create(
        amount=amount,
        currency='usd',
        source=token,
        idempotency_key=idempotency_key
    )

Generowanie kluczy na podstawie intencji

import hashlib, json

def intent_key(tool_name, args, run_id):
    payload = json.dumps([tool_name, args, run_id], sort_keys=True)
    return hashlib.sha256(payload.encode()).hexdigest()

print(intent_key('transfer_money', {'to': 'acc2', 'amount': 100}, 'run-1'))

Inwentaryzacja skutków ubocznych

Opakowywanie narzędzi powodujących skutki uboczne

def send_email(to, subject, body, run_id):
    key = intent_key('send_email', {'to': to, 'subject': subject, 'body': body}, run_id)
    if redis.set(f'sent:{key}', 1, nx=True, ex=86400):
        smtp.send(to, subject, body)
        return {'sent': True}
    return {'sent': False, 'reason': 'already-sent'}

Potwierdzanie przed wykonaniem działania

def transfer_money(from_acc, to_acc, amount, confirmed=False):
    if not confirmed:
        return {'pending': True, 'msg': 'Call again with confirmed=True to proceed.'}
    return {'pending': False, 'msg': f'Transferred {amount} from {from_acc} to {to_acc}.'}

print(transfer_money('acc1', 'acc2', 100))
print(transfer_money('acc1', 'acc2', 100, confirmed=True))

Narzędzia dwuetapowe

def draft_email(to, subject, body):
    draft_id = save_draft(to, subject, body)
    return {'draft_id': draft_id, 'preview': body[:200]}

def send_draft(draft_id):
    return smtp.send(load_draft(draft_id))

Działania kompensacyjne

def refund_payment(payment_id):
    return stripe.Refund.create(payment=payment_id)
# Now an erroneous create_payment can be undone.

Dzienniki audytowe

Budżet skutków ubocznych na uruchomienie

if state.side_effect_count >= 5:
    return {'error': 'Side-effect budget exhausted for this run.'}

Tryb dry-run

import os
os.environ['DRY_RUN'] = '1'
DRY_RUN = os.getenv('DRY_RUN') == '1'

def send_email(to, subject, body):
    if DRY_RUN:
        print(f'would send email to {to}')
        return {'sent': False, 'dry_run': True}
    print(f'sending email to {to}')
    return {'sent': True, 'dry_run': False}

result = send_email('user@example.com', 'Hi', 'Hello there')
print(result)

Definicja

Podsumowanie