Skip to main content

Policy SDK Methods

All AxonFlow SDKs provide methods for policy management. This guide covers CRUD operations across TypeScript, Python, Go, and Java.

Method Overview

OperationTypeScriptPythonGoJava
List policieslistStaticPolicies()list_static_policies()ListStaticPolicies()listStaticPolicies()
Get policygetStaticPolicy()get_static_policy()GetStaticPolicy()getStaticPolicy()
Create policycreateStaticPolicy()create_static_policy()CreateStaticPolicy()createStaticPolicy()
Update policyupdateStaticPolicy()update_static_policy()UpdateStaticPolicy()updateStaticPolicy()
Delete policydeleteStaticPolicy()delete_static_policy()DeleteStaticPolicy()deleteStaticPolicy()
Toggle policytoggleStaticPolicy()toggle_static_policy()ToggleStaticPolicy()toggleStaticPolicy()
Get effectivegetEffectiveStaticPolicies()get_effective_static_policies()GetEffectiveStaticPolicies()getEffectiveStaticPolicies()
Test patterntestPattern()test_pattern()TestPattern()testPattern()

TypeScript SDK

Installation

npm install @axonflow/sdk

Client Setup

import { AxonFlow } from '@axonflow/sdk';

const client = new AxonFlow({
endpoint: 'http://localhost:8080',
clientId: 'my-tenant',
clientSecret: 'my-secret', // Optional for self-hosted
});

List Policies

// List all policies
const policies = await client.listStaticPolicies();

// With filters
const filtered = await client.listStaticPolicies({
tier: 'system',
category: 'security-sqli',
enabled: true,
limit: 50,
offset: 0,
});

// Response type
interface StaticPolicy {
id: string;
name: string;
description?: string;
category: string;
tier: 'system' | 'organization' | 'tenant';
pattern: string;
action: 'block' | 'redact' | 'warn' | 'log';
severity: 'critical' | 'high' | 'medium' | 'low';
enabled: boolean;
hasOverride?: boolean;
}

Get Single Policy

const policy = await client.getStaticPolicy('sys_sqli_union_select');
console.log(policy.pattern); // (?i)\bUNION\s+(ALL\s+)?SELECT\b

Create Policy

// Use an existing category that best fits your use case
const policy = await client.createStaticPolicy({
name: 'Block Internal IPs',
description: 'Prevent exposure of internal IP addresses',
category: 'security-admin', // security-sqli, security-admin, pii-global, pii-us, pii-eu, pii-india, code-secrets, code-unsafe, code-compliance
pattern: '\\b10\\.\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}\\b',
action: 'block',
severity: 'medium',
enabled: true,
});

console.log(`Created policy: ${policy.id}`);

Update Policy

const updated = await client.updateStaticPolicy('my-policy-id', {
enabled: false,
action: 'warn',
});

Delete Policy

await client.deleteStaticPolicy('my-policy-id');

Toggle Policy

// Disable
await client.toggleStaticPolicy('my-policy-id', false);

// Enable
await client.toggleStaticPolicy('my-policy-id', true);

Get Effective Policies

const effective = await client.getEffectiveStaticPolicies({
category: 'pii-global',
includeDisabled: false,
});

Test Pattern

const result = await client.testPattern({
pattern: '\\b[A-Z]{5}[0-9]{4}[A-Z]\\b',
inputs: [
'My PAN is ABCDE1234F',
'No PAN here',
],
});

console.log(result.matches);
// [{ input: 'My PAN is ABCDE1234F', matched: true, matchedText: 'ABCDE1234F' }]

Python SDK

Installation

pip install axonflow

Client Setup

from axonflow import AxonFlow

client = AxonFlow(
endpoint="http://localhost:8080",
client_id="my-tenant",
client_secret="my-secret", # Optional for self-hosted
)

List Policies

# List all policies
policies = await client.list_static_policies()

# With filters
filtered = await client.list_static_policies(
tier="system",
category="security-sqli",
enabled=True,
limit=50,
)

for policy in filtered:
print(f"{policy.name}: {policy.action}")

Get Single Policy

policy = await client.get_static_policy("sys_sqli_union_select")
print(policy.pattern)

Create Policy

# Use an existing category that best fits your use case
policy = await client.create_static_policy(
name="Block Internal IPs",
description="Prevent exposure of internal IP addresses",
category="security-admin", # security-sqli, security-admin, pii-global, pii-us, pii-eu, pii-india, code-secrets, code-unsafe, code-compliance
pattern=r"\b10\.\d{1,3}\.\d{1,3}\.\d{1,3}\b",
action="block",
severity="medium",
enabled=True,
)

