Python SDK

pip install agimus

from agimus import AgimusClient

client = AgimusClient(api_key="agm_your_key_here")

# Query objects
customers = client.objects("Customer").filter(status="active").all()

# Get by primary key
customer = client.objects("Customer").get(123)

# Create
client.objects("Customer").create({"customerId": 1, "name": "Acme Corp"})

# Update
client.objects("Customer").update(1, {"status": "premium"})

# Delete
client.objects("Customer").delete(1)

client = AgimusClient(
    api_key="agm_...",
    timeout=30.0  # Optional: request timeout in seconds
)

# Equals (default)
.filter(status="active")

# Comparison
.filter(age__gt=18)        # greater than
.filter(age__gte=18)       # greater than or equal
.filter(age__lt=65)        # less than
.filter(age__lte=65)       # less than or equal
.filter(age__ne=0)         # not equal

# Lists
.filter(region__in=["US", "EU"])      # in list
.filter(status__nin=["deleted"])      # not in list

# Strings
.filter(name__like="Acme%")           # SQL LIKE
.filter(name__ilike="%acme%")         # case-insensitive
.filter(name__starts_with="A")        # starts with

# Null checks
.filter(deletedAt__is_null=True)
.filter(verifiedAt__is_not_null=True)

# Multiple filters (AND)
.filter(status="active", region="US")

.sort("name")                    # ascending
.sort("-createdAt")              # descending (prefix with -)
.sort("-createdAt", "name")      # multiple fields

# Return only specific fields
.fields("customerId", "name", "email").all()

.expand("orders").all()
.expand("orders", "orders.items").all()  # nested

# Or on single object fetch
.get(123, expand=["orders"])

# Limit total results
.limit(100).all()

# Set page size for API calls (max 100, default 50)
.page_size(25).all()

# Auto-pagination with iteration
for customer in client.objects("Customer").filter(status="active"):
    print(customer["name"])

.all()      # Get all results as list
.first()    # Get first result (or None)
.exists()   # Check if any results exist
.count()    # Get count of matching objects
.iter()     # Iterator with auto-pagination

result = client.objects("Order").filter(status="completed").aggregate(
    metrics=[
        {"op": "count", "alias": "orderCount"},
        {"op": "sum", "field": "total", "alias": "revenue"},
        {"op": "avg", "field": "total", "alias": "avgOrder"},
    ],
    group_by=[
        {"field": "region"},
        {"field": "createdAt", "granularity": "month"}
    ],
    sort=["-revenue"],
    limit=100
)

# Customer -> orders (forward link)
orders = client.objects("Customer").links(123, "orders")
# Returns: {"data": [...], "hasMore": False}

# Order -> customer (reverse link)
customer = client.objects("Order").links(456, "customer")

# With pagination
orders = client.objects("Customer").links(123, "orders", page_size=10, offset=0)

# Count related objects
count = client.objects("Customer").count_links(123, "orders")

customer = client.objects("Customer").create({
    "customerId": 1,        # PK required
    "name": "Acme Corp",    # non-nullable fields required
    "email": "contact@acme.com",
    "status": "active"
})

# Partial update - only specified fields change
updated = client.objects("Customer").update(1, {
    "status": "premium"
})

# Create if not exists, update if exists
customer = client.objects("Customer").upsert(1, {
    "name": "Acme Corp",
    "status": "active"
})

deleted = client.objects("Customer").delete(1)  # Returns: True

result = client.objects("Customer").batch([
    {"op": "create", "data": {"customerId": 1, "name": "Customer 1"}},
    {"op": "update", "pk": 2, "data": {"status": "active"}},
    {"op": "delete", "pk": 3},
])
# Returns: {"results": [...], "succeeded": 2, "failed": 1}

# List all accessible entities
entities = client.list_entities()
for e in entities:
    print(f"{e['apiName']}: {e['displayName']}")

# Get full entity schema
schema = client.get_entity_schema("Customer")
print(f"Primary key: {schema['primaryKey']}")

for prop in schema["properties"]:
    print(f"  {prop['apiName']}: {prop['baseType']}")

for link in schema["links"]:
    print(f"  -> {link['apiName']}: {link['targetEntity']}")

from agimus import AsyncAgimusClient

async with AsyncAgimusClient(api_key="agm_...") as client:
    customers = await client.objects("Customer").filter(status="active").all()
    customer = await client.objects("Customer").get(123)

    # Async iteration
    async for customer in client.objects("Customer").filter(status="active"):
        print(customer["name"])

    # Write operations
    await client.objects("Customer").create({"customerId": 1, "name": "Acme"})
    await client.objects("Customer").update(1, {"status": "premium"})
    await client.objects("Customer").delete(1)

from agimus import (
    AgimusError,          # Base class for all errors
    AuthenticationError,  # Invalid or missing API key (401)
    AccessDeniedError,    # Permission denied (403)
    NotFoundError,        # Entity or object not found (404)
    ValidationError,      # Invalid request data (400/422)
    RateLimitError,       # Rate limit exceeded (429)
    ServerError,          # Server error (5xx)
)

try:
    customer = client.objects("Customer").get(999)
except NotFoundError as e:
    print(f"Not found: {e.entity}")
except ValidationError as e:
    print(f"Validation error: {e.message}, field: {e.field}")
except AuthenticationError:
    print("Invalid API key")
except RateLimitError as e:
    print(f"Rate limited. Retry after: {e.retry_after}s")
except AgimusError as e:
    print(f"API error: {e.message}")