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/v3@latest

Client Setup

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

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>3.2.0</version>
</dependency>

Client Setup

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

AxonFlow client = new AxonFlow(
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
PolicyNotFoundErrorPolicy ID doesn't exist
ValidationErrorInvalid request (bad pattern, missing fields)
ForbiddenErrorCannot modify system policy patterns
UnauthorizedErrorInvalid credentials

TypeScript Example

try {
await client.deleteStaticPolicy('sys_sqli_union_select');
} catch (error) {
if (error.code === 'FORBIDDEN') {
console.log('Cannot delete system policies');
} else if (error.code === 'NOT_FOUND') {
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: 'custom',
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 PolicyNotFoundError, ForbiddenError

try:
await client.delete_static_policy("sys_sqli_union_select")
except ForbiddenError:
print("Cannot delete system policies")
except PolicyNotFoundError:
print("Policy not found")

Java Example

try {
client.deleteStaticPolicy("sys_sqli_union_select");
} catch (ForbiddenException e) {
System.out.println("Cannot delete system policies");
} catch (PolicyNotFoundException e) {
System.out.println("Policy not found");
} catch (AxonFlowException e) {
System.out.println("Error: " + e.getMessage());
}