circuits quick start guidepip install suitkaise
Circuit )Sleeps after N failures, then resets and continues.
from suitkaise import Circuit
circuit = Circuit( num_shorts_to_trip=5, sleep_time_after_trip=1.0)
for request in incoming_requests:
try:
process(request)
except ServiceError:
circuit.short() # after 5 failures, sleeps 1s, then resets
BreakingCircuit )Stays broken after N failures until you manually reset.
from suitkaise import BreakingCircuit
breaker = BreakingCircuit( num_shorts_to_trip=3, sleep_time_after_trip=1.0)
while not breaker.broken :
try:
result = risky_operation()
break
except OperationError:
breaker.short()
if breaker.broken :
handle_failure()
breaker.reset() # manually reset when ready
circuit = Circuit(
num_shorts_to_trip=5,
sleep_time_after_trip=1.0,
backoff_factor=2.0, # double sleep time after each trip
max_sleep_time=30.0 # cap at 30 seconds
)
# Trip 1: 1s, Trip 2: 2s, Trip 3: 4s, Trip 4: 8s, ...
circuit = Circuit(
num_shorts_to_trip=5,
sleep_time_after_trip=5.0,
jitter=0.2 # ±20% randomness to prevent thundering herd
)
try:
result = call_service()
except CriticalError:
circuit.trip() # skip the counter, trip immediately
except MinorError:
circuit.short() # count normally
# sync
circuit.short()
# async — .asynced() returns the async version, second () calls it
await circuit.short.asynced()()
# equivalent to:
async_short = circuit.short.asynced()
await async_short()
circuit.times_shorted # failures since last trip
circuit.total_trips # lifetime trip count
circuit.current_sleep_time # current backoff delay
breaker.broken # is it broken?
circuit = Circuit(
num_shorts_to_trip=3,
sleep_time_after_trip=1.0,
backoff_factor=2.0,
max_sleep_time=60.0
)
for batch in get_batches():
try:
result = process_batch(batch)
circuit.reset_backoff() # success! next failure starts at 1s, not wherever backoff left off
except BatchError:
circuit.short()
circuits exists, coordinated shutdown, and cross-process circuit breakingCircuit and BreakingCircuit Share integration) (level: beginner-intermediate)