Bridges¶

TaipanStack's bridges module provides adapters connecting TaipanStack's security, resilience, and error handling with popular Python libraries.

DB Bridge¶

DB Bridge — resilient database wrappers.

Wraps SQLAlchemy async engine and Redis async client with TaipanStack's circuit breaker and retry patterns.

ResilientDatabase ¶

ResilientDatabase(
    engine: AsyncEngine,
    *,
    circuit_breaker: CircuitBreaker | None = None,
    retry_config: RetryConfig | None = None,
)

Wraps a SQLAlchemy async engine with resilience patterns.

PARAMETER	DESCRIPTION
`engine`	SQLAlchemy `AsyncEngine` instance. TYPE: `AsyncEngine`
`circuit_breaker`	Optional circuit breaker. TYPE: `CircuitBreaker \| None` DEFAULT: `None`
`retry_config`	Optional retry configuration. TYPE: `RetryConfig \| None` DEFAULT: `None`

Example

db = ResilientDatabase(engine, circuit_breaker=breaker) result = await db.execute(text("SELECT 1"))

Initialize the resilient database wrapper.

PARAMETER	DESCRIPTION
`engine`	SQLAlchemy AsyncEngine. TYPE: `AsyncEngine`
`circuit_breaker`	Optional circuit breaker. TYPE: `CircuitBreaker \| None` DEFAULT: `None`
`retry_config`	Optional retry config. TYPE: `RetryConfig \| None` DEFAULT: `None`

execute `async` ¶

execute(
    statement: Executable, **kwargs: object
) -> Result[Result[object], Exception]

Execute a SQL statement with resilience.

PARAMETER	DESCRIPTION
`statement`	SQLAlchemy statement to execute. TYPE: `Executable`
`**kwargs`	Passed to `session.execute`. TYPE: `object` DEFAULT: `{}`

RETURNS	DESCRIPTION
`Result[Result[object], Exception]`	`Ok(result)` on success, `Err` on failure.

health_check `async` ¶

health_check() -> Result[bool, Exception]

Check database connectivity.

Executes SELECT 1 to verify the connection is alive.

RETURNS	DESCRIPTION
`Result[bool, Exception]`	`Ok(True)` if healthy, `Err` on failure.

ResilientRedis ¶

ResilientRedis(
    client: Redis[bytes | str],
    *,
    circuit_breaker: CircuitBreaker | None = None,
)

Wraps a Redis async client with resilience patterns.

PARAMETER	DESCRIPTION
`client`	`redis.asyncio.Redis` instance. TYPE: `Redis[bytes \| str]`
`circuit_breaker`	Optional circuit breaker. TYPE: `CircuitBreaker \| None` DEFAULT: `None`

Example

r = ResilientRedis(redis_client, circuit_breaker=breaker) result = await r.execute("GET", "my_key")

Initialize the resilient Redis wrapper.

PARAMETER	DESCRIPTION
`client`	Redis async client. TYPE: `Redis[bytes \| str]`
`circuit_breaker`	Optional circuit breaker. TYPE: `CircuitBreaker \| None` DEFAULT: `None`

execute `async` ¶

execute(
    command: str, *args: object
) -> Result[object, Exception]

Execute a Redis command with resilience.

PARAMETER	DESCRIPTION
`command`	Redis command name (e.g. `"GET"`, `"SET"`). TYPE: `str`
`*args`	Command arguments. TYPE: `object` DEFAULT: `()`

RETURNS	DESCRIPTION
`Result[object, Exception]`	`Ok(result)` on success, `Err` on failure.

health_check `async` ¶

health_check() -> Result[bool, Exception]

Check Redis connectivity via PING.

RETURNS	DESCRIPTION
`Result[bool, Exception]`	`Ok(True)` if healthy, `Err` on failure.

HTTP Bridge¶

HTTP Bridge — safe httpx client with SSRF protection and resilience.

Wraps httpx.AsyncClient with TaipanStack's guard_ssrf, retry, and circuit breaker integrations. All outbound URLs are validated against SSRF before the request is sent.

HttpRequestKwargs ¶

Bases: TypedDict

Type definitions for HTTP request kwargs.

HttpClientKwargs ¶

Bases: TypedDict

Type definitions for HTTP client kwargs.

SafeHttpClient ¶

SafeHttpClient(
    *,
    ssrf_protection: bool = True,
    retry_config: RetryConfig | None = None,
    circuit_breaker: CircuitBreaker | None = None,
    retryable_status_codes: frozenset[
        int
    ] = _RETRYABLE_STATUS_CODES,
    timeout: float = 10.0,
    **client_kwargs: Unpack[HttpClientKwargs],
)

Async context manager wrapping httpx with TaipanStack safety.

PARAMETER	DESCRIPTION
`ssrf_protection`	Enable SSRF validation on all requests. TYPE: `bool` DEFAULT: `True`
`retry_config`	Retry configuration for transient failures. TYPE: `RetryConfig \| None` DEFAULT: `None`
`circuit_breaker`	Optional circuit breaker for all requests. TYPE: `CircuitBreaker \| None` DEFAULT: `None`
`retryable_status_codes`	HTTP status codes to retry on. TYPE: `frozenset[int]` DEFAULT: `_RETRYABLE_STATUS_CODES`
`**client_kwargs`	Passed to `httpx.AsyncClient`. TYPE: `Unpack[HttpClientKwargs]` DEFAULT: `{}`

