EN SDK Sync and Async

CRUD, RPC and pagination with the SDK

English · Türkçe · Languages / Diller

Scope: JSON API Forge 0.5.1 · Source review: 17 September 2026 · Language: English

Synchronous CRUD

import os
from json_api_forge import ForgeClient

with ForgeClient("https://api.example.com", api_key=os.environ["FORGE_API_KEY"]) as forge:
    created = forge.create_item("wiki-demo", "items", {"name": "SDK record"})
    print(created.status_code, created.data, created.request_id)
    page = forge.list_items("wiki-demo", "items", params={"limit": 20, "sort": "id"})
    item_id = str(page.data["items"][0]["id"])
    forge.get_item("wiki-demo", "items", item_id)
    forge.update_item("wiki-demo", "items", item_id, {"name": "Changed"})
    forge.delete_item("wiki-demo", "items", item_id)

replace=True selects PUT in update_item; PATCH is the default. Convert IDs to strings. Responses are ForgeResponse objects rather than bare dictionaries. data reflects the server JSON; do not assume every endpoint returns items.

Async usage

import asyncio
import os
from json_api_forge import AsyncForgeClient

async def main():
    async with AsyncForgeClient(
        "https://api.example.com", api_key=os.environ["FORGE_API_KEY"]
    ) as forge:
        result = await forge.list_items("wiki-demo", "items", params={"limit": 20})
        print(result.data)

if __name__ == "__main__":
    asyncio.run(main())

Inside an existing event loop, such as an async web application, await the coroutine instead of calling asyncio.run. Do not put blocking synchronous client calls directly on the event loop.

Calling operations

result = forge.call_operation(
    "wiki-demo", "items.rename", {"id": 1, "name": "New name"},
    idempotency_key="rename-command-001",
)

This helper sends POST /api/<project>/v1/rpc/<operation>. For a custom path or GET operation:

result = forge.request(
    "GET", "api/guild-ledger/v1/rpc/ledger.balance/member-123"
)

An operation.name is not necessarily the entire URL. call_operation does not discover every operation's configured method/path from its name.

Offset iteration

for item in forge.iter_items("wiki-demo", "items", page_size=50, max_items=500):
    print(item)

In v0.5.1 this helper exists only on ForgeClient. AsyncForgeClient has no matching async iterator. The helper starts at offset 0; an offset inside params is not a resume mechanism. Choose page_size within the server's max_limit: a server-clamped short page can be interpreted as the last page.

Implement a bounded offset loop for async use. For cursor resources, follow next_cursor:

async def read_cursor_pages(forge):
    cursor = None
    for _ in range(100):
        params = {"limit": 50}
        if cursor is not None:
            params["cursor"] = cursor
        response = await forge.list_items("notes", "notes", params=params)
        for item in response.data["items"]:
            yield item
        cursor = response.data.get("next_cursor")
        if not response.data.get("has_more") or cursor is None:
            break

This example assumes a server-side notes resource configured for cursor pagination. Handle repeated cursors, deleted records and concurrent inserts according to your application's consistency requirements.

request supports JSON, multipart files, query parameters, headers and binary/text responses with expect_json=False. JSON bodies and files cannot be sent together. This buffered HTTP API is not an unlimited SSE/WebSocket streaming client.

Sources and verification

This page describes the behavior at the pinned commits below. Recheck these contracts when upgrading.

• python-library/json_api_forge/client.py
• python-library/json_api_forge/models.py

---

English home · All English pages · Version maintenance