feat: adding circuit breaker feature

[leandrodamascena]

Issue number: closes #8257

Summary

Changes

This PR adds a Circuit Breaker utility (under the circuit_breaker_alpha namespace) so a Lambda function can stop calling an unhealthy downstream and let it recover, instead of piling on retries.

It ships as alpha on purpose: I want about a month of feedback before we lock the public API and promote it to GA.

The failure counter lives in memory per execution environment, so a healthy circuit writes nothing; we only persist state transitions. State is shared via DynamoDB, fails open if the store is unreachable, and uses a conditional write to elect a single probe during recovery (no thundering herd). You handle rejected requests with an on_circuit_open callback and observe state changes with an on_transition hook.

User experience

Before, you had to build the state machine, shared storage, and recovery logic yourself. Now you wrap the function that makes the downstream call:

from aws_lambda_powertools.utilities.circuit_breaker_alpha import circuit_breaker
from aws_lambda_powertools.utilities.circuit_breaker_alpha.persistence import (
    CircuitBreakerDynamoDBPersistence,
)

persistence = CircuitBreakerDynamoDBPersistence(table_name="CircuitBreakerState")


@circuit_breaker(name="payment-backend", persistence_store=persistence)
def charge(order: dict) -> dict:
    return payment_api.charge(order)

With no config, sensible defaults apply (open after 5 failures, probe after 30s, close after 3 successes). When open, the call is skipped and the callback's value is returned, or a CircuitBreakerOpenError is raised.

---

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

Disclaimer: We value your time and bandwidth. As such, any pull requests created on non-triaged issues might not be successful.

[svozza]

Does _get_record really need to throw an exception here? Can we not just check if the value returned is `None?

[svozza]

What happens if this fails? Does it leave the DynamoDB row permanently stale?

[svozza]

I think this query creation logic could be a separate private method

[codecov]

Codecov Report

❌ Patch coverage is 98.18182% with 7 lines in your changes missing coverage. Please review.
✅ Project coverage is 96.62%. Comparing base (f7239b8) to head (1c5f1a1).

Files with missing lines	Patch %	Lines
...ties/circuit_breaker_alpha/persistence/dynamodb.py	93.40%	3 Missing and 3 partials ⚠️
...powertools/utilities/circuit_breaker_alpha/base.py	99.03%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff             @@
##           develop    #8266      +/-   ##
===========================================
+ Coverage    96.58%   96.62%   +0.04%     
===========================================
  Files          286      296      +10     
  Lines        14294    14679     +385     
  Branches      1192     1227      +35     
===========================================
+ Hits         13806    14184     +378     
- Misses         357      360       +3     
- Partials       131      135       +4     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[sonarqubecloud]

Quality Gate passed

Issues
0 New issues
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud

[svozza]

Nit: I would put the delete in the conditional rather than in the main path:

def _retrieve_from_cache(self, name: str) -> CircuitStateRecord | None:
      """Return a cached record if present and still within its local freshness window."""
      cached = self._cache.get(self._cache_key(name))
      if cached is None:
          return None

      local_expiry, record = cached
      if int(datetime.datetime.now().timestamp()) >= local_expiry:
          del self._cache[self._cache_key(name)]
          return None

      return record

[svozza]

Could we add a simple helper function to unpack these values rather than having to dig into the the test N field and cast to string multiple times in this module?

[svozza]

Suggested change

	**State** is the circuit's current mode: `CLOSED` (normal), `OPEN` (downstream considered unhealthy, calls skipped), or `HALF_OPEN` (testing recovery).
	**State** is the circuit's current mode: `CLOSED` (healthy), `OPEN` (downstream considered unhealthy, calls skipped), or `HALF_OPEN` (testing recovery).

[svozza]

Not sure about the tone of this message and also it's a bit verbose.