π TaipanStack¶
The Modern Python Foundation β Launch secure, high-performance Python applications in seconds.
β¨ What's New¶
- Clean Code & Refactoring: Extracted validation logic into helpers and applied guard clauses to reduce nesting in
result.pyandcircuit_breaker.py(PR #1195).
v0.6.3¶
- Security & Resilience: Hardened RateLimiter against extreme type mutations (e.g., custom mutated types like strings masquerading as numbers) and anomalous mathematical operations that could crash the service. Operations are now gracefully handled via strict Look-Before-You-Leap (LBYL) checks, falling back to safe defaults or returning appropriate
Errstates (PR #1139).
v0.6.2¶
- Clean Code & Refactoring: Proactively refactored and extracted nested structures into dedicated helpers across resilience and rate limiting components to reduce cyclomatic complexity (PR #1124).
v0.6.1¶
- Security & Hardening: Hardened RateLimiter against infinite tokens mutations, key decoding in JWT, and validate_project_name argument parsing. Fixed large password payload crashes and removed unused constants (PR #966, #967, #968, #973, #989).
- Concurrency & Resilience: Fixed a bulkhead acquisition memory leak in ResilienceOrchestrator and documented AdaptiveRetry features (PR #974, #980).
- Clean Code & Refactoring: Significantly reduced cyclomatic complexity across compatibility shims, validators, filesystems, and token buckets. Enforced strict web bridge type hints and clean keyword parameters (PR #971, #976, #979, #984, #988, #993, #997, #1002, #1003).
- QA and Standardized Testing: Fully standardized test file nomenclature, verified comprehensive test suite, and added rigorous microservice chaos engineering simulations (PR #972, #975, #981, #986, #990, #992, #994, #999, #1004).
β¨ Why TaipanStack?¶
"Write less, build better."
TaipanStack is a battle-tested foundation for production-grade Python projects that combines security, performance, and developer experience into a single, cohesive toolkit.
-
Security First
Path traversal protection, command injection guards, subprocess isolation, adaptive limiters, adaptive resilience pipeline, input sanitizers & validators, secret detection, SBOM + SLSA attestation.
-
High Performance
uvloopasync event loop,orjsonfast JSON,Pydantic v2validation. -
Rust-Style Error Handling
Ok/ErrResult types, explicit error propagation, pattern matching, no silent failures. -
Developer Experience
Pre-configured quality tools, 100% code coverage (comprehensive test suite), architecture enforcement, hardened Docker template.
π Quick Start¶
From PyPI¶
bash
pip install taipanstacktext
From Source¶
bash
git clone https://github.com/gabrielima7/TaipanStack.git
cd TaipanStack
poetry install --with devtext
Verify Installation¶
```bash
Run tests with 100% coverage¶
make test
Check architecture contracts¶
make lint-imports
Run security scans¶
make security ```text
π API Highlights¶
Result Types¶
```python from taipanstack.core.result import Result, Ok, Err, safe
@safe def divide(a: int, b: int) -> float: return a / b
match divide(10, 0): case Ok(value): print(f"Result: {value}") case Err(error): print(f"Error: {error}") ```text
Security Guards¶
```python from taipanstack.security.guards import ( guard_command_injection, guard_env_variable, guard_path_traversal, )
safe_path = guard_path_traversal(user_input, base_dir="/app/data") safe_cmd = guard_command_injection(["git", "clone", repo_url], allowed_commands=["git"]) safe_key = guard_env_variable("API_KEY", allowed_names=["API_KEY"]) ```text
Rate Limiting¶
```python from taipanstack.utils.rate_limit import rate_limit from taipanstack.core.result import Result, Ok, Err
@rate_limit(max_calls=5, time_window=10.0) def fetch_data() -> str: return "data"
Returns an Ok('data') or Err(RateLimitError('Rate limit exceeded'))¶
match fetch_data(): case Ok(value): print(f"Data: {value}") case Err(error): print(f"Rate limited: {error}") ```text
π Combining Result + Circuit Breaker¶
```python from taipanstack.core.result import safe, Ok, Err from taipanstack.resilience.circuit_breaker import CircuitBreaker
breaker = CircuitBreaker(failure_threshold=3, timeout=60, name="payments")
@breaker @safe def charge_customer(customer_id: str, amount: float) -> dict: return payment_gateway.charge(customer_id, amount)
Both circuit protection AND explicit error handling¶
result = charge_customer("cust_123", 49.99) match result: case Ok(receipt): print(f"Payment successful: {receipt}") case Err(error): print(f"Payment failed safely: {error}") ```text
π Combining Result + Retry with Monitoring¶
```python from taipanstack.core.result import safe from taipanstack.resilience.retry import retry
@retry( max_attempts=3, on=(ConnectionError, TimeoutError), on_retry=lambda attempt, max_a, exc, delay: print( f"β οΈ Attempt {attempt}/{max_a} failed, retrying in {delay:.1f}s..." ), ) @safe def fetch_user_profile(user_id: str) -> dict: return api_client.get(f"/users/{user_id}")
Retry handles transient failures, Result handles business errors¶
profile = fetch_user_profile("usr_456").unwrap_or({"name": "Unknown"}) ```text
π Adaptive Resilience Pipeline¶
```python from taipanstack.core.result import Result, Ok, Err from taipanstack.resilience.adaptive import ResilienceOrchestrator, AdaptiveCircuitBreaker from taipanstack.resilience.retry import RetryConfig
Compose an intelligent pipeline: Bulkhead -> Breaker -> Retry -> Timeout -> Fallback¶
orch = ( ResilienceOrchestrator("billing_api") .with_bulkhead(max_concurrent=10, max_queue=50) # Prevent resource exhaustion .with_circuit_breaker(AdaptiveCircuitBreaker("billing", target_error_rate=0.1)) # Auto-tunes thresholds
.with_retry(RetryConfig(max_attempts=3, initial_delay=0.1))
.with_fallback({"status": "unavailable"})
)
async def process_billing() -> Result[dict, Exception]: # The orchestrator handles all concurrency, retry, circuit breaking, and fallbacks return await orch.execute(stripe_gateway.charge) ```text
Intelligent Caching¶
```python from taipanstack.utils.cache import cached from taipanstack.core.result import Result
@cached(ttl=60) async def get_user_data(user_id: int) -> Result[dict, Exception]: return await db.fetch(user_id) # Only Ok() results are cached ```text
Fallbacks & Timeouts¶
```python from taipanstack.resilience.resilience import fallback, timeout from taipanstack.core.result import Result
@fallback({"status": "offline"}, exceptions=(TimeoutError,)) @timeout(seconds=5.0) async def fetch_remote_status() -> Result[dict, Exception]: return await api.get_status() ```text
π Architecture¶
text
ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
β Application β
β (src/app/main.py) β
βββββββββββββββββββββββββββββ¬βββββββββββββββββββββββββββββ
β
βββββββββββββββββ¬ββββββββββββββββββββββΌββββββββββββββββ¬βββββββββββββββ
βΌ βΌ βΌ βΌ βΌ
βββββββββββββββββββββββββββββββββββ ββββββββββββββββββββββββββββββββββββββββββββββββ
β Security ββ Config β β Bridges ββ Resilience ββ Utils β
β guards, saniti- ββ models, gens β β http, db ββ retry, break ββ logging, fs β
β zers, validatorsββ β β ββ watchdogs ββ context β
ββββββββββ¬βββββββββββββββββ¬ββββββββ βββββββββ¬ββββββββββββββββ¬βββββββββββββββ¬ββββββββ
β β β β β
ββββββββββββββββββ΄ββββββββββββββββββββββΌββββββββββββββββ΄βββββββββββββββ
βΌ
ββββββββββββββββββββββββββββββββββ
β Core β
β Result types, optimizations β
ββββββββββββββββββββββββββββββββββtext
Read the full architecture guide β
π DevSecOps¶
| Category | Tools | Purpose |
|---|---|---|
| SAST | Bandit, Semgrep + custom rules | Static Application Security Testing |
| SCA | pip-audit | Dependency vulnerability scanning |
| SBOM | Syft (CycloneDX) | Software Bill of Materials |
| SLSA | Cosign (Sigstore) | Artifact signing & attestation |
| Types | Mypy (strict) | Compile-time type checking |
| Lint | Ruff | Lightning-fast linting & formatting |
| Arch | Import Linter | Dependency rule enforcement |
| Test | Pytest, Hypothesis, mutmut | Property-based & mutation testing |
π Live Reports¶
| Report | Description |
|---|---|
| π§ͺ Coverage Report | Full HTML coverage report β 100% (comprehensive test suite) |
π€ Contributing¶
Contributions are welcome! See the Contributing Guide for details.
π License¶
Open-sourced under the MIT License.