Metadata-Version: 2.4
Name: b24api
Version: 2.2.0
Summary: Bitrix24 API
Requires-Python: >=3.12
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: httpx[http2]>=0.28.1
Requires-Dist: pydantic>=2.11.7
Requires-Dist: pydantic-settings>=2.10.1
Dynamic: license-file

# b24api 2.x

`b24api` is a thin asynchronous Bitrix24 REST client for Python 3.12+. It knows how to send
requests, split logical batches, traverse lists, retry safely, preserve caller correlation and
close resources. It does not contain a Tasks, CRM or IM method catalog and does not impose
application storage rules.

## Install and configure

```console
uv sync --frozen
export BITRIX24_API_WEBHOOK_URL='https://portal.example/rest/.../'
```

Keep the webhook out of source, logs and command arguments. Reuse one client for a related unit of
work so its HTTP/2 connection pool and rate state are reused.

<!-- tested: tests/client_v2_test.py::test_call_and_call_response_have_stable_detached_types -->
```python
from b24api import Bitrix24, Request

async with Bitrix24() as client:
    profile = await client.call(Request("profile"))
```

The client owns its default transport. An injected transport remains caller-owned. `aclose()` is
idempotent and closes active streams before the owned transport.

## Direct calls

Use `call()` for detached decoded JSON and `call_response()` when you also need the immutable
response envelope: `result`, `total`, `next`, timing and bounded diagnostic evidence.

Use `call_bytes()` when a successful method response is a file rather than a Bitrix JSON envelope.
The operation is explicit and never hides malformed JSON by falling back to bytes.

<!-- tested: tests/client_findings_3_test.py::test_binary_call_returns_every_success_byte_without_json_sniffing -->
```python
archive = await client.call_bytes(Request("example.export.download", replay_safety=ReplaySafety.SAFE))
payload = archive.body
```

<!-- tested: tests/client_v2_test.py::test_call_and_call_response_have_stable_detached_types -->
```python
from b24api import ReplaySafety

request = Request("example.item.get", {"id": 7}, ReplaySafety.SAFE)
decoded = await client.call(request)
response = await client.call_response(request)
```

### Replay safety

`Request.replay_safety` describes what the client may do when a connection fails after the request
may already have reached Bitrix:

| Value | Meaning | After possible dispatch |
|---|---|---|
| `SAFE` | Repeating the request cannot create a second business effect. Typical reads and explicitly idempotent operations belong here. | Automatic retry is allowed within policy budgets. |
| `UNSAFE` | Repeating the request is known to risk a duplicate effect, for example creating an entity without an idempotency key. | No automatic replay; the caller receives an ambiguous-execution error and reconciles state. |
| `UNKNOWN` | The caller has not established whether replay is safe. This is the default. | Same conservative behavior as `UNSAFE`, while diagnostics preserve that safety was unknown rather than known unsafe. |

A failure proved to occur before dispatch may still be retried. Method names never imply safety;
mark a request `SAFE` only when the operation's semantics justify it.

Use `ExecutionPolicy` to narrow attempts or resource budgets for one operation:

<!-- tested: tests/execution_test.py::test_ambiguous_dispatch_never_retries_unproven_request -->
```python
from b24api import ExecutionPolicy

one_attempt = ExecutionPolicy(max_attempts_per_request=1)
result = await client.call(request, policy=one_attempt)
```

## Logical batch and correlation

`batch()` accepts an arbitrary-length synchronous or asynchronous command source. It consumes the
source incrementally and splits it into physical Bitrix batches of at most 50 commands; the full
input is never materialized.

`Command.correlation` is arbitrary caller-owned state. It is retained by reference, returned with
the outcome, never serialized to Bitrix and never included in safe diagnostics. This is useful for
matching a result to the object, file, chat or database row that produced its request.

<!-- tested: tests/client_v2_test.py::test_logical_batch_is_unbounded_ordered_and_correlation_is_strictly_off_wire -->
```python
from b24api import Command, CommandSuccess

commands = (
    Command(
        Request("example.item.get", {"id": item_id}, ReplaySafety.SAFE),
        correlation=item_id,
    )
    for item_id in source_ids
)

async with client.batch(commands, batch_size=25) as stream:
    async for outcome in stream:
        assert isinstance(outcome, CommandSuccess)
        consume(outcome.correlation, outcome.result)
```