Example

async with SafeHttpClient() as client: ... result = await client.get("https://api.example.com/data") ... if isinstance(result, Ok): print(result.unwrap().json()) ... else: print(f"Error: {result.unwrap_err()}")

Initialize the safe HTTP client.

PARAMETER	DESCRIPTION
`ssrf_protection`	Enable SSRF validation. TYPE: `bool` DEFAULT: `True`
`retry_config`	Retry configuration. TYPE: `RetryConfig \| None` DEFAULT: `None`
`circuit_breaker`	Circuit breaker instance. TYPE: `CircuitBreaker \| None` DEFAULT: `None`
`retryable_status_codes`	Status codes to retry. TYPE: `frozenset[int]` DEFAULT: `_RETRYABLE_STATUS_CODES`
`**client_kwargs`	Keyword args for httpx.AsyncClient. Default timeout is 10.0 seconds if not provided. TYPE: `Unpack[HttpClientKwargs]` DEFAULT: `{}`

aenter `async` ¶

__aenter__() -> SafeHttpClient

Enter the async context manager.

aexit `async` ¶

__aexit__(
    _exc_type: type[BaseException] | None,
    _exc_val: BaseException | None,
    _exc_tb: object,
) -> None

Exit the async context manager.

request `async` ¶

request(
    method: str,
    url: str,
    **kwargs: Unpack[HttpRequestKwargs],
) -> Result[Response, Exception]

Send an HTTP request with safety features.

PARAMETER	DESCRIPTION
`method`	HTTP method. TYPE: `str`
`url`	Target URL. TYPE: `str`
`**kwargs`	Passed to the underlying client. TYPE: `Unpack[HttpRequestKwargs]` DEFAULT: `{}`

RETURNS	DESCRIPTION
`Result[Response, Exception]`	`Ok(Response)` on success, `Err` on failure.

get `async` ¶

get(
    url: str, **kw: Unpack[HttpRequestKwargs]
) -> Result[Response, Exception]

Send a GET request.

post `async` ¶

post(
    url: str, **kw: Unpack[HttpRequestKwargs]
) -> Result[Response, Exception]

Send a POST request.

put `async` ¶

put(
    url: str, **kw: Unpack[HttpRequestKwargs]
) -> Result[Response, Exception]

Send a PUT request.

delete `async` ¶

delete(
    url: str, **kw: Unpack[HttpRequestKwargs]
) -> Result[Response, Exception]

Send a DELETE request.

patch `async` ¶

patch(
    url: str, **kw: Unpack[HttpRequestKwargs]
) -> Result[Response, Exception]

Send a PATCH request.

safe_request `async` ¶

safe_request(
    method: str,
    url: str,
    *,
    ssrf_protection: bool = True,
    retry_config: RetryConfig | None = None,
    circuit_breaker: CircuitBreaker | None = None,
    retryable_status_codes: frozenset[
        int
    ] = _RETRYABLE_STATUS_CODES,
    timeout: float | None = 10.0,
    **kwargs: Unpack[HttpRequestKwargs],
) -> Result[Response, Exception]

Perform a one-shot HTTP request with safety features.

PARAMETER	DESCRIPTION
`method`	HTTP method (GET, POST, etc.). TYPE: `str`
`url`	Target URL. TYPE: `str`
`ssrf_protection`	Validate URL against SSRF. TYPE: `bool` DEFAULT: `True`
`retry_config`	Optional retry configuration. TYPE: `RetryConfig \| None` DEFAULT: `None`
`circuit_breaker`	Optional circuit breaker. TYPE: `CircuitBreaker \| None` DEFAULT: `None`
`retryable_status_codes`	Status codes that trigger retries. TYPE: `frozenset[int]` DEFAULT: `_RETRYABLE_STATUS_CODES`
`timeout`	Explicit timeout in seconds (default: 10.0). TYPE: `float \| None` DEFAULT: `10.0`
`**kwargs`	Passed to `httpx.AsyncClient.request`. TYPE: `Unpack[HttpRequestKwargs]` DEFAULT: `{}`

RETURNS	DESCRIPTION
`Result[Response, Exception]`	`Ok(Response)` on success, `Err` on failure.

Web Bridge¶

Web Bridge — ASGI middleware for rate limiting and security headers.

Provides a framework-agnostic ASGI middleware that integrates TaipanStack's rate limiter and security headers into any ASGI application (FastAPI, Litestar, Starlette, etc.).

SecurityHeadersConfig `dataclass` ¶

