π TaipanStack¶
The Modern Python Foundation β Launch secure, high-performance Python applications in seconds.
β¨ 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 (1347 tests), architecture enforcement, hardened Docker template.
π Quick Start¶
From PyPI¶
pip install taipanstack
From Source¶
git clone https://github.com/gabrielima7/TaipanStack.git
cd TaipanStack
poetry install --with dev
Verify Installation¶
# Run tests with 100% coverage
make test
# Check architecture contracts
make lint-imports
# Run security scans
make security
π API Highlights¶
Result Types¶
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}")
Security Guards¶
from taipanstack.security.guards import guard_path_traversal, guard_command_injection
safe_path = guard_path_traversal(user_input, base_dir="/app/data")
safe_cmd = guard_command_injection(["git", "clone", repo_url], allowed_commands=["git"])
π Combining Result + Circuit Breaker¶
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}")
π Combining Result + Retry with Monitoring¶
from taipanstack.core.result import safe, unwrap_or
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 = unwrap_or(fetch_user_profile("usr_456"), {"name": "Unknown"})
π Adaptive Resilience Pipeline¶
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)
Intelligent Caching¶
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
Fallbacks & Timeouts¶
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()
π Architecture¶
βββββββββββββββββββββββββββββββββββββββ
β Application β
βββββββββββββββββββ¬ββββββββββββββββββββ
β
βββββββββββββββββββββββββββββΌββββββββββββββββββββββββββββ
βΌ βΌ βΌ
βββββββββββββββββββ βββββββββββββββββββ βββββββββββββββββββ
β Security β β Config β β Utils β
β guards, saniti- β β models, β β logging, retry β
β zers, validatorsβ β generators β β metrics, fs β
ββββββββββ¬βββββββββ ββββββββββ¬βββββββββ ββββββββββ¬βββββββββ
β β β
βββββββββββββββββββββββββββΌββββββββββββββββββββββββββ
βΌ
βββββββββββββββββββββββββββββββββββββββ
β Core β
β Result types, base patterns β
βββββββββββββββββββββββββββββββββββββββ
Read the full architecture guide β
π DevSecOps¶
| Category | Tools | Purpose |
|---|---|---|
| SAST | Bandit, Semgrep + custom rules | Static Application Security Testing |
| SCA | Safety, 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% (1347 tests) |
| π§ͺ Coverage Report | Full HTML coverage report β 100% (1315 tests) |
π€ Contributing¶
Contributions are welcome! See the Contributing Guide for details.
π License¶
Open-sourced under the MIT License.