Skip to content

🐍 TaipanStack

The Modern Python Foundation β€” Launch secure, high-performance Python applications in seconds.

CI Python Coverage PyPI License


✨ 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


    uvloop async event loop, orjson fast JSON, Pydantic v2 validation.

  • 🎯 Rust-Style Error Handling


    Ok/Err Result 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.