SecurityHeadersConfig(
    x_content_type_options: Literal["nosniff"] = "nosniff",
    x_frame_options: Literal["DENY", "SAMEORIGIN"] = "DENY",
    x_xss_protection: Literal[
        "1; mode=block", "0"
    ] = "1; mode=block",
    strict_transport_security: str = "max-age=31536000; includeSubDomains",
    referrer_policy: Literal[
        "no-referrer",
        "no-referrer-when-downgrade",
        "origin",
        "origin-when-cross-origin",
        "same-origin",
        "strict-origin",
        "strict-origin-when-cross-origin",
        "unsafe-url",
    ] = "strict-origin-when-cross-origin",
    content_security_policy: str = "default-src 'self'",
)

Configuration for security response headers.

ATTRIBUTE	DESCRIPTION
`x_content_type_options`	Value for X-Content-Type-Options. TYPE: `Literal['nosniff']`
`x_frame_options`	Value for X-Frame-Options. TYPE: `Literal['DENY', 'SAMEORIGIN']`
`x_xss_protection`	Value for X-XSS-Protection. TYPE: `Literal['1; mode=block', '0']`
`strict_transport_security`	Value for Strict-Transport-Security. TYPE: `str`
`referrer_policy`	Value for Referrer-Policy. TYPE: `Literal['no-referrer', 'no-referrer-when-downgrade', 'origin', 'origin-when-cross-origin', 'same-origin', 'strict-origin', 'strict-origin-when-cross-origin', 'unsafe-url']`
`content_security_policy`	Value for Content-Security-Policy. TYPE: `str`

to_headers ¶

to_headers() -> list[tuple[bytes, bytes]]

Convert config to ASGI header pairs.

RETURNS	DESCRIPTION
`list[tuple[bytes, bytes]]`	List of (name, value) byte tuples.

TaipanMiddleware ¶

TaipanMiddleware(
    app: ASGIApp,
    *,
    rate_limiter: RateLimiter | None = None,
    security_headers: bool = True,
    headers_config: SecurityHeadersConfig | None = None,
)

ASGI middleware providing rate limiting and security headers.

PARAMETER	DESCRIPTION
`app`	The wrapped ASGI application. TYPE: `ASGIApp`
`rate_limiter`	Optional rate limiter instance. TYPE: `RateLimiter \| None` DEFAULT: `None`
`security_headers`	Whether to inject security headers. TYPE: `bool` DEFAULT: `True`
`headers_config`	Custom security headers configuration. TYPE: `SecurityHeadersConfig \| None` DEFAULT: `None`

Example

from taipanstack.utils.rate_limit import RateLimiter app = TaipanMiddleware( ... my_asgi_app, ... rate_limiter=RateLimiter(max_calls=100, time_window=60), ... security_headers=True, ... )

Initialize the middleware.

PARAMETER	DESCRIPTION
`app`	ASGI application to wrap. TYPE: `ASGIApp`
`rate_limiter`	Optional rate limiter. TYPE: `RateLimiter \| None` DEFAULT: `None`
`security_headers`	Inject security headers. TYPE: `bool` DEFAULT: `True`
`headers_config`	Custom headers config. TYPE: `SecurityHeadersConfig \| None` DEFAULT: `None`

call `async` ¶

__call__(
    scope: Scope, receive: Receive, send: Send
) -> None

Process an ASGI request.

PARAMETER	DESCRIPTION
`scope`	ASGI scope dict. TYPE: `Scope`
`receive`	ASGI receive callable. TYPE: `Receive`
`send`	ASGI send callable. TYPE: `Send`

result_to_response ¶

result_to_response(
    result: Result[T, Exception],
    *,
    status_ok: int = 200,
    status_err: int = 500,
) -> dict[str, object]

Convert a Result to a JSON-friendly response dict.

PARAMETER	DESCRIPTION
`result`	The Result to convert. TYPE: `Result[T, Exception]`
`status_ok`	HTTP status for `Ok` values. TYPE: `int` DEFAULT: `200`
`status_err`	HTTP status for `Err` values. TYPE: `int` DEFAULT: `500`

RETURNS	DESCRIPTION
`dict[str, object]`	Dict with `status`, `data`/`error` keys.

Example

result_to_response(Ok({"id": 1})) {"status": 200, "data": {"id": 1}}

Bridges¶

DB Bridge¶

ResilientDatabase ¶

execute async ¶

health_check async ¶

ResilientRedis ¶

execute async ¶

health_check async ¶

HTTP Bridge¶

HttpRequestKwargs ¶

HttpClientKwargs ¶

SafeHttpClient ¶

__aenter__ async ¶

__aexit__ async ¶

request async ¶

get async ¶

post async ¶

put async ¶

delete async ¶

patch async ¶

safe_request async ¶

Web Bridge¶

SecurityHeadersConfig dataclass ¶

to_headers ¶

TaipanMiddleware ¶

__call__ async ¶

result_to_response ¶

execute `async` ¶

health_check `async` ¶

execute `async` ¶

health_check `async` ¶

aenter `async` ¶

aexit `async` ¶

request `async` ¶

get `async` ¶

post `async` ¶

put `async` ¶

delete `async` ¶

patch `async` ¶

safe_request `async` ¶

SecurityHeadersConfig `dataclass` ¶

call `async` ¶