b1sl-sdk

name b1sl-sdk description Modern, async-first Python SDK for SAP Business One Service Layer (b1sl-python). Use this skill to interact with SAP B1 entities (Items, Business Partners, Orders, Invoices, and 100+ more) using type-safe Pydantic v2 models, a fluent OData query builder with operator overloading, automatic session management, and structured observability. Covers installation, configuration, async client patterns, query building, UDF handling, and the metadata generation pipeline. SAP B1 Python SDK (b1sl) Overview b1sl-python is a metadata-driven , async-first SDK for SAP Business One Service Layer. It automates the entire model and resource layer by parsing SAP's OData metadata, delivering full type safety, IDE autocompletion, and production-grade session management. PyPI : b1sl-python Repository : operator-ita/b1sl-python Verified Baseline : Service Layer 1.27 (SAP 10.0 FP 2405) Protocol : OData V4 ( v2 endpoint) Minimum for ETags : Service Layer 1.21+ Companion : b1sl.api_gateway — SAP API Gateway client (Crystal Reports layouts → PDF, port 60000, /rs/v1/ ); separate service, separate session, opt-in import. See docs/20-api-gateway.md . Installation # Recommended (uv) uv add b1sl-python # pip pip install b1sl-python # Optional extras uv add "b1sl-python[django]" # Django integration uv add "b1sl-python[generator]" # Metadata generation pipeline Configuration The SDK uses a hierarchical, environment-agnostic configuration system. Required Environment Variables B1SL_BASE_URL=https://your-server:50000 B1SL_USERNAME=manager B1SL_PASSWORD=your_password B1SL_COMPANY_DB=SBODEMOUS B1SL_ENV=dev # dev | test | prod B1SL_DRY_RUN=0 # 1 to enable Dry Run mode Loading Config from b1sl.b1sl import B1Environment, B1Config # Automatic: reads B1SL_ENV and merges .env + configs/{env}.json env = B1Environment.load() config = env.config # Or directly from environment variables config = B1Config.from_env() Environments B1SL_ENV Log Format Use Case dev (default) Human-readable Local development test Human-readable CI / test isolation prod Structured JSON Production observability pipelines Dry Run intercepted B1SL_DRY_RUN=1 . Intercepts POST/PATCH/DELETE. Never store B1SL_PASSWORD in configs/*.json . Only non-sensitive test data IDs belong there. Temporary Dry Run (Context Manager) You can toggle Dry Run mode temporarily for a specific block of code using the dry_run() context manager available in both sync and async clients: # Globally False, but locally True (Task-Safe via ContextVar) with b1.dry_run(): await b1.items.create(new_item) # Intercepted # Globally True, but locally False (Force execution) with b1.dry_run(enabled= False ): await b1.items.update(item) # Sent to SAP # NOTE: Always use 'with' (sync CM), NOT 'async with', even in async code. ❗ Critical Guidelines: Flat Namespace & Enums Always use the flat public namespace for models and enums to ensure clean code and IDE support. Never import from _generated internal paths. # ✅ Best Practice: Flat namespace for data models from b1sl.b1sl import entities as en # ✅ Best Practice: Field referencing from b1sl.b1sl.fields import Item, Order # Static "Pythonic" fields (recommended) from b1sl.b1sl.resources.odata import F # Raw proxy — UDFs / dynamic names only # Use 'en' for model instantiation new_item = en.Item(item_code= "A100" , item_name= "New Item" ) The recommended client for all production use cases. Basic Usage import asyncio from b1sl.b1sl import AsyncB1Client, B1Config async def main (): config = B1Config.from_env() async with AsyncB1Client(config) as b1: item = await b1.items.get( "A0001" ) print ( f"[ {item.item_code} ] {item.item_name} " ) asyncio.run(main()) The async with block handles POST /Logout automatically — even on exceptions. Manual Lifecycle (Long-running services) client = AsyncB1Client(config) await client.connect() # Manual Login # ... use client ... await client.aclose() # Manual Logout Top 16 Canonical Aliases Category Aliases Master Data items , business_partners , users Sales quotations , orders , delivery_notes , invoices , incoming_payments Purchasing purchase_orders , purchase_delivery_notes , purchase_invoices , vendor_payments Operations production_orders , journal_entries , service_calls , activities Dynamic Access (Any Endpoint) from b1sl.b1sl.models._generated.entities.inventory import ItemWarehouseInfo whse_resource = b1.get_resource(ItemWarehouseInfo, "ItemWarehouseInfo" ) data = await whse_resource.get( "A0001" ) Custom Client Alias (Enterprise Pattern) from b1sl.b1sl import AsyncB1Client from b1sl.b1sl.resources.async_base import AsyncGenericResource from b1sl.b1sl.models._generated.entities.inventory import ItemWarehouseInfo class MyB1Client ( AsyncB1Client ): @property def warehouses ( self ) -> AsyncGenericResource[ItemWarehouseInfo]: return self .get_resource(ItemWarehouseInfo, "ItemWarehouseInfo" ) High Concurrency with asyncio.gather async with AsyncB1Client(config) as b1: codes = [ "A0001" , "A0002" , "A0003" ] items = await asyncio.gather(*[b1.items.get(c) for c in codes]) The SDK uses a shared httpx.AsyncClient and an asyncio.Lock to prevent session floods. Key Async Features 401 Auto-Retry : Expired sessions are transparently renewed and the original request is retried once. Session Hydration : Reuse an existing B1SESSION token across serverless functions or Temporal activities. Optimistic Concurrency (ETags) : Automated ETag handling with smart cache invalidation on 412 conflicts. CRUD Operations (Master Data & Transactions) The SDK provides a consistent set of methods for interacting with resources. Create (POST) Instantiate a model and pass it to the .create() method. from b1sl.b1sl import entities as en new_item = en.Item(item_code= "A0001" , item_name= "New Item" ) await b1.items.create(new_item) Read (GET) Fetch by ID or check for existence. # Fast existence check if await b1.items.exists( "A0001" ): pass # Count total records total = await b1.items.count() Optimistic Concurrency (ETags) The SDK manages ETags behind the scenes. Every model instance has a .etag property. item = await b1.items.get( "P001" ) print (item.etag) # Displays the server-side version token Update (PATCH) - The "Surgical Delta" Pattern Best Practice : Never resubmit a full object. Only send the fields you want to change. # Create a minimal object for the update delta = en.Item(item_name= "Updated Name" ) # This sends ONLY the name change to SAP await b1.items.update( "A0001" , delta) Delete (DELETE) await b1.items.delete( "A0001" ) Official Print Layouts to PDF (API Gateway) Not Service Layer: SAP's API Gateway renders Crystal Reports layouts to PDF — the same document the SAP client prints. Sync and async twins, same surface. from b1sl.api_gateway import APIGatewayConfig, AsyncAPIGatewayClient cfg = APIGatewayConfig.from_env() # B1SL_GATEWAY_BASE_URL + B1SL_* credentials async with AsyncAPIGatewayClient(cfg) as gw: reports = await gw.list_reports() # catalog RCRI00xx only params = await gw.get_report_parameters( "QUT20020" ) # layout definition pdf = await gw.export_document_pdf( "QUT20020" , doc_entry= 12345 ) # DocKey@ set explicitly Rules the client enforces (all learned from the live gateway, none in SAP's manual): Failure is signalled by body , not HTTP status: (---) → APIGatewayParameterError (retried once — also appears under concurrent exports), {} from LoadCR → APIGatewayLayoutNotFoundError , non- %PDF- → APIGatewayPDFError , bad login → 200 + {"code":-1} → APIGatewayAuthError . Empty nullable parameters are omitted; xsd:date needs explicit ISO values (ranges = two strings in one inner array); never trust the preloaded DocKey@ . Document-bound layout codes ( QUT200xx …) are not discoverable by API — read them in Print Layout Designer and version the mapping in your app. max_concurrent_exports (default 3) bounds parallel renders per client. Layout-specific required parameters (fiscal folios, ext params…) are never guessed : missing_required_parameters(params, values) lists what a layout still needs, and export_pdf(..., resolver=fn) lets the application supply them ( fn(param) -> value | None ). Transparent Pagination Streams When dealing with large datasets, SAP Service Layer automatically paginates results. The SDK provides a .stream() method to transparently handle these pages using Python generators. Usage Available on any resource or builder. from b1sl.b1sl.fields import Item # 1. Async iteration async for item in b1.items. filter (Item.quantity_on_stock > 0 ).stream(page_size= 100 ): process(item) # 2. Sync iteration — same constants, same semantics for item in b1.items. filter (Item.quantity_on_stock > 5 ).stream(): process(item) Configuration page_size : Controls B1S-PageSize header (HTTP efficiency). max_pages : Safety limit on number of HTTP requests. .top(N) : Hard global limit on total items yielded across all pages. Common Patterns Progress : total = await b1.items.count(); async for i in b1.items.stream(): ... Collect : items = [i async for i in b1.items.stream()] Safety : .stream(max_pages=5) Guarantee The SDK ensures that all query parameters ( $filter , $select , etc.) are re-applied to every subsequent page fetch, even if SAP omits them in the nextLink . OData $batch Operations (Performance & Atomicity) The SDK supports grouping multiple operations into a single HTTP request using a Proxy-based recording pattern. Use Case High Concurrency : Fetching hundreds of records using generic queries in one Go. Transaction Integrity : Ensuring multiple creates/updates succeed or fail together as a unit. Basic Pattern async with b1.batch() as batch: # Operations are enqueued via Recording Proxy await batch.items.top( 1 ).execute() # Atomic ChangeSet scope async with batch.changeset() as cs: await cs.items.create(en.Item(item_code= "B1001" )) await cs.orders.create(new_order) # Dispatch and parse results results = await batch.execute() Result Analysis Results are flattened and indexed according to their original enqueueing order. if results.all_ok: print ( f"Operation 0 found { len (results[ 0 ].entity)} items" ) print ( f"New Item Code: {results[ 2 ].entity.item_code} " ) else : for r in results.failed: print ( f"Op {r.index} failed: {r.error} " ) Error Handling & Atomicity Partial Success : Top-level operations are independent. If one fails, others still succeed. Atomic ChangeSets : If one operation inside a changeset() fails, the entire ChangeSet is rolled back. No Exceptions : batch.execute() returns results even on failure. Use results.all_ok or results.failed . Sync parity : B1Client.batch() works identically with plain with blocks and a sync execute() . Dry Run aware : under with b1.dry_run(): , batch.execute() returns synthesized per-op 204 s without sending anything to SAP. [!IMPORTANT] OData Rule : GET operations are not permitted inside a changeset() block. The SDK will raise a ValueError if this is attempted. Error Handling The SDK maps Service Layer HTTP errors to specialized Python exceptions for cleaner flow control: B1NotFoundError : Resource missing (404). B1ValidationError : Bad request or validation failure (400). SAPConcurrencyError : ETag version mismatch (412). B1AuthError : Authentication or session failure (401). B1Exception : Base class for all SDK-specific errors. Pattern: Safe Existence Check Instead of catching 404s manually, use the .exists() helper: if await b1.items.exists( "A0001" ): # item exists, proceed with logic pass Pattern: Defensive Error Parsing The SDK handles cases where SAP returns string-based error nodes instead of dictionaries, ensuring e.details is always safe to inspect if it contains valid JSON. FastAPI Integration from fastapi import FastAPI from contextlib import asynccontextmanager from b1sl.b1sl import AsyncB1Client, B1Config