# tnm-circuit-breaker
A lightweight Python circuit breaker for protecting downstream systems.
> See **[USAGE](https://github.com/tnm-circuit-breaker/USAGE.md)** for full examples.
---
## Quick summary
* Purpose: protect downstream services (i.e., Elasticsearch, RabbitMQ, Kafka, Postgres, external HTTP APIs, ...) from
cascading failures.
* Backends:
* `local`: no extra dependencies needed
* `redis`: distributed backend using Redis for atomic operations.
---
## Install
```bash
# core package
pip install tnm-circuit-breaker
# redis support
pip install "tnm-circuit-breaker[redis]"
```
---
## `CircuitBreaker` Public APIs
```python
class CircuitBreaker:
def record_failure(*args) -> int: ...
def last_failure(*args): ...
def record_success(*args) -> None: ...
def protect(*args): ...
# a decorator
def execute(*args): ...
async def execute_async(*args): ...
def raise_circuit_open_error(*args): ...
```
---
## Quick Usage
### Using the `protect` decorator
```py
from tnm.circuit import get_breaker
from tnm.circuit.exceptions import CircuitOpenError
breaker = get_breaker() # defaults
@breaker.protect("service-name")
def my_function():
pass
try:
my_function()
except CircuitOpenError:
...
# handle circuit open
```
### Using the `execute` method
```py
from tnm.circuit import get_breaker
from tnm.circuit.exceptions import CircuitOpenError
breaker = get_breaker() # defaults
def my_function():
pass
try:
breaker.execute("service-name", my_function)
except CircuitOpenError:
...
# handle circuit open
```
### Using the `execute_async` method
```py
import asyncio
from tnm.circuit import get_breaker
from tnm.circuit.exceptions import CircuitOpenError
breaker = get_breaker() # defaults
async def my_async_function():
pass
async def main():
try:
await breaker.execute_async("service-name", my_async_function)
except CircuitOpenError:
...
# handle circuit open
if __name__ == "__main__":
asyncio.run(main())
```
---
## Exceptions
All library exceptions inherit from a small, focused hierarchy:
* `CircuitError`: Base exception for all circuit breaker errors.
* `CircuitBackendError`: backend operational issues.
* `CircuitOpenError`: raised when an operation is attempted while a service's circuit is open.
* `ReturnValuePolicyError`: raised when a return-value rule matched; contains `.retval` and `.retval_policy`.
Handle `CircuitOpenError` or `ReturnValuePolicyError` as an expected operational outcome.
When catching, inspect `.args` or `.__cause__` for low-level detail.
---
## Contributing
Contributions welcome.
---
## License
MIT
Raw data
{
"_id": null,
"home_page": null,
"name": "tnm-circuit-breaker",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.11",
"maintainer_email": null,
"keywords": "breaker, circuit, circuit breaker, redis",
"author": null,
"author_email": "Justice Sandram <justicesandram@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/13/ef/7ab8645ff78d320c100edc5acd782a9750c8b50dcfe1b7b2c1a41a0ee748/tnm_circuit_breaker-0.2.2.tar.gz",
"platform": null,
"description": "# tnm-circuit-breaker\r\n\r\nA lightweight Python circuit breaker for protecting downstream systems.\r\n\r\n> See **[USAGE](https://github.com/tnm-circuit-breaker/USAGE.md)** for full examples.\r\n\r\n---\r\n\r\n## Quick summary\r\n\r\n* Purpose: protect downstream services (i.e., Elasticsearch, RabbitMQ, Kafka, Postgres, external HTTP APIs, ...) from\r\n cascading failures.\r\n* Backends:\r\n\r\n * `local`: no extra dependencies needed\r\n * `redis`: distributed backend using Redis for atomic operations.\r\n\r\n---\r\n\r\n## Install\r\n\r\n```bash\r\n# core package\r\npip install tnm-circuit-breaker\r\n\r\n# redis support\r\npip install \"tnm-circuit-breaker[redis]\"\r\n\r\n```\r\n\r\n---\r\n\r\n## `CircuitBreaker` Public APIs\r\n\r\n```python\r\nclass CircuitBreaker:\r\n def record_failure(*args) -> int: ...\r\n\r\n def last_failure(*args): ...\r\n\r\n def record_success(*args) -> None: ...\r\n\r\n def protect(*args): ...\r\n\r\n # a decorator\r\n\r\n def execute(*args): ...\r\n\r\n async def execute_async(*args): ...\r\n\r\n def raise_circuit_open_error(*args): ...\r\n\r\n```\r\n\r\n---\r\n\r\n## Quick Usage\r\n\r\n### Using the `protect` decorator\r\n\r\n```py\r\nfrom tnm.circuit import get_breaker\r\nfrom tnm.circuit.exceptions import CircuitOpenError\r\n\r\nbreaker = get_breaker() # defaults\r\n\r\n\r\n@breaker.protect(\"service-name\")\r\ndef my_function():\r\n pass\r\n\r\n\r\ntry:\r\n my_function()\r\nexcept CircuitOpenError:\r\n ...\r\n # handle circuit open\r\n```\r\n\r\n### Using the `execute` method\r\n\r\n```py\r\nfrom tnm.circuit import get_breaker\r\nfrom tnm.circuit.exceptions import CircuitOpenError\r\n\r\nbreaker = get_breaker() # defaults\r\n\r\n\r\ndef my_function():\r\n pass\r\n\r\n\r\ntry:\r\n breaker.execute(\"service-name\", my_function)\r\nexcept CircuitOpenError:\r\n ...\r\n # handle circuit open\r\n```\r\n\r\n### Using the `execute_async` method\r\n\r\n```py\r\nimport asyncio\r\nfrom tnm.circuit import get_breaker\r\nfrom tnm.circuit.exceptions import CircuitOpenError\r\n\r\nbreaker = get_breaker() # defaults\r\n\r\n\r\nasync def my_async_function():\r\n pass\r\n\r\n\r\nasync def main():\r\n try:\r\n await breaker.execute_async(\"service-name\", my_async_function)\r\n except CircuitOpenError:\r\n ...\r\n # handle circuit open\r\n\r\n\r\nif __name__ == \"__main__\":\r\n asyncio.run(main())\r\n```\r\n\r\n---\r\n\r\n## Exceptions\r\n\r\nAll library exceptions inherit from a small, focused hierarchy:\r\n\r\n* `CircuitError`: Base exception for all circuit breaker errors.\r\n* `CircuitBackendError`: backend operational issues.\r\n* `CircuitOpenError`: raised when an operation is attempted while a service's circuit is open.\r\n* `ReturnValuePolicyError`: raised when a return-value rule matched; contains `.retval` and `.retval_policy`.\r\n\r\nHandle `CircuitOpenError` or `ReturnValuePolicyError` as an expected operational outcome.\r\n\r\nWhen catching, inspect `.args` or `.__cause__` for low-level detail.\r\n\r\n---\r\n\r\n## Contributing\r\n\r\nContributions welcome.\r\n\r\n---\r\n\r\n## License\r\n\r\nMIT\r\n",
"bugtrack_url": null,
"license": null,
"summary": "A ligthweight circuit breaker implementation for Python.",
"version": "0.2.2",
"project_urls": {
"Repository": "https://github.com/mpamba/tnm-circuit-breaker"
},
"split_keywords": [
"breaker",
" circuit",
" circuit breaker",
" redis"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "3120c28e8d420dd87e0510625147b68d85b6405873b7cb339a00dc410e82631b",
"md5": "53de0abf495b09836deb6bbfcc6b6767",
"sha256": "cccbc05dc62c2f663c4b51d3a60bb552b26ab9e833a9af1c741c8d09ee3e607b"
},
"downloads": -1,
"filename": "tnm_circuit_breaker-0.2.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "53de0abf495b09836deb6bbfcc6b6767",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.11",
"size": 22482,
"upload_time": "2025-10-09T07:22:09",
"upload_time_iso_8601": "2025-10-09T07:22:09.196623Z",
"url": "https://files.pythonhosted.org/packages/31/20/c28e8d420dd87e0510625147b68d85b6405873b7cb339a00dc410e82631b/tnm_circuit_breaker-0.2.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "13ef7ab8645ff78d320c100edc5acd782a9750c8b50dcfe1b7b2c1a41a0ee748",
"md5": "4d811ed02661e6da0026a165a9636495",
"sha256": "a81db1711121f5e240aed966bb2daf1b42142cd7c1c1c1a10bdb2d3960967b9a"
},
"downloads": -1,
"filename": "tnm_circuit_breaker-0.2.2.tar.gz",
"has_sig": false,
"md5_digest": "4d811ed02661e6da0026a165a9636495",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.11",
"size": 28302,
"upload_time": "2025-10-09T07:22:11",
"upload_time_iso_8601": "2025-10-09T07:22:11.285803Z",
"url": "https://files.pythonhosted.org/packages/13/ef/7ab8645ff78d320c100edc5acd782a9750c8b50dcfe1b7b2c1a41a0ee748/tnm_circuit_breaker-0.2.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-10-09 07:22:11",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "mpamba",
"github_project": "tnm-circuit-breaker",
"github_not_found": true,
"lcname": "tnm-circuit-breaker"
}
JSON