print(f"Created policy: {policy.id}")

Update Policy

updated = await client.update_static_policy(
policy_id="my-policy-id",
enabled=False,
action="warn",
)

Delete Policy

await client.delete_static_policy("my-policy-id")

Sync Usage

The Python SDK supports both async and sync usage patterns:

# Synchronous usage (blocking)
from axonflow import AxonFlow

client = AxonFlow.sync(
endpoint="http://localhost:8080",
client_id="my-tenant",
client_secret="my-secret",
)

# All methods work synchronously — no await needed
policies = client.list_static_policies()
for policy in policies:
print(f"{policy.name}: {policy.action}")

policy = client.create_static_policy(
name="Block Internal IPs",
description="Prevent exposure of internal IP addresses",
category="security-admin",
pattern=r"\b10\.\d{1,3}\.\d{1,3}\.\d{1,3}\b",
action="block",
severity="medium",
enabled=True,
)
print(f"Created: {policy.id}")

Async Usage

import asyncio
from axonflow import AxonFlow

async def main():
client = AxonFlow(
endpoint="http://localhost:8080",
client_id="my-tenant",
client_secret="my-secret",
)

# All methods return coroutines — use await
policies = await client.list_static_policies(
tier="system",
category="security-sqli",
)

# Concurrent operations with asyncio.gather
sqli_policies, pii_policies = await asyncio.gather(
client.list_static_policies(category="security-sqli"),
client.list_static_policies(category="pii-global"),
)

print(f"SQLi: {len(sqli_policies)}, PII: {len(pii_policies)}")

asyncio.run(main())

Go SDK

Installation

go get github.com/getaxonflow/axonflow-sdk-go/v8@latest

Client Setup

import "github.com/getaxonflow/axonflow-sdk-go/v8"

client := axonflow.NewClient(axonflow.AxonFlowConfig{
Endpoint: "http://localhost:8080",
ClientID: "my-tenant",
ClientSecret: "my-secret",
})

List Policies

// List all policies
policies, err := client.ListStaticPolicies(nil)
if err != nil {
log.Fatal(err)
}

// With filters
enabled := true
options := &axonflow.ListStaticPoliciesOptions{
Tier: axonflow.TierSystem,
Category: axonflow.CategorySecuritySQLI,
Enabled: &enabled,
Limit: 50,
}
filtered, err := client.ListStaticPolicies(options)

Get Single Policy

policy, err := client.GetStaticPolicy("sys_sqli_union_select")
if err != nil {
log.Fatal(err)
}
fmt.Println(policy.Pattern)

Create Policy

// Use an existing category that best fits your use case
policy, err := client.CreateStaticPolicy(&axonflow.CreateStaticPolicyRequest{
Name: "Block Internal IPs",
Description: "Prevent exposure of internal IP addresses",
Category: "security-admin", // security-sqli, security-admin, pii-global, pii-us, pii-eu, pii-india, code-secrets, code-unsafe, code-compliance
Pattern: `\b10\.\d{1,3}\.\d{1,3}\.\d{1,3}\b`,
Action: axonflow.ActionBlock,
Severity: axonflow.SeverityMedium,
Enabled: true,
})
if err != nil {
log.Fatal(err)
}
fmt.Printf("Created policy: %s\n", policy.ID)

Update Policy

enabled := false
action := axonflow.ActionWarn
updated, err := client.UpdateStaticPolicy("my-policy-id", &axonflow.UpdateStaticPolicyRequest{
Enabled: &enabled,
Action: &action,
})

Delete Policy

err := client.DeleteStaticPolicy("my-policy-id")

Toggle Policy

// Disable
policy, err := client.ToggleStaticPolicy("my-policy-id", false)

// Enable
policy, err = client.ToggleStaticPolicy("my-policy-id", true)

Java SDK

Installation

<dependency>
<groupId>com.getaxonflow</groupId>
<artifactId>axonflow-sdk</artifactId>
<version>4.3.0</version>
</dependency>

Client Setup

import com.getaxonflow.sdk.AxonFlow;
import com.getaxonflow.sdk.AxonFlowConfig;

AxonFlow client = AxonFlow.create(AxonFlowConfig.builder()
.endpoint("http://localhost:8080")
.clientId("my-tenant")
.clientSecret("my-secret")
.build());

List Policies

// List all policies
List<StaticPolicy> policies = client.listStaticPolicies();

