onec-odata

A modern, typed Python client for the 1C:Enterprise OData standard interface (/odata/standard.odata). Built for Python 3.13+ on top of httpx.

It exists because generic OData libraries assume OData v4 and choke on 1C, which speaks OData v3 with several of its own conventions. This library handles the 1C reality directly:

• OData v3 typed literals — guid'...', datetime'...' in URLs and keys (v4 dropped these, which is why other clients send the wrong thing).
• Correct $metadata parsing — a namespace-tolerant CSDL v3 parser that copes with platform-version namespace drift and Cyrillic identifiers.
• 1C property conventions — _Key references, _Type dispatch/composite fields, _Base64Data value storage, and the four-underscore ____Presentation suffix.
• 1C-only operations — document Post/Unpost, register virtual tables (SliceLast, Balance, BalanceAndTurnovers, …), allowedOnly, optimistic locking via DataVersion/If-Match, and the data-load mode header.
• A fluent filter DSL — every operator and function from the 1C docs, including cast/isof for composite types and any/all lambdas.

Installation

pip install onec-odata

Usage

Connect

from onec_odata import ODataClient, Query, F, Guid

client = ODataClient("http://host/base", auth=("user", "password"))

The odata/standard.odata path is appended automatically. Use the client as a context manager to close the underlying connection pool:

with ODataClient("http://host/base", auth=("user", "pass")) as client:
    ...

Read a collection

goods = client.catalog("Товары")          # -> Catalog_Товары

page = goods.list(
    Query()
    .filter(F("Цена") > 1000)
    .select("Ref_Key", "Code", "Description")
    .order_by("Description")
    .top(50)
    .with_count()                          # $inlinecount=allpages
)

print(page.total_count)                    # total across all pages
for item in page:
    print(item.ref_key, item["Description"])

Read one entity by key

item = goods.get(Guid("41aa6331-954f-11e3-814b-005056c00008"))

# Composite key (e.g. an information register):
import datetime as dt
rate = client.information_register("КурсыВалют").get({
    "Period": dt.datetime(2008, 2, 5),
    "Валюта_Key": Guid("9d5c4222-8c4c-11db-a9b0-00055d49b45e"),
})

Stream every match (transparent paging)

for item in goods.iterate(Query().filter(F("DeletionMark") == False), page_size=500):
    ...

Count

n = goods.count(Query().filter(F("Цена") > 500))

Filters

from onec_odata import F, and_, or_, cast, isof, startswith

# Comparisons and boolean composition (& and |):
q = Query().filter((F("Цена") > 1000) & (F("Цена") < 5000))

# String functions:
Query().filter(startswith("Производитель", "ООО"))

# Navigation through references:
Query().filter(F("Контрагент/ИНН") == "7700000000").order_by("Контрагент/ИНН")

# Composite (multi-type) attribute compared to a typed reference:
Query().filter(
    F("ДокументПрихода") == cast(Guid("0d4a79cb-9843-4147-bcd9-80ac3ca2b9c7"),
                                 "Document_ПриходнаяНакладная")
)

# Lambda over a tabular section: documents with any line priced over 10000
Query().filter(F("Товары").any(lambda d: d.nav("Цена") > 10000))

Create, update, delete

created = goods.create({
    "Description": "Шлепанцы",
    "Артикул": "SL56X",
    "Поставщик_Key": Guid("086715b0-f348-11db-a9c5-00055d49b45e"),
})

# PATCH — only the given properties change:
goods.update(created.ref_key, {"Description": "Новое имя"})

# PUT — full replace; references use the @odata.bind form:
goods.replace(created.ref_key, {
    "Description": "Шлепанцы",
    "Поставщик@odata.bind": "Catalog_Поставщики(guid'...')",
    ...
})

# Optimistic locking:
goods.update(item.ref_key, {...}, if_match=item.data_version)

goods.delete(created.ref_key)   # immediate delete, not a deletion mark

Documents

docs = client.document("РасходТовара")
docs.post_document(doc_key, operational=False)   # провести
docs.unpost_document(doc_key)                    # отмена проведения

Register virtual tables (functions)

reg = client.information_register("КурсыВалют")
slice_last = reg.call("SliceLast", {"Condition": "Валюта/ОсновнаяВалюта_Key eq guid'...'"})

acc = client.accumulation_register("ТоварныеЗапасы")
turnovers = acc.call("BalanceAndTurnovers", {
    "StartPeriod": dt.datetime(2014, 1, 1),
    "EndPeriod": dt.datetime(2014, 2, 1),
    "Condition": "Товар_Key eq guid'...'",
})

Metadata

meta = client.metadata()                     # parsed once, then cached
et = meta.entity_type_for_set("Catalog_Товары")
for prop in et.properties:
    print(prop.name, prop.type, "key" if prop.name in et.key else "")

Error handling

from onec_odata import EntityNotFoundError, ConcurrencyError, AccessDeniedError

try:
    goods.get(some_key)
except EntityNotFoundError as e:
    print(e.status_code, e.internal_code, e.message)

Every 1C internal error code (section 17.4.10 of the docs) is mapped onto a specific exception subclass where it makes sense, with the raw code available as error.internal_code.

Debugging — see the actual OData request

Pass debug=True to print every request (decoded, so Cyrillic and the $filter operators are readable) to stderr:

client = ODataClient("http://host/base", auth=("user", "pass"), debug=True)
goods.list(Query().filter(F("Posted") == True).select("Ref_Key", "Number").top(3))
# [onec-odata] GET http://host/base/odata/standard.odata/Document_Заём?$filter=Posted eq true&$select=Ref_Key,Number&$top=3 -> 200 (212 ms)

debug can also be a callback receiving a RequestDebug (method, decoded url, raw_url, redacted headers, body, status_code, elapsed_ms):

client.debug = lambda info: my_logger.info("%s %s", info.method, info.url)

Or route it through logging (requests are always logged at DEBUG):

import logging
logging.basicConfig(level=logging.DEBUG)
logging.getLogger("onec_odata").setLevel(logging.DEBUG)

Preview a URL without sending it, or inspect the last exchange:

print(goods.url(Query().filter(F("Цена") > 1000).select("Ref_Key").top(5)))
# http://host/base/odata/standard.odata/Catalog_Товары?$filter=Цена gt 1000&$select=Ref_Key&$top=5

client.last_request   # the raw httpx.Request that was sent
client.last_response  # the raw httpx.Response

Sensitive headers (Authorization, Cookie) are redacted in debug output.

Development

python -m venv .venv && source .venv/bin/activate
pip install -e ".[dev]"
pytest
ruff check src tests

License

MIT