`batch()` is fail-fast. `batch_outcomes()` continues where safe and yields one of
`CommandSuccess`, `CommandFailure`, `CommandNotExecuted` or `CommandOutcomeUnknown` in input order.

<!-- tested: tests/client_v2_test.py::test_batch_outcomes_retains_typed_failure_without_halting_later_commands -->
```python
from b24api import CommandFailure, CommandNotExecuted, CommandOutcomeUnknown

async with client.batch_outcomes(commands) as stream:
    async for outcome in stream:
        match outcome:
            case CommandSuccess() as success:
                consume(success.correlation, success.result)
            case CommandFailure() | CommandNotExecuted() | CommandOutcomeUnknown():
                handle(outcome)
```

For independently dispatchable commands, use `fan_out()` or `fan_out_outcomes()` with
`DirectDispatch` or `BatchDispatch`. Delivery order is explicitly `READY` or `INPUT`.

## Choosing a list operation

The unsuffixed operation is the basic strategy with the fewest endpoint assumptions. Faster or
more specialized mechanics have explicit names and explicit preconditions.

| Operation | Use it when | Network mechanics | Completion proof |
|---|---|---|---|
| `iter_list` | The method supports ordinary offset pagination. | Sequential requests follow server `next` (the next offset). Ordinary counted Bitrix list endpoints do server-side COUNT for `total` plus LIMIT/OFFSET page retrieval. | Continuation and empty terminal page; add identity for duplicate detection. |
| `iter_list_counted` | The first response provides an exact filtered `total` and stable offset pages. | Head page is direct; all known tail offsets are grouped into physical Bitrix batches. | Exact total, ranges and identities. |
| `iter_list_keyset` | The method may omit `total`, but reliably supports ordering and filtering by a unique integer identity. | Auto by default: it plans first, then selects boundary-only, sequential, range, or partitioned execution; runtime sends no diagnostic canaries. | Caller-asserted keyset contract, strict monotonic identity, active bound validation, and terminal empty confirmation. |
| `iter_list_cursor` | Each next request depends on a cursor from the previous response. | Sequential dependent cursor requests. | Strict unique monotonic cursor and empty terminal page. |
| `iter_cursors` | One or many parent-bound cursor traversals need correlation and shared batching. | Lazy per-parent drivers share the reference batch queue; each binding may have `start_cursor`. | Isolated strict cursor progress and terminal event per binding. |
| `iter_references` | The same list method must run for many parent parameter sets, such as comments per owner or messages per chat. | Bindings are scheduled with direct or physical-batch dispatch; each binding has its own traversal state. | Per-binding rows, completion/failure and caller correlation. |

`page_size` is a local decoded-page cap. It is sent to Bitrix only when you provide the endpoint's
exact `limit_path`; the client never guesses method-specific parameter names.

### List traversal comparison

![List traversal comparison](list-traversal-comparison.svg)

### Sequential offset

This is the canonical default. It follows the `next` returned by the server and confirms the end
with an empty page. For ordinary counted Bitrix list endpoints, `next` is an offset for the next
LIMIT/OFFSET page, not a keyset cursor. Producing `total` involves a separate server-side count
query in addition to retrieving the page. These database operations are performed inside the same
REST request: the client does not issue an additional HTTP call just for the count. This distinction
matters for performance: not making a separate HTTP count call does **not** mean avoiding server-side
COUNT work. `iter_list` does not suppress that work; a returned `total` is observational and does not
control this strategy's completion. Exact database implementation is endpoint-specific.

<!-- tested: tests/client_v2_test.py::test_iter_list_is_sequential_mechanics_only_and_report_is_post_cleanup -->
```python
from b24api import IdentityCoercion, IdentitySpec, ResultSelector

identity = IdentitySpec(
    item_path=("ID",),
    filter_key="ID",
    order_key="ID",
    coercion=IdentityCoercion.DECIMAL_STRING_INTEGER,
)

stream = client.iter_list(
    Request("example.item.list", replay_safety=ReplaySafety.SAFE),
    selector=ResultSelector(("items",)),
    identity=identity,
)
async with stream:
    async for item in stream:
        consume(item)
```

Without `identity`, successful exhaustion is reported as `MECHANICS_ONLY`: pagination completed,
but the client cannot prove that the portal did not duplicate or substitute rows.