// With filters
ListStaticPoliciesOptions options = ListStaticPoliciesOptions.builder()
.tier(PolicyTier.SYSTEM)
.category(PolicyCategory.SECURITY_SQLI)
.enabled(true)
.limit(50)
.build();

List<StaticPolicy> filtered = client.listStaticPolicies(options);

Get Single Policy

StaticPolicy policy = client.getStaticPolicy("sys_sqli_union_select");
System.out.println(policy.getPattern());

Create Policy

// Use an existing category that best fits your use case
CreateStaticPolicyRequest request = CreateStaticPolicyRequest.builder()
.name("Block Internal IPs")
.description("Prevent exposure of internal IP addresses")
.category("security-admin") // security-sqli, security-admin, pii-global, pii-us, pii-eu, pii-india, code-secrets, code-unsafe, code-compliance
.pattern("\\b10\\.\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}\\b")
.action(PolicyAction.BLOCK)
.severity(PolicySeverity.MEDIUM)
.enabled(true)
.build();

StaticPolicy policy = client.createStaticPolicy(request);
System.out.println("Created policy: " + policy.getId());

Update Policy

UpdateStaticPolicyRequest update = UpdateStaticPolicyRequest.builder()
.enabled(false)
.action(PolicyAction.WARN)
.build();

StaticPolicy updated = client.updateStaticPolicy("my-policy-id", update);

Delete Policy

client.deleteStaticPolicy("my-policy-id");

Error Handling

All SDKs throw/return errors for common scenarios:

ErrorDescription
AxonFlowError / AxonFlowExceptionBase error for all SDK errors
AuthenticationError / AuthenticationExceptionInvalid or missing credentials
PolicyViolationError / PolicyViolationExceptionRequest blocked by policy
RateLimitErrorRate limit exceeded (Python/TypeScript/Go)

TypeScript Example

import { AxonFlowError } from '@axonflow/sdk';

try {
await client.deleteStaticPolicy('sys_sqli_union_select');
} catch (error) {
if (error instanceof AxonFlowError) {
if (error.statusCode === 403) {
console.log('Cannot delete system policies');
} else if (error.statusCode === 404) {
console.log('Policy not found');
} else {
console.error(`Unexpected error: ${error.message}`);
}
}
}

TypeScript Async/Await Pattern

All TypeScript SDK methods return Promises and should be used with async/await:

async function managePolicies() {
const client = new AxonFlow({
endpoint: 'http://localhost:8080',
clientId: 'my-tenant',
});

// Sequential operations
const policy = await client.createStaticPolicy({
name: 'Block Test Pattern',
category: 'sensitive-data',
pattern: '\\btest_data\\b',
action: 'warn',
severity: 'low',
enabled: true,
});

// Concurrent reads with Promise.all
const [systemPolicies, tenantPolicies] = await Promise.all([
client.listStaticPolicies({ tier: 'system' }),
client.listStaticPolicies({ tier: 'tenant' }),
]);

console.log(`System: ${systemPolicies.length}, Tenant: ${tenantPolicies.length}`);
}

Go Example

import "errors"

err := client.DeleteStaticPolicy("sys_sqli_union_select")
if err != nil {
var httpErr *axonflow.HTTPError
if errors.As(err, &httpErr) {
switch httpErr.StatusCode {
case 403:
log.Println("Cannot delete system policies")
case 404:
log.Println("Policy not found")
default:
log.Printf("HTTP error %d: %s", httpErr.StatusCode, httpErr.Message)
}
} else {
// Network or other non-HTTP error
log.Printf("Unexpected error: %v", err)
}
}

Python Example

from axonflow.exceptions import AxonFlowError, AuthenticationError, PolicyViolationError, RateLimitError

try:
await client.delete_static_policy("sys_sqli_union_select")
except AuthenticationError:
print("Invalid credentials")
except PolicyViolationError:
print("Cannot delete system policies")
except AxonFlowError as e:
print(f"Error: {e}")

Java Example

import com.getaxonflow.sdk.exceptions.AxonFlowException;
import com.getaxonflow.sdk.exceptions.AuthenticationException;
import com.getaxonflow.sdk.exceptions.PolicyViolationException;

try {
client.deleteStaticPolicy("sys_sqli_union_select");
} catch (AuthenticationException e) {
System.out.println("Invalid credentials");
} catch (PolicyViolationException e) {
System.out.println("Cannot delete system policies");
} catch (AxonFlowException e) {
System.out.println("Error: " + e.getMessage());
}