GitHub - Netrun-Systems/netrun-oss: Open source Python libraries from Netrun Systems - netrun-config, netrun-logging, netrun-auth

# Core namespace package (required for netrun.* imports)
pip install netrun-core

# Individual packages
pip install netrun-auth netrun-config netrun-logging netrun-llm

# New infrastructure packages
pip install netrun-cache netrun-resilience netrun-validation netrun-websocket

# With optional dependencies
pip install netrun-auth[all] netrun-llm[all] netrun-cache[redis] netrun-websocket[all]

# Full suite (19 packages)
pip install netrun-core netrun-auth netrun-cache netrun-config netrun-cors \
    netrun-db-pool netrun-dee netrun-dogfood netrun-env netrun-errors netrun-llm \
    netrun-logging netrun-oauth netrun-pytest-fixtures netrun-ratelimit \
    netrun-rbac netrun-resilience netrun-validation netrun-websocket

# New v2.0.0 imports (recommended)
from netrun.auth import JWTHandler, require_permission
from netrun.config import SecretCache
from netrun.logging import get_logger
from netrun.llm import LLMFallbackChain
from netrun.rbac import TenantContext

# Old imports still work (deprecated, will be removed in v3.0.0)
from netrun_auth import JWTHandler  # Still works

from netrun.llm import TenantPolicy, PolicyEnforcer, CostTier

policy = TenantPolicy(
    tenant_id="acme-corp",
    monthly_budget_usd=100.0,
    daily_budget_usd=10.0,
    provider_policies={
        "openai": ProviderPolicy(
            provider="openai",
            allowed_models=["gpt-4o-mini", "gpt-4o"],
            cost_tier_limit=CostTier.MEDIUM,
            rate_limit_rpm=60,
        ),
    },
    fallback_to_local=True,  # Fall back to Ollama if budget exceeded
)

enforcer = PolicyEnforcer(policy)
enforcer.validate_request(provider="openai", model="gpt-4o-mini", estimated_tokens=2000)

from netrun.llm import LLMTelemetry

telemetry = LLMTelemetry()
telemetry.record_request(
    provider="openai", model="gpt-4o-mini",
    tokens_input=1000, tokens_output=500,
    cost_usd=0.002, latency_ms=1200, success=True
)

report = telemetry.get_report(days=30)
print(f"Total cost: ${report['total_cost_usd']:.4f}")
print(f"P95 latency: {report['latency_p95_ms']}ms")

from netrun.rbac import TenantTestContext, TenantEscapePathScanner, ci_fail_on_findings

# Contract test: Tenant B MUST NOT see Tenant A's data
async with TenantTestContext(db_session) as ctx:
    secret = Item(name="Secret", tenant_id=ctx.tenant_a_id)
    session.add(secret)
    await session.commit()

    await ctx.switch_to_tenant_b()
    result = await session.execute(select(Item))
    assert len(result.scalars().all()) == 0  # Cross-tenant access impossible

# CI/CD escape path scanner - finds SQL without tenant filters
scanner = TenantEscapePathScanner()
findings = scanner.scan_directory("./src")
sys.exit(ci_fail_on_findings(findings))  # Fails build on critical findings

from netrun.auth import AzureADIntegration
# If azure-identity not installed:
# ImportError: AzureADIntegration requires 'azure-identity'.
# Install with: pip install netrun-auth[azure]

from netrun.llm import LLMFallbackChain

# Default chain: Claude -> OpenAI -> Ollama
chain = LLMFallbackChain()
response = chain.execute("Explain quantum computing")

print(f"Response: {response.content}")
print(f"Provider: {response.adapter_name}")
print(f"Cost: ${response.cost_usd:.6f}")

from netrun.config import SecretCache

cache = SecretCache(vault_url="https://your-vault.vault.azure.net/")
secret = cache.get_secret("database-password")

from netrun.logging import get_logger

logger = get_logger(__name__)
logger.info("Application started", version="2.0.0")

from netrun.auth import CasbinRBACManager

rbac = CasbinRBACManager(model_path="rbac_model.conf")
if rbac.check_permission(user_id, resource, action):
    # proceed

from netrun.rbac import (
    # One-line middleware setup
    setup_tenancy_middleware, TenancyConfig, IsolationMode,
    # Tenant isolation testing
    TenantTestContext, TenantEscapePathScanner, ci_fail_on_findings,
)

# One-line middleware setup
app = FastAPI()
setup_tenancy_middleware(
    app,
    config=TenancyConfig(isolation_mode=IsolationMode.HYBRID),
    get_session=get_db_session
)

# Contract test: Prove tenant isolation works
async with TenantTestContext(db_session) as ctx:
    secret = Item(name="Secret", tenant_id=ctx.tenant_a_id)
    session.add(secret)
    await session.commit()

    await ctx.switch_to_tenant_b()
    result = await session.execute(select(Item))
    assert len(result.scalars().all()) == 0  # Cross-tenant access impossible

# CI/CD escape path scanner
scanner = TenantEscapePathScanner()
findings = scanner.scan_directory("./src")
sys.exit(ci_fail_on_findings(findings))  # Fails build on critical findings

from netrun.cache import CacheManager, cached

cache = CacheManager()

@cached(ttl=300, namespace="users")
async def get_user(user_id: str):
    return await db.fetch_user(user_id)

from netrun.resilience import retry, circuit_breaker, timeout

@retry(max_attempts=3, backoff_factor=2.0)
@circuit_breaker(failure_threshold=5)
@timeout(seconds=10)
async def call_external_api():
    return await client.fetch_data()

from netrun.validation import validate_ip_address, validate_api_key_format
from netrun.validation.custom_types import SafeEmail, StrongPassword

class UserConfig(BaseModel):
    email: SafeEmail
    password: StrongPassword

from netrun.websocket import WebSocketSessionManager

manager = WebSocketSessionManager(redis_url="redis://localhost:6379")
await manager.initialize()

connection_id = await manager.create_connection(
    websocket=ws,
    user_id="user_123",
    metadata={"room": "general"}
)

# Before (v1.x)
from netrun_auth import JWTHandler
from netrun_config import SecretCache
from netrun_llm import LLMFallbackChain

# After (v2.x) - recommended
from netrun.auth import JWTHandler
from netrun.config import SecretCache
from netrun.llm import LLMFallbackChain