secploy 1.0.0

Event ingestion, log capture, realtime config sync, and security gating SDK for Python applications

Secploy Python SDK

Secploy Python SDK provides event ingestion, structured log capture, runtime configuration sync, request gating, and security control actions for Python services.

Table Of Contents

• Secploy Python SDK
  • Table Of Contents
  • Highlights
  • Installation
  • Quick Start
  • Security Gate
  • Runnable Example
  • CLI
  • Framework Examples
    • FastAPI
    • Flask
    • Django
  • Configuration
    • Supported Keys
    • Environment Variables
  • Runtime Config Access
    • Realtime Config Updates
    • Manual Polling (Optional)
  • Structured Log Capture
  • Endpoint Blocking
    • Basic Usage
    • Authentication
    • Safe Defaults
  • API Surface
    • SecployClient
    • ConfigManager
  • Production Notes
  • Troubleshooting
    • Missing dependency errors
    • Config sync authentication failure (401)
    • YAML output fails in CLI sync
  • License

Highlights

• Send application events to Secploy ingest.
• Capture Python logs and uncaught exceptions with structured context.
• Pull project configs from the Secploy API.
• Receive config updates through WebSocket with automatic 15-second polling fallback.
• Bootstrap project config and local .env files from CLI.

Installation

Install from PyPI:

pip install secploy

Install with explicit realtime extra:

pip install "secploy[realtime]"

Quick Start

Create a .secploy file in your project root:

api_key: YOUR_API_KEY
environment_key: YOUR_ENVIRONMENT_KEY
organization_id: YOUR_ORGANIZATION_ID
environment: production

Use the SDK:

from secploy import SecployClient

client = SecployClient()

client.send_event(
    "user.signup",
    {
        "user_id": "u_123",
        "plan": "pro",
        "source": "landing_page",
    },
)

# Dot-access project configs
google_api_key = client.env.google_api_key
var_a = client.env.var_a

# Graceful shutdown
client.stop()

Security Gate

If you want Secploy to evaluate requests before your application or HTTP client executes them, use SecployGate.

from secploy import SecployGate, SecurityGateBlocked

secploy_gate = SecployGate()

request_data = {
    "method": "POST",
    "url": "https://api.example.com/api/billing/charge",
    "headers": {
        "Authorization": "Bearer token",
        "X-User-Id": "user_123",
        "X-Session-Id": "sess_456",
    },
}

try:
    request_data = secploy_gate(request=request_data)
    # Request is allowed.
except SecurityGateBlocked as exc:
    print(exc)
    print(exc.decision)

SecployGate accepts dictionaries plus common request objects from requests, Flask, Django, and FastAPI / Starlette.

You can also build it from an existing client:

from secploy import SecployClient

client = SecployClient()
secploy_gate = client.security_gate()

To inspect the decision without raising, call inspect(...) or initialize the gate with raise_on_block=False.

To automatically protect outbound HTTP calls:

response = secploy_gate.request(
    "POST",
    "https://api.example.com/api/billing/charge",
    headers={"Authorization": "Bearer token"},
    json={"amount": 2000},
)

If you prefer a reusable requests-style client with Secploy checks on every call:

secured_session = client.security_session(
    auth={"auth_provider": "bearer"},
    metadata={"service": "billing-api"},
)

response = secured_session.post(
    "https://api.example.com/api/billing/charge",
    secploy_auth={"identity_key": "user_123"},
    json={"amount": 2000},
)

When blocked, SecurityGateBlocked now includes richer context:

• exc.reason
• exc.rule
• exc.controls
• exc.action_type
• exc.target

And framework helpers return structured block payloads including rule/control details by default.

To install it in frameworks:

# FastAPI
app.middleware("http")(secploy_gate.fastapi_middleware())

# Flask
app.before_request(secploy_gate.flask_before_request())

# Django
class SecployGateMiddleware:
    def __init__(self, get_response):
        self._middleware = secploy_gate.django_middleware(get_response)

    def __call__(self, request):
        return self._middleware(request)

Runnable Example

A complete runnable example is available at examples/basic_usage.py.

Security gate examples are available at:

• examples/endpoint_blocking_example.py
• examples/fastapi_gate_example.py
• examples/flask_gate_example.py
• examples/django_gate_middleware.py

Run it from the repository root:

python examples/basic_usage.py

Or set explicit config file path:

python examples/basic_usage.py --config-file .secploy

CLI

The package installs a secploy command.

Initialize project config:

secploy init

Force overwrite existing file:

secploy init --force

Sync remote configs to local file:

secploy sync --configs

Default output is .env in KEY="value" format.

Common options:

secploy sync --configs --output .env.local
secploy sync --configs --format json --output configs.json
secploy sync --configs --format yaml --output configs.yaml
secploy sync --configs --config-file /path/to/.secploy

Framework Examples

FastAPI

from fastapi import FastAPI, Request
from secploy import SecployClient

app = FastAPI()
client = SecployClient()


@app.middleware("http")
async def secploy_http_events(request: Request, call_next):
    response = await call_next(request)
    client.send_event(
        "http.request",
        {
            "method": request.method,
            "path": request.url.path,
            "status_code": response.status_code,
        },
    )
    return response

Flask

from flask import Flask, request
from secploy import SecployClient

app = Flask(__name__)
client = SecployClient()


@app.after_request
def secploy_http_events(response):
    client.send_event(
        "http.request",
        {
            "method": request.method,
            "path": request.path,
            "status_code": response.status_code,
        },
    )
    return response

Django

from secploy import SecployClient

client = SecployClient()


class SecployEventMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        response = self.get_response(request)
        client.send_event(
            "http.request",
            {
                "method": request.method,
                "path": request.path,
                "status_code": response.status_code,
            },
        )
        return response

Configuration

Configuration precedence (highest first):

1. Direct constructor arguments (api_key, environment_key, organization_id)
2. Environment variables (SECPLOY_*)
3. Config file (.secploy, {project}.secploy, or *.secploy discovered upward)
4. SDK defaults

Supported Keys

Key	Type	Default	Notes
api_key	str	Required	Secploy API key
environment_key	str	Required	Environment key
organization_id	str	Required	Organization identifier
environment	str	development	Environment label
ingest_url	str	https://ingest.secploy.com	Ingest base URL
api_url	str	https://api.secploy.com	API base URL
heartbeat_interval	int	60	Heartbeat interval
max_retry	int	5	Event processor retry cap
sampling_rate	float	1.0	Event sampling
log_level	str	INFO	Logger level
batch_size	int	100	Max events per batch
max_queue_size	int	10000	Queue size limit
flush_interval	int	5	Batch flush interval (seconds)
retry_attempts	int	3	Retry attempts
ignore_errors	bool	true	Continue on non-critical issues
debug	bool	false	Enables SDK debug logging
source_root	str	None	Optional source root metadata
realtime	bool	true	Enable/disable realtime config stream

Environment Variables

Use the SECPLOY_ prefix with uppercased keys:

export SECPLOY_API_KEY=YOUR_API_KEY
export SECPLOY_ENVIRONMENT_KEY=YOUR_ENVIRONMENT_KEY
export SECPLOY_ORGANIZATION_ID=YOUR_ORGANIZATION_ID
export SECPLOY_ENVIRONMENT=production
export SECPLOY_DEBUG=false

Runtime Config Access

SecployClient exposes a config manager at client.configs.

client = SecployClient()

# Ergonomic dot-access (case-insensitive fallback)
google_api_key = client.env.google_api_key
var_a = client.env.var_a

# Lazy-fetch single value
api_token = client.configs.get("THIRD_PARTY_TOKEN")

# Get full snapshot
all_configs = client.configs.all()

For optional values, use:

optional_value = client.env.get("missing_key", default="")

Realtime Config Updates

By default, the client starts realtime config delivery from:

• wss://<api-host>/ws/sdk/configs/ (derived from api_url)

Behavior:

• On websocket config.update, SDK refreshes config cache immediately.
• If websocket disconnects, SDK falls back to polling every 15 seconds.
• When websocket reconnects, polling fallback stops automatically.

Disable realtime in config:

realtime: false

Manual Polling (Optional)

def on_change(key, old_value, new_value):
    print(f"Config changed: {key}: {old_value} -> {new_value}")


client.configs.start_refresh(interval=60, on_change=on_change)
# ...
client.configs.stop_refresh()

Structured Log Capture

Capture root logger:

client = SecployClient()
client.capture_logs()

Capture specific logger(s):

client.capture_logs("my.service")
client.capture_logs(["uvicorn", "my.service"])

Stop capture:

client.stop_capturing_logs("my.service")

# Stop all resources on shutdown
client.stop()

Endpoint Blocking

Check if an endpoint is blocked before performing an action. This is useful for preventing actions on sensitive endpoints that have been administratively blocked.

Basic Usage

from secploy import SecployClient

client = SecployClient()

# Check if an endpoint is blocked
if client.endpoint_blocked(method='DELETE', endpoint='/api/users/123'):
    print("Cannot delete user - endpoint is blocked")
else:
    print("Safe to delete user")

Authentication

No extra project identifier is required. The backend resolves the project from your SDK headers:

api_key: YOUR_API_KEY
environment_key: YOUR_ENVIRONMENT_KEY
organization_id: YOUR_ORGANIZATION_ID

Then use the method without parameters:

if client.endpoint_blocked(method='POST', endpoint='/api/billing/charge'):
    print("Billing endpoint is blocked")

Safe Defaults

• Returns False if unable to determine block status (network error, missing config, etc.)
• Backend applies the blocked-endpoint rule matching for you
• All errors are logged but don't raise exceptions
• Uses your organization ID automatically from client configuration

API Surface

SecployClient

• send_event(event_type: str, payload: dict) -> bool
• track_http_request(method: str, endpoint: str, status_code: int, message: str | None = None, context: dict | None = None) -> bool
• track_error(error: Exception, endpoint: str | None = None, method: str | None = None, status_code: int = 500, context: dict | None = None) -> bool
• track_metric(name: str, value: int | float, unit: str | None = None, tags: dict | None = None, context: dict | None = None, message: str | None = None) -> bool
• endpoint_blocked(method: str, endpoint: str) -> bool
  • Check if an endpoint is blocked before performing an action
    • Queries the Secploy API for a server-side blocked-endpoint decision
  • Returns True if blocked, False if not blocked or on error (safe default)
    • Uses the client's configured api_key, environment_key, and organization_id headers
• submit_security_control_actions(actions: list[dict], timeout: int = 5) -> dict
  • Submit one or more post-auth security control actions via API-key authenticated ingest endpoint
  • Returns created/executed action statuses from backend
• submit_security_control_action(action_type: str, target_type: str, target: str, ..., timeout: int = 5) -> dict
  • Convenience wrapper for sending a single control action
• capture_logs(loggers: str | list[str] | None = None) -> None
• stop_capturing_logs(loggers: str | list[str] | None = None) -> None
• start() -> None
• stop() -> None
• env (dot-access config proxy)
• configs (ConfigManager)

ConfigManager

• fetch() -> dict[str, str]
• get(key: str, default: str | None = None) -> str | None
• all() -> dict[str, str]
• start_refresh(interval: int = 60, on_change: callable | None = None) -> None
• stop_refresh() -> None
• start_realtime(ws_url: str, headers_callback: callable) -> None
• stop_realtime() -> None

Production Notes

• Reuse one SecployClient instance per service process.
• Ensure client.stop() is called during graceful shutdown.
• Keep api_url and ingest_url aligned with your Secploy environment.
• For containerized apps, mount .secploy via secret management or use environment variables.

Troubleshooting

Missing dependency errors

pip install -U secploy

Config sync authentication failure (401)

Verify these values in .secploy:

• api_key
• environment_key
• organization_id

YAML output fails in CLI sync

pip install pyyaml

License

MIT