pyodata Python OData Client

When consuming services with invalid metadata, you can use the Config class to define how the parser handles errors. This is passed to the pyodata.Client via the config argument.

Parser Policies:

• PolicyFatal: Raises an exception and terminates parsing (default).
• PolicyWarning: Reports the problem, executes fallback code, and continues.
• PolicyIgnore: Executes fallback code without reporting the problem and continues.

You can specify policies for individual error types using the custom_error_policies dictionary and keys from the ParserError enum.

import pyodata
from pyodata.v2.model import PolicyFatal, PolicyWarning, PolicyIgnore, ParserError, Config
import requests

SERVICE_URL = 'http://services.odata.org/V2/Northwind/Northwind.svc/'

# Define custom namespaces and error handling
namespaces = {
    'edmx': 'customEdmxUrl.com',
    'edm': 'customEdmUrl.com'
}

custom_config = Config(
    xml_namespaces=namespaces,
    default_error_policy=PolicyFatal(),
    custom_error_policies={
         ParserError.ANNOTATION: PolicyWarning(),
         ParserError.ASSOCIATION: PolicyIgnore()
    })

northwind = pyodata.Client(SERVICE_URL, requests.Session(), config=custom_config)

# Check if the parser encountered any errors
print(northwind.schema.is_valid)

For services requiring certificate-based authentication, you must provide the certificate and key to the requests.Session object.

1. Split your .p12 certificate into a public key (client.pem), a private key (key_decrypt.pem), and a CA certificate (ca.pem) using OpenSSL.
2. Set session.verify to the CA certificate path.
3. Set session.cert to a tuple of (client_cert_path, private_key_path).

# Extracting components from a .p12 file
openssl pkcs12 -in mycert.p12 -out ca.pem -cacerts -nokeys
openssl pkcs12 -in mycert.p12 -out client.pem -clcerts -nokeys
openssl pkcs12 -in mycert.p12 -out key.pem -nocerts
openssl rsa -in key.pem -out key_decrypt.pem

import pyodata
import requests

SERVICE_URL = 'https://odata.example.com/Secret.svc'

session = requests.Session()
session.verify = 'ca.pem'
session.cert = ('client.pem', 'key_decrypt.pem')

client = pyodata.Client(SERVICE_URL, session)

Some SAP OData services return a 200 OK status but signal domain errors through the sap-message response header. Because standard error handlers do not inspect this header, you should use the sap_header_error_hook provided by pyodata.vendor.SAP.

This hook reads the sap-message header, parses it as JSON, and raises a pyodata.vendor.SAP.BusinessGatewayError if the severity field in the JSON is set to "error".

Usage:

Pass sap_header_error_hook as the response_hook argument when initializing your pyodata.Client or pyodata.Service. The hook is stateless and safe for concurrent or async use.

Error Handling:

Catch pyodata.vendor.SAP.BusinessGatewayError to handle these specific domain errors.

import pyodata
from pyodata.vendor.SAP import sap_header_error_hook
import requests

SERVICE_URL = 'https://example.com/sap/opu/odata/sap/ZMyService'

session = requests.Session()
client = pyodata.Client(SERVICE_URL, session, response_hook=sap_header_error_hook)

try:
    result = client.entity_sets.Employees.get_entity(1).execute()
except pyodata.vendor.SAP.BusinessGatewayError as ex:
    print(f"SAP domain error: {ex}")

Pyodata supports three ways to apply filters to an entity set request:

1. String-based filtering: Pass a raw OData filter string to .filter().
2. Pythonic filtering: Use GetEntitySetFilter (aliased as esf) to build filters using logical operators like esf.and_() and comparison operators.
3. ORM-style filtering: Pass keyword arguments directly to .filter(). For complex patterns, use double-underscore syntax (e.g., __contains, __startswith).

from pyodata.v2.service import GetEntitySetFilter as esf

# 1. String-based
request = northwind.entity_sets.Employees.get_entities().filter("FirstName eq 'John' and LastName eq 'Smith'")

# 2. Pythonic style
request = northwind.entity_sets.Employees.get_entities().filter(esf.and_(
    request.FirstName == 'John',
    request.LastName == 'Smith'
))

# 3. ORM style (simple)
request = northwind.entity_sets.Employees.get_entities().filter(FirstName="John", LastName="Smith")

# 3. ORM style (complex patterns)
request = northwind.entity_sets.Employees.get_entities().filter(FirstName__contains="oh", LastName__startswith="Smi")

for entity in request.execute():
    print(entity.EmployeeID)