Mapping-backed collections are explicit as well. `MAPPING_VALUES` yields values from a selected
mapping in insertion order; `MAPPING_VALUES_OR_EMPTY` additionally accepts only an empty terminal
sequence and records that degradation in the operation report.

<!-- tested: tests/client_findings_3_test.py::test_shape_rejection_is_retained_as_zero_admission_page_evidence -->
```python
from b24api import ResultCollectionShape

stream = client.iter_list(
    Request("example.dictionary.list", replay_safety=ReplaySafety.SAFE),
    selector=ResultSelector(("items",)),
    collection_shape=ResultCollectionShape.MAPPING_VALUES,
)
async with stream:
    async for value in stream:
        consume(value)
```

### Counted, physically batched tail

The first direct page must contain an exact filtered `total` and, when more rows exist, `next`.
The client derives all remaining offsets from the observed head width and sends tail pages through
bounded physical batches.

<!-- tested: tests/client_v2_test.py::test_counted_traversal_preserves_frozen_request_shape_and_exact_identity -->
```python
stream = client.iter_list_counted(
    Request("example.item.list", replay_safety=ReplaySafety.SAFE),
    selector=ResultSelector(("items",)),
    identity=identity,
    page_size=50,
    batch_size=50,
)
```

Use it only when `total` is exact for the supplied filter and offset pages are stable. Any missing
range, overlap, duplicate identity or total contradiction raises `IncompleteTraversalError`.

Physical batching reduces HTTP exchanges, but does not suppress server-side COUNT in ordinary
counted list subrequests. Each command still performs its own offset page retrieval and associated
total calculation on the server. Do not confuse batching these commands with a no-count traversal.

### No-count keyset

Keyset traversal uses automatic execution by default. Omitting `execution` is equivalent to
`AutoKeysetExecution(StableIntegerKeysetContract())`: the client captures both ordered boundaries,
then selects boundary-only, sequential, range, or partitioned execution from the observed geometry
and available policy capacity. Planning completes before any row is emitted, so partial consumption
still pays that barrier cost. A sequential selection made by auto is a cost decision; a failed or
contradictory fast plan is never silently restarted as sequential.

By using the default, the caller asserts that the endpoint has a stable, unique integer key, honors
strict numeric bounds and ordering, and satisfies empty-confirmation completion. Concurrent mutation
outside the captured middle is handled by the finishing sweep; mutation inside it is outside this
assertion. Pass `SequentialKeysetExecution()` explicitly when an endpoint cannot satisfy the fast
contract or when the previous request-by-request behavior is required.

Static incompatibility with the auto contract raises `CapabilityError` from the
`iter_list_keyset(...)` call before iteration begins. A portal that accepts but contradicts the
declared controls fails before emission with `IncompleteTraversalError`; auto never restarts that
operation silently. `KeysetTraversal` inside reference traversal remains sequential-only. The
terminal report now includes `keyset_execution` for omitted-execution keyset calls so consumers can
see the requested and selected plan.

Use `await client.verify_keyset_capability(...)` as a development/CI/staging guard on a stable
representative fixture. It performs five strict-bound checks and returns only a `VERIFIED` report;
unsupported and inconclusive verdicts raise `KeysetCapabilityError`. The ordinary
`iter_list_keyset()` remains a separate caller-asserted operation with zero verifier canaries and
may emit a partial prefix before a late endpoint contradiction is detected.

Every list operation also accepts an immutable `PageAdapter` strategy. The adapter synchronously
maps selected frozen items using sibling result metadata while preserving cardinality, order and
configured identities. The identity adapter is the default and preserves existing JSON output.

An advisory `total` may only raise an automatic cost estimate and never proves completion. Reports
record the selected strategy and reason: unbounded auto continuation has the same
`ordered_prefix_only` assurance as sequential traversal, while bounded plans additionally report
`caller_asserted_bounds`. Runtime traversal does not perform verifier canaries.

<!-- tested: tests/keyset_fast_test.py::test_omitted_execution_defaults_to_auto -->
```python
from b24api import KeysetSpec, ParameterPath

stream = client.iter_list_keyset(
    Request("example.item.list", replay_safety=ReplaySafety.SAFE),
    selector=ResultSelector(("items",)),
    identity=identity,
    keyset=KeysetSpec(
        filter_path=ParameterPath(("filter",)),
        order_path=ParameterPath(("order",)),
    ),
)
```

### Dependent cursor

Use a cursor when the next boundary is returned or derived from the previous page, as with many
message-list methods.

<!-- tested: tests/client_v2_test.py::test_keyset_and_cursor_are_explicit_strict_alternatives -->
```python
from b24api import CursorSpec, ParameterPath

stream = client.iter_list_cursor(
    Request("example.message.list", replay_safety=ReplaySafety.SAFE),
    selector=ResultSelector(("items",)),
    cursor=CursorSpec(
        parameter_path=ParameterPath(("LAST_ID",)),
        item_path=("ID",),
        coercion=IdentityCoercion.DECIMAL_STRING_INTEGER,
        direction="ascending",
        take="last",
    ),
)
```

Cursor values must be unique and strictly monotonic. If an endpoint exposes only a non-unique
boundary, use an application-owned direct-call workflow or supply a unique tie-breaker.

For multiple parent-bound cursor chains, `iter_cursors()` keeps cursor progress and correlation
isolated per binding while ready pages share the physical batch queue.

![Cursor batching across independent chats](cursor-batching.svg)

See [architecture](docs/architecture.md), [migration](docs/migration.md),
[performance](docs/performance.md), and [endpoint recipes](docs/recipes.md) for the complete
contracts and selection guidance.

### One list method across many parent entities

`Binding` applies exact parameter updates to a base request and carries parent correlation. The
client remains unaware of entity types: a binding can represent a deal, lead, chat or any other
caller-defined parent.

![Reference batching across leads and deals](references-batching.svg)

<!-- tested: tests/client_v2_test.py::test_bound_references_apply_nested_updates_off_wire_and_emit_exact_completion -->
```python
from b24api import (
    BatchDispatch,
    Binding,
    ParameterPath,
    ParameterUpdate,
    ReferenceComplete,
    ReferenceItem,
    SequentialTraversal,
)

bindings = (
    Binding(
        summary=f"owner {parent_id}",
        updates=(ParameterUpdate(ParameterPath(("filter", "OWNER_ID")), parent_id),),
        correlation=parent_id,
    )
    for parent_id in parent_ids
)

stream = client.iter_references(
    Request("example.comment.list", replay_safety=ReplaySafety.SAFE),
    bindings,
    traversal=SequentialTraversal(selector=ResultSelector(("items",)), identity=identity),
    dispatch=BatchDispatch(batch_size=25, concurrency=2),
)
async with stream:
    async for event in stream:
        if isinstance(event, ReferenceItem):
            consume(event.correlation, event.item)
        elif isinstance(event, ReferenceComplete):
            record_completion(event.correlation, event.row_count)
```

For messages across chats, use the same `iter_references()` shape: each binding updates the chat
parameter and carries the chat correlation; choose `CursorTraversal` when the message endpoint is
cursor-based. Identity tracking and completion remain scoped to each binding, so equal child IDs
under different parents are not conflated.

<!-- tested: tests/client_v2_test.py::test_bound_references_apply_nested_updates_off_wire_and_emit_exact_completion -->
```python
from b24api import (
    Binding,
    CursorSpec,
    CursorTraversal,
    DirectDispatch,
    IdentityCoercion,
    ParameterPath,
    ParameterUpdate,
    ResultSelector,
)

chat_bindings = (
    Binding(
        summary=f"chat {chat_id}",
        updates=(ParameterUpdate(ParameterPath(("DIALOG_ID",)), chat_id),),
        correlation={"chat_id": chat_id},
    )
    for chat_id in chat_ids
)

messages = client.iter_references(
    Request("example.message.list", replay_safety=ReplaySafety.SAFE),
    chat_bindings,
    traversal=CursorTraversal(
        selector=ResultSelector(("items",)),
        cursor=CursorSpec(
            parameter_path=ParameterPath(("LAST_ID",)),
            item_path=("ID",),
            coercion=IdentityCoercion.DECIMAL_STRING_INTEGER,
            direction="ascending",
            take="last",
        ),
    ),
    dispatch=DirectDispatch(concurrency=4),
)
```

`iter_reference_outcomes()` additionally yields correlated `ReferenceFailure`,
`ReferenceNotExecuted` and `ReferenceOutcomeUnknown`. A malformed source object that is not a
`Binding` has no valid caller correlation, so it terminates the source with `InputSourceError`
rather than fabricating a reference outcome. Already accepted bindings retain their real outcomes.

## Streams, partial results and reports

Every multi-item operation returns an `OperationStream`. Prefer `async with`: a plain `break` does
not close an arbitrary async iterator. After cleanup, `stream.report` permanently exposes one
immutable `OperationReport`; before termination it is `None`.

<!-- tested: tests/client_v2_test.py::test_partial_helper_closes_without_claiming_completion -->
```python
first = await client.iter_list(request).first()
page = await client.iter_list(request).collect(limit=100)

assert first.report.partial
assert page.report.partial
```

Helpers do not pull an extra row just to prove exhaustion. Reaching a requested limit is therefore
`EARLY_CLOSED`, never a false `COMPLETED`. Cancellation and cleanup preserve the primary exception
and publish the same final report where the Python exception type permits it.

## Resource boundaries

`ExecutionPolicy` bounds requests, pages, elapsed time, attempts, decompressed response bytes,
buffered commands and rows, direct concurrency and active references. The default response ceiling
is 16 MiB and is enforced while streaming, before JSON decoding.

Sequential and counted exact traversal retain observed identities in memory. There is no database,
spill file or identity-count refusal. Crossing 100,000 distinct identities emits one
`RuntimeWarning`; exact tracking continues. Strict keyset and cursor traversal retain only
monotonic progression state when sufficient.

## CLI

The wheel installs `b24api`. Stdout contains only result data; list rows are JSONL. Reports and safe
errors go to stderr. Credentials come only from `Settings` and cannot be passed as CLI arguments.

<!-- tested-console: tests/cli_test.py::test_call_routes_replay_safety_and_keeps_success_data_on_stdout -->
```console
b24api call profile
b24api call example.item.get --params '{"id":7}' --raw --replay-safety safe
b24api list example.item.list --params @params.json
b24api list example.item.list --strategy counted --contract @counted-contract.json
```

The `--raw` CLI option selects the response envelope; it does not alter the Python API. Advanced
list strategies use closed JSON `version: 1` contracts. The entire contract is validated before
client construction. Run `b24api --help` and `b24api list --help` for the compact option surface.

Exit codes are `0` success, `2` usage/contract error, `3` unavailable configuration, `4`
remote/protocol/correctness/incomplete failure, `5` broken output consumer and `130` cancellation.

## Correctness boundaries

The client fails closed on contradictory pagination, missing counted ranges, duplicate identities,
unsafe ambiguous replay, oversized responses and incomplete cleanup. It can prove only facts visible
through the transport contract: continuation, totals, identity, order, budgets and lifecycle.

It cannot generically prove that Bitrix honored the business meaning of a filter, choose an
application's composite storage key or reconcile an ambiguous write. Applications must validate
expected business sets and verify writes where needed.

## Performance and profiling

The current deterministic profile covers request counts, wall/CPU time, time to first row,
high-water counters, retained resources and optional Memray allocations:

```console
uv run python tools/b24api_evidence/profile_runtime.py --capability-suite
uv run python tools/b24api_evidence/profile_runtime.py --samples 7 --warmups 2
uv run --with memray python tools/b24api_evidence/profile_runtime.py \
  --case dense-10k --plan counted_batch --samples 7 --warmups 2 \
  --memray-output /tmp/b24api.bin
uv run --with memray memray stats /tmp/b24api.bin
```

These deterministic fixtures characterize local resources and network shape; they are not live
portal latency admission. See [docs/performance.md](docs/performance.md) for current measurements
and [docs/architecture.md](docs/architecture.md) for guarantees and ownership boundaries.

Projects moving from an earlier API surface can use [docs/migration.md](docs/migration.md).

## Verification

```console
uv sync --frozen
.venv/bin/pytest -q -p no:cacheprovider
.venv/bin/ruff check . --no-fix --no-cache
.venv/bin/ruff format --check . --no-cache
.venv/bin/mypy --strict b24api tools/b24api_evidence
git diff --check
```

The wheel regression installs into an isolated environment, executes the `b24api` entry point and
checks that tests, live/evidence tooling and credentials are excluded.
