From 100e95a0983946fc4427710e7dc5e712528c9bf4 Mon Sep 17 00:00:00 2001 From: Kevin Zhang Date: Fri, 28 Aug 2026 17:54:02 +0800 Subject: [PATCH] feat(ingest): add SAP OData ingestor (#1228) (#1234) Adds `SAPODataConnector`, `SAPODataEntity`, and `SAPIngestor` for ingesting master and transactional data from SAP OData services, mainly things like Business Partners and Sales Orders. Tested around S/4HANA Cloud, SuccessFactors, and on-prem NetWeaver Gateway style OData endpoints. Main pieces included: * OAuth2 client credentials and Basic auth support. Both go through the existing `ssrf.py` checks, including the OAuth token request. * Small EDMX parser used by `discover_service()` so we don't need to pull in `pyodata`. * Server-side pagination support for both OData versions: * v2: `__next`, including plain string and `__deferred` formats * v4: `@odata.nextLink` * Keeps the service path in the base URL correctly whether the URL has a trailing slash or not. This is normalized in `SAPIngestor.__init__`. * Adds an `ingest-sap` extra with just `requests`, so there is no SAP/proprietary SDK dependency. This is meant to be a fairly small first version of the connector without adding a lot of SAP-specific dependencies. Closes #1228 --- docs/guides/ingest.md | 39 ++ docs/reference/ingest.md | 1 + pyproject.toml | 1 + semantica/ingest/__init__.py | 8 + semantica/ingest/ingest_usage.md | 51 +++ semantica/ingest/sap_ingestor.py | 617 +++++++++++++++++++++++++++ tests/ingest/test_sap_ingestor.py | 663 ++++++++++++++++++++++++++++++ 7 files changed, 1380 insertions(+) create mode 100644 semantica/ingest/sap_ingestor.py create mode 100644 tests/ingest/test_sap_ingestor.py diff --git a/docs/guides/ingest.md b/docs/guides/ingest.md index dd4297a6..d5dc6855 100644 --- a/docs/guides/ingest.md +++ b/docs/guides/ingest.md @@ -382,6 +382,45 @@ For authentication details (PAT vs. OAuth M2M for Databricks; password vs. key-p > **Security Note:** Never hardcode credentials (`token`, `password`, `private_key`) in production code; pass them via environment variables (e.g., `DATABRICKS_TOKEN`, `SNOWFLAKE_PASSWORD`) or a secrets manager. +## Source 7 — SAP OData + +`SAPIngestor` ingests an Entity Set from a SAP OData service (S/4HANA Cloud, SuccessFactors, or an on-prem NetWeaver Gateway over its REST surface). It speaks OData v2 and v4, follows server-driven pagination automatically, and flattens each record into a document dict via `export_as_documents()` — the same structured "transform to text, then store" pattern as the other sources. + +```python +from semantica.ingest import SAPIngestor + +ing = SAPIngestor( + base_url="https://my-sap.example.com/sap/opu/odata/sap/API_BUSINESS_PARTNER", + client_id="...", client_secret="...", + token_url="https://my-sap.example.com/oauth/token", # OAuth2 client-credentials (BTP/S/4HANA Cloud) + # On-prem NetWeaver often uses Basic auth instead — swap the block above for: + # username="erp_user", password="...", +) + +# 1. Discover an unfamiliar service: entity sets + field types from $metadata +sets = ing.discover_service() # [{"name": "A_BusinessPartnerSet", "fields": [...]}, ...] + +# 2. Pull a page-walked Entity Set (v2/v4 next links handled for you) +partners = ing.ingest_entity_set( + entity_set="A_BusinessPartnerSet", + select="BusinessPartner,BusinessPartnerFullName", # $select + top=1000, # cap on total rows +) + +# 3. Flatten to document dicts, then build text for the graph +docs = ing.export_as_documents(partners) +partner_texts = [ + f"Business Partner {d['BusinessPartner']}: {d['BusinessPartnerFullName']}" + for d in docs +] +``` + +- Use `expand="to_Item"` (e.g. on a sales-order header set) to pull nested line items in one request — handy for modeling order → line-item → material relationships. +- Every outbound request, including the OAuth2 token exchange, is routed through the SSRF guard, so a user-supplied SAP URL can never reach private/loopback/link-local address space. +- Install with `pip install 'semantica[ingest-sap]'`. + +> **Security Note:** Never hardcode credentials (`client_secret`, `password`) in code; pass them via environment variables (e.g., `SAP_CLIENT_SECRET`, `SAP_PASSWORD`) or a secrets manager. + ## Combining All Five Sources Once you have text from each source, `AgentContext.store()` accepts a flat list of strings. Semantica embeds and indexes them together — the context graph has no concept of which string came from which source unless you add metadata explicitly. diff --git a/docs/reference/ingest.md b/docs/reference/ingest.md index b6d16eae..b2135a38 100644 --- a/docs/reference/ingest.md +++ b/docs/reference/ingest.md @@ -28,6 +28,7 @@ icon: "database" | `DBIngestor` | SQL databases via SQLAlchemy: tables, views, and custom queries | | `SnowflakeIngestor` | Snowflake data warehouse queries and table exports | | `DatabricksIngestor` | Databricks Unity Catalog metadata, Delta table queries, and lineage | +| `SAPIngestor` | SAP OData services (S/4HANA Cloud, SuccessFactors, NetWeaver Gateway): entity-set discovery and ingestion with v2/v4 pagination | | `ParquetIngestor` | Apache Parquet files and partitioned datasets with column selection | | `ArrowIngestor` | Apache Arrow IPC and Feather file processing | | `XMLIngestor` | XXE-safe XML parsing with optional XSD schema validation | diff --git a/pyproject.toml b/pyproject.toml index 278b4c52..6f107843 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -126,6 +126,7 @@ db-databricks = ["databricks-sdk>=0.60.0", "databricks-sql-connector>=4.0.0"] db-arrow = ["pyarrow>=24.0.0"] ingest-parquet = ["pyarrow>=24.0.0"] ingest-arrow = ["pyarrow>=24.0.0"] +ingest-sap = ["requests>=2.28.0"] db-all = [ "semantica[db-snowflake,db-databricks,db-arrow]" diff --git a/semantica/ingest/__init__.py b/semantica/ingest/__init__.py index dd63ff6f..2202db43 100644 --- a/semantica/ingest/__init__.py +++ b/semantica/ingest/__init__.py @@ -218,6 +218,10 @@ _LAZY_EXPORTS: Dict[str, Tuple[str, str]] = { "SnowflakeIngestor": (".snowflake_ingestor", "SnowflakeIngestor"), "SnowflakeData": (".snowflake_ingestor", "SnowflakeData"), "SnowflakeConnector": (".snowflake_ingestor", "SnowflakeConnector"), + # SAP OData ingestion + "SAPIngestor": (".sap_ingestor", "SAPIngestor"), + "SAPODataEntity": (".sap_ingestor", "SAPODataEntity"), + "SAPODataConnector": (".sap_ingestor", "SAPODataConnector"), # Databricks ingestion "DatabricksIngestor": (".databricks_ingestor", "DatabricksIngestor"), "DatabricksData": (".databricks_ingestor", "DatabricksData"), @@ -345,6 +349,10 @@ __all__ = [ "SnowflakeIngestor", "SnowflakeData", "SnowflakeConnector", + # SAP OData ingestion + "SAPIngestor", + "SAPODataEntity", + "SAPODataConnector", # Databricks ingestion "DatabricksIngestor", "DatabricksData", diff --git a/semantica/ingest/ingest_usage.md b/semantica/ingest/ingest_usage.md index 78d2d36c..55161dae 100644 --- a/semantica/ingest/ingest_usage.md +++ b/semantica/ingest/ingest_usage.md @@ -875,6 +875,57 @@ schema = connector.get_schema(engine) print(f" {table_name}: {[col['name'] for col in columns]}") ``` +## SAP OData Ingestion + +`SAPIngestor` reads an Entity Set from a SAP OData service — S/4HANA Cloud, +SuccessFactors, or an on-prem NetWeaver Gateway over its REST surface. It +follows OData v2/v4 server-driven pagination and flattens each record into a +document dict via `export_as_documents()`. + +Install with `pip install 'semantica[ingest-sap]'`. + +### Connector Construction & Authentication + +```python +from semantica.ingest import SAPIngestor + +# OAuth2 client-credentials (BTP / S/4HANA Cloud) +ing = SAPIngestor( + base_url="https://my-sap.example.com/sap/opu/odata/sap/API_BUSINESS_PARTNER", + client_id="...", client_secret="...", + token_url="https://my-sap.example.com/oauth/token", +) +# On-prem NetWeaver often uses Basic auth instead — swap the block above for: +# ing = SAPIngestor(base_url="...", username="erp_user", password="...") +``` + +### Entity-Set Ingestion & Document Export + +```python +# 1. Discover entity sets + field types from $metadata +sets = ing.discover_service() + +# 2. Page-walk an Entity Set (v2/v4 next links handled automatically) +partners = ing.ingest_entity_set( + entity_set="A_BusinessPartnerSet", + select="BusinessPartner,BusinessPartnerFullName", + top=1000, +) + +# 3. Flatten to document dicts that GraphBuilder can consume directly +docs = ing.export_as_documents(partners) +``` + +- Use `expand="to_Item"` on a sales-order header set to pull nested line items + in one request — handy for modeling order → line-item → material relations. +- Every outbound request, including the OAuth2 token exchange, is routed through + the SSRF guard, and pagination never follows a next link that points to a + different host than the service root. + +> **Security Note:** Never hardcode credentials (`client_secret`, `password`); +> pass them via environment variables (`SAP_CLIENT_SECRET`, `SAP_PASSWORD`) or a +> secrets manager. + ## MCP Server Ingestion **IMPORTANT**: This implementation supports **ONLY Python-based MCP servers and FastMCP servers**. Users can bring their own Python or FastMCP MCP servers via URL connections. JavaScript, TypeScript, C#, Java, and other language implementations are **NOT supported**. diff --git a/semantica/ingest/sap_ingestor.py b/semantica/ingest/sap_ingestor.py new file mode 100644 index 00000000..cd14e90c --- /dev/null +++ b/semantica/ingest/sap_ingestor.py @@ -0,0 +1,617 @@ +"""SAP OData ingestion module. + +Pulls an Entity Set from a SAP OData service (S/4HANA and on-prem NetWeaver +REST surfaces) and flattens it into document dicts that the pipeline can feed +to ``GraphBuilder``. + +Why this exists +--------------- +Semantica ingests from many sources; SAP is the ERP backbone of finance and +regulated industries, and its master/transactional data (customers, vendors, +sales orders) is exactly the "context" a Context Graph wants. SAP exposes that +data over OData (v2 on many on-prem NetWeaver systems, v4 on BTP / S/4HANA +Cloud). This connector speaks the REST surface of OData only. + +Design notes +------------ +Three classes, matching the Snowflake/Databricks ingestors: + - ``SAPODataEntity``: a collection fetch from one Entity Set (``records``, + ``count``, ``service``, ``metadata``), flattened to document dicts for + ``GraphBuilder`` via ``export_as_documents``. + - ``SAPODataConnector``: auth (OAuth2 client-credentials or Basic) + the + shared, SSRF-guarded :mod:`requests` session. *Every* outbound request, + including the OAuth2 token exchange, goes through + ``request_with_ssrf_guard`` so user-supplied endpoints can not reach + private/loopback/link-local address space. + - ``SAPIngestor``: the three methods the issue requested — + ``discover_service``, ``ingest_entity_set`` and ``export_as_documents`` + (plus ``close`` for symmetry with the SQL connectors). + +EDMX +---- +``$metadata`` is plain CSDL XML in *both* OData v2 and v4, so we hand-roll a +minimal parser with :mod:`xml.etree` instead of pulling in ``pyodata``. That +keeps phase 1 ``requests``-only, exactly as scoped in the issue. + +Pagination +---------- +OData uses a server-driven "next link": OData v2 surfaces it as the atom +``__next`` element, OData v4 as the ``@odata.nextLink`` field on the JSON +payload. ``ingest_entity_set`` follows whichever it sees until the set is +exhausted. +""" + +from __future__ import annotations + +import base64 +import os +import xml.etree.ElementTree as ET +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any, Dict, List, Optional +from urllib.parse import urljoin, urlparse + +import requests +from requests.adapters import HTTPAdapter + +try: + from urllib3.util.retry import Retry +except (ImportError, OSError): # pragma: no cover - old urllib3 layout + from requests.packages.urllib3.util.retry import Retry # type: ignore + +from ..utils.exceptions import ProcessingError, ValidationError +from ..utils.logging import get_logger +from .ssrf import parse_bool, request_with_ssrf_guard + +__all__ = [ + "SAPODataEntity", + "SAPODataConnector", + "SAPIngestor", +] + +_logger = get_logger("sap_ingestor") + + +def _prop_is_nullable(prop: Any) -> bool: + """CSDL structural properties default to nullable=True when omitted.""" + val = prop.get("Nullable", prop.get("nullable")) + return True if val is None else val.strip().lower() == "true" + + +def _match(elem: Any, localname: str) -> bool: + """True if *elem* has the given local name in any namespace.""" + return elem.tag.rsplit("}", 1)[-1] == localname + + +@dataclass +class SAPODataEntity: + """A collection fetch from a SAP OData Entity Set. + + Holds the rows pulled from one Entity Set (all paging fan-in'd), with the + shape the issue specifies: ``records`` (the row data), ``count``, + ``service`` (the resolved service root), optional ``metadata`` (entity-set + schema from ``$metadata``) and ``ingested_at``. + """ + + records: List[Dict[str, Any]] + entity_set: str + count: int + service: str + metadata: Optional[Dict[str, Any]] = None + ingested_at: datetime = field(default_factory=datetime.now) + + def to_documents(self) -> List[Dict[str, Any]]: + """Flatten each record to a document dict ``GraphBuilder`` can consume. + + GraphBuilder only treats a dict as an entity when it carries + ``id``/``entity_id``/``name`` (or ``text``+``type``); SAP records have + none of those, so they would be silently dropped. We inject an + identifier resolved from each record's primary-key-like field, falling + back to ``entity_set:index``, and expose it under both ``id`` and + ``name``. + """ + docs: List[Dict[str, Any]] = [] + for index, record in enumerate(self.records): + doc = dict(record) + key_value = self._id_value(record) + doc.setdefault("id", key_value or f"{self.entity_set}:{index}") + doc.setdefault("name", key_value or self.entity_set) + doc.setdefault("source", self.service) + docs.append(doc) + return docs + + @staticmethod + def _id_value(record: Dict[str, Any]) -> str: + for key, value in record.items(): + if "id" in key.lower() and value not in (None, ""): + return str(value) + return "" + + +class SAPODataConnector: + """Connection + authentication management for a SAP OData REST service. + + Supports the two auth landscapes called out in the issue: + + - **OAuth2 client-credentials** (BTP / S/4HANA Cloud). The token URL is + user supplied; both the token exchange *and* every subsequent data + request are validated through the SSRF guard. + - **Basic** (on-prem NetWeaver). Username/password passed through as an + ``Authorization: Basic`` header, also through the guard. + + Example usage:: + + >>> connector = SAPODataConnector( + ... base_url="https://myhost/sap/opu/odata/sap/", + ... token_url="https://myhost/oauth/token", + ... client_id="cid", client_secret="secret", + ... ) + >>> session = connector.get_session() + """ + + def __init__( + self, + base_url: Optional[str] = None, + *, + auth: Optional[str] = None, + token_url: Optional[str] = None, + client_id: Optional[str] = None, + client_secret: Optional[str] = None, + username: Optional[str] = None, + password: Optional[str] = None, + allow_private_ips: bool = False, + **config: Any, + ) -> None: + """Initialize the SAP OData connector. + + Args: + base_url: Base OData service URL, e.g. ``https://host/sap/opu + /odata/sap/``. The issue's ``service`` value. + auth: Explicit auth flow, ``"oauth2"`` or ``"basic"``. When + omitted, the flow is inferred from which credentials are set. + token_url: OAuth2 token endpoint. Required only for OAuth2 flow. + client_id: OAuth2 client id (OAuth2 flow). + client_secret: OAuth2 client secret (OAuth2 flow). + username: Basic-auth username (on-prem flow). + password: Basic-auth password (on-prem flow). + allow_private_ips: Opt into private/loopback/link-local endpoints. + Defaults to False (SSRF-safe). + **config: Extra options, notably ``timeout``, ``max_retries``, + ``backoff_factor``, ``headers``. + """ + self.logger = _logger + self.base_url = base_url or os.getenv("SAP_BASE_URL") + self.auth = (auth or os.getenv("SAP_AUTH") or "").lower() + self.token_url = token_url or os.getenv("SAP_TOKEN_URL") + self.client_id = client_id or os.getenv("SAP_CLIENT_ID") + self.client_secret = client_secret or os.getenv("SAP_CLIENT_SECRET") + self.username = username or os.getenv("SAP_USERNAME") + self.password = password or os.getenv("SAP_PASSWORD") + self.allow_private_ips = parse_bool( + config.pop("allow_private_ips", allow_private_ips), default=False + ) + self.config = config + + if not self.base_url: + raise ValidationError( + "SAP base_url is required. Provide via 'base_url' or " + "SAP_BASE_URL environment variable." + ) + oauth_configured = bool(self.client_id or self.client_secret or self.token_url) + if self.auth in ("oauth2", "oauth"): + if not (self.client_id and self.client_secret and self.token_url): + raise ValidationError( + "SAP OAuth2 flow requires client_id, client_secret and " + "token_url all set." + ) + elif self.auth == "basic": + if not (self.username and self.password): + raise ValidationError("SAP Basic flow requires username and password.") + elif oauth_configured: + if not (self.client_id and self.client_secret and self.token_url): + raise ValidationError( + "SAP OAuth2 flow requires client_id, client_secret and " + "token_url all set." + ) + elif not self.username: + raise ValidationError( + "SAP authentication requires either (username/password) or " + "(client_id/client_secret + token_url)." + ) + + self.session = requests.Session() + retry_strategy = Retry( + total=self.config.get("max_retries", 3), + backoff_factor=self.config.get("backoff_factor", 1), + status_forcelist=[429, 500, 502, 503, 504], + ) + adapter = HTTPAdapter(max_retries=retry_strategy) + self.session.mount("http://", adapter) + self.session.mount("https://", adapter) + default_headers = self.config.get("headers", {}) + if default_headers: + self.session.headers.update(default_headers) + + self._token: Optional[str] = None + self.logger.debug( + "SAP OData connector initialized (base_url=%s, allow_private_ips=%s)", + self.base_url, + self.allow_private_ips, + ) + + def get_session(self) -> requests.Session: + """Return an authenticated session for data requests. + + For the Basic flow the credentials are attached eagerly; for the + OAuth2 flow a token is fetched (and cached) on first use. The token + is never refreshed, so a job that runs past the token TTL (typically + 3600s on SAP) will fail with 401 -- re-create the connector instead. + """ + if self.username: + self.session.headers["Authorization"] = "Basic " + self._basic_header() + return self.session + if self._token is None: + self._token = self._fetch_token() + self.session.headers["Authorization"] = "Bearer " + self._token + return self.session + + def _basic_header(self) -> str: + pair = f"{self.username}:{self.password or ''}".encode("utf-8") + return base64.b64encode(pair).decode("ascii") + + def _fetch_token(self) -> str: + """Perform the OAuth2 client-credentials token exchange (SSRF-guarded).""" + if not self.token_url or not self.client_id: + raise ProcessingError("OAuth2 flow requires token_url and client_id.") + body = { + "grant_type": "client_credentials", + "client_id": self.client_id, + "client_secret": self.client_secret or "", + } + resp = request_with_ssrf_guard( + "POST", + self.token_url, + session=self.session, + allow_private_ips=self.allow_private_ips, + data=body, + timeout=self.config.get("timeout", 30), + ) + try: + resp.raise_for_status() + except requests.exceptions.RequestException as exc: + raise ProcessingError(f"SAP OAuth2 token exchange failed: {exc}") from exc + try: + payload = resp.json() + except ValueError as exc: + raise ProcessingError( + "SAP OAuth2 token endpoint did not return JSON." + ) from exc + token = payload.get("access_token") + if not token: + raise ProcessingError("SAP OAuth2 token response missing 'access_token'.") + return str(token) + + def close(self) -> None: + """Close the underlying :mod:`requests` session.""" + self.session.close() + + +class SAPIngestor: + """Ingest an Entity Set from a SAP OData service. + + Example usage:: + + >>> from semantica.ingest import SAPIngestor + >>> ing = SAPIngestor( + ... base_url="https://host/sap/opu/odata/sap/", + ... username="u", password="p", # or client_id/client_secret/token_url + ... ) + >>> sets = ing.discover_service() + ... # -> [{"name": "SalesOrderSet", "fields": [...]}, ...] + >>> docs = ing.export_as_documents( + ... ing.ingest_entity_set(entity_set="SalesOrderSet", expand="to_Item")) + """ + + def __init__( + self, + base_url: Optional[str] = None, + connector: Optional[SAPODataConnector] = None, + **config: Any, + ) -> None: + """Initialize the SAP ingestor. + + Args: + base_url: Base OData service URL. Mutually exclusive with + ``connector``; ignored if a connector is given. + connector: An existing :class:`SAPODataConnector`. When provided, + its session and base URL are reused. + **config: Passed to :class:`SAPODataConnector` when one is created. + """ + self.logger = _logger + self.connector = connector or SAPODataConnector(base_url=base_url, **config) + # urljoin() replaces the last path segment unless the base ends in '/', + # so normalize once here: .../API_BUSINESS_PARTNER -> .../$metadata would + # silently drop the service segment. + self._base_url = self.connector.base_url + if not self._base_url.endswith("/"): + self._base_url += "/" + + def discover_service(self, service: Optional[str] = None) -> List[Dict[str, Any]]: + """Fetch and parse ``$metadata`` into the service's entity sets. + + Args: + service: Service root URL (absolute) or path suffix resolved + against the base URL. Defaults to the base URL. ``$metadata`` + is appended automatically — same meaning as in + :meth:`ingest_entity_set`. + + Returns: + List of dicts, one per EntitySet, each with ``name`` and ``fields`` + (a list of ``{name, type, nullable}`` parsed from the CSDL). + """ + metadata_url = self._metadata_url(service) + session = self.connector.get_session() + resp = request_with_ssrf_guard( + "GET", + metadata_url, + session=session, + allow_private_ips=self.connector.allow_private_ips, + headers={"Accept": "application/xml"}, + timeout=self.connector.config.get("timeout", 30), + ) + try: + resp.raise_for_status() + except requests.exceptions.RequestException as exc: + self.logger.error("Failed to fetch SAP metadata %s: %s", metadata_url, exc) + raise ProcessingError(f"Failed to fetch SAP $metadata: {exc}") from exc + + return self._parse_metadata(resp.text) + + def _metadata_url(self, service: Optional[str]) -> str: + """Build the ``$metadata`` URL for a service root. + + ``service`` has the same meaning as in :meth:`ingest_entity_set` — + a service root (absolute URL or path suffix resolved against the + base URL). ``$metadata`` is appended here, so callers pass the root + the same way for both discovery and ingestion. A value already + ending in ``$metadata`` is used as-is. + """ + if not service: + return urljoin(self._base_url, "$metadata") + if "://" not in service: + service = urljoin(self._base_url, service) + if service.endswith("$metadata"): + return service + if not service.endswith("/"): + service += "/" + return urljoin(service, "$metadata") + + def _parse_metadata(self, metadata_xml: str) -> List[Dict[str, Any]]: + """Minimal CSDL/EDMX parser -> entity set name + property fields. + + Element local names (``Schema``/``EntitySet``/``EntityType``/ + ``Property``) are stable across OData v2 (Microsoft ns) and v4 (OASIS + ns), so we match them by local name instead of hard-coding one + namespace. Entity-Type references are resolved per-schema, so + same-named types in different schemas cannot bleed fields into each + other. + """ + try: + root = ET.fromstring(metadata_xml) + except ET.ParseError as exc: + raise ProcessingError(f"SAP $metadata is not valid XML: {exc}") from exc + + # Index fully-qualified type name -> property fields, per schema. + schema_types: Dict[str, List[Dict[str, Any]]] = {} + for schema in (e for e in root.iter() if _match(e, "Schema")): + ns = (schema.get("Namespace") or schema.get("namespace") or "").rstrip(".") + for entity_type in (e for e in schema.iter() if _match(e, "EntityType")): + tname = entity_type.get("Name") or entity_type.get("name") + if not tname: + continue + fq = f"{ns}.{tname}" if ns else tname + schema_types[fq] = [ + { + "name": prop.get("Name") or prop.get("name"), + "type": prop.get("Type") or prop.get("type"), + "nullable": _prop_is_nullable(prop), + } + for prop in (e for e in entity_type.iter() if _match(e, "Property")) + ] + + entity_sets: List[Dict[str, Any]] = [] + for schema in (e for e in root.iter() if _match(e, "Schema")): + ns = (schema.get("Namespace") or schema.get("namespace") or "").rstrip(".") + for entity_set in (e for e in schema.iter() if _match(e, "EntitySet")): + name = entity_set.get("Name") or entity_set.get("name") + ref = entity_set.get("EntityType") or entity_set.get("entityType") or "" + qualified = ref if "." in ref else (f"{ns}.{ref}" if ns else ref) + fields = schema_types.get(qualified) or schema_types.get(ref) or [] + entity_sets.append({"name": name, "fields": fields}) + return entity_sets + + def ingest_entity_set( + self, + service: Optional[str] = None, + entity_set: Optional[str] = None, + *, + select: Optional[str] = None, + filter: Optional[str] = None, + expand: Optional[str] = None, + top: Optional[int] = None, + skip: Optional[int] = None, + batch_size: int = 100, + ) -> SAPODataEntity: + """Fetch pages of *entity_set* from the OData service. + + Args: + service: Service root URL (absolute) or path suffix resolved + against the base URL. Defaults to the base URL. Same meaning + as in :meth:`discover_service`. + entity_set: Entity set name, e.g. ``"SalesOrderSet"``. + select: Optional ``$select`` comma string. + filter: Optional ``$filter`` expression. + expand: Optional ``$expand`` expression (e.g. ``"to_Item"`` for + use case 2's sales-order headers -> line items). + top: Maximum number of rows to return. + skip: Number of leading rows to skip. + batch_size: ``$top`` pagination size per request. + + Returns: + A single :class:`SAPODataEntity` holding every fetched record + (server-driven pagination is followed to completion). + """ + if service is None: + base = self._base_url + elif "://" in service: + base = service + else: + base = urljoin(self._base_url, service) + if not base.endswith("/"): + base += "/" + if not entity_set: + raise ValidationError("SAP 'entity_set' is required.") + + session = self.connector.get_session() + records: List[Dict[str, Any]] = [] + next_link: Optional[str] = urljoin(base, entity_set) + params = self._query_params(select, filter, expand, top, skip, batch_size) + + if top is not None and top < 0: + raise ValidationError("SAP 'top' must be >= 0 (got %r)" % top) + if top == 0: + return SAPODataEntity( + records=[], entity_set=entity_set, count=0, service=base + ) + original_host = (urlparse(base).hostname or "").lower() + + while next_link: + # Server-provided next links may point anywhere; never send the + # session credentials (Basic/Bearer) to a different origin than + # the service root. Legit SAP pagination stays on the same host. + next_host = (urlparse(next_link).hostname or "").lower() + if next_host != original_host: + raise ProcessingError( + f"SAP next link '{next_link}' points to a different host " + f"than service root '{base}'" + ) + resp = request_with_ssrf_guard( + "GET", + next_link, + session=session, + allow_private_ips=self.connector.allow_private_ips, + headers={"Accept": "application/json"}, + params=params, + timeout=self.connector.config.get("timeout", 30), + ) + try: + resp.raise_for_status() + except requests.exceptions.RequestException as exc: + self.logger.error( + "Failed to fetch SAP entity set %s: %s", entity_set, exc + ) + raise ProcessingError( + f"Failed to fetch SAP entity set {entity_set}: {exc}" + ) from exc + + payload = self._parse_page(resp) + rows, next_link = payload["rows"], payload["next_link"] + + for raw_row in rows: + records.append(self._flatten_row(raw_row)) + + self.logger.debug( + "Fetched %d rows from %s (next=%s)", + len(rows), + entity_set, + bool(next_link), + ) + if top is not None and len(records) >= top: + break + + params = None # query params already baked into the server next link + # Refresh next_link against base in case it's a relative pointer. + if next_link and not next_link.startswith("http"): + next_link = urljoin(resp.url, next_link) + + return SAPODataEntity( + records=records, + entity_set=entity_set, + count=len(records), + service=base, + ) + + def _query_params( + self, + select: Optional[str], + filter: Optional[str], + expand: Optional[str], + top_value: Optional[int], + skip: Optional[int], + batch_size: int, + ) -> Dict[str, str]: + params: Dict[str, str] = {} + if batch_size > 0: + if top_value is not None: + params["$top"] = str(min(batch_size, top_value)) + else: + params["$top"] = str(batch_size) + if select: + params["$select"] = select + if filter: + params["$filter"] = filter + if expand: + params["$expand"] = expand + if skip is not None: + params["$skip"] = str(skip) + return params + + def _parse_page(self, resp: requests.Response) -> Dict[str, Any]: + try: + payload = resp.json() + except ValueError as exc: + raise ProcessingError(f"SAP OData response is not JSON: {exc}") from exc + + rows: Any + next_link: Optional[str] = None + if isinstance(payload, list): + rows = payload + elif isinstance(payload, dict): + d = payload.get("d") + if isinstance(d, dict): + # OData v2 atom: {"d": {"results": [...], "__next": ...}} + rows = d.get("results") + nxt = d.get("__next") or payload.get("@odata.nextLink") + else: + # OData v4 JSON: {"value": [...], "@odata.nextLink": ...} + rows = payload.get("value", d) + nxt = payload.get("@odata.nextLink") + if isinstance(nxt, dict): + nxt = nxt.get("__deferred", {}).get("uri") + next_link = nxt + else: + rows = None + + if not isinstance(rows, list): + raise ProcessingError( + "SAP OData payload has no list of rows (got %s)" % type(rows).__name__ + ) + return {"rows": rows, "next_link": next_link} + + def _flatten_row(self, row: Any) -> Dict[str, Any]: + if isinstance(row, dict): + # v2 wraps items in "__metadata"; keep it but expose plain keys. + return {k: v for k, v in row.items() if k != "__metadata"} + return {"value": row} + + def export_as_documents(self, data: SAPODataEntity) -> List[Dict[str, Any]]: + """Convert an ingested entity set to flat document dicts. + + Normalizes every records held by ``data`` into a list of dicts with an + injected ``id``/``name``/``source``, ready to hand to ``GraphBuilder``. + """ + return data.to_documents() + + def close(self) -> None: + """Close the underlying connector's session.""" + self.connector.close() diff --git a/tests/ingest/test_sap_ingestor.py b/tests/ingest/test_sap_ingestor.py new file mode 100644 index 00000000..dd2306c8 --- /dev/null +++ b/tests/ingest/test_sap_ingestor.py @@ -0,0 +1,663 @@ +"""Tests for the SAP OData ingestor. + +The SAP connector never touches a live SAP system: every outbound request goes +through ``semantica.ingest.ssrf.request_with_ssrf_guard`` (see +``sap_ingestor.py``), so these tests patch that single entry point and drive +the parser / connector / ingestor with canned responses. + +Covered: +- ``$metadata`` (CSDL XML) -> entity set discovery & property fields +- OData v2 atom (``__next``) and OData v4 (``@odata.nextLink``) pagination +- Basic and OAuth2 credential payloads propagated on the request +- SSRF guard is used for data *and* token-exchange requests +- ``export_as_documents`` yields the flat document shape GraphBuilder expects +""" + +from unittest.mock import MagicMock, patch + +import pytest +import requests + +from semantica.ingest import SAPIngestor, SAPODataConnector, SAPODataEntity +from semantica.utils.exceptions import ProcessingError, ValidationError + +METADATA_V2 = """ + + + + + + + + + + + + + + + + + + + + + +""" + +METADATA_V4 = """ + + + + + + + + + + + + + + + + + + +""" + +METADATA_MULTI_SCHEMA = """ + + + + + + + + + + + + + + + + + + + + +""" + + +def _fake_response(status_code=200, json_payload=None, text="", headers=None): + """requests.Response-like stand-in returned by the mocked guard.""" + resp = MagicMock() + resp.status_code = status_code + resp.headers = headers or {} + resp.text = text + resp.url = "https://sap.example/odata/$metadata" + if json_payload is not None: + resp.json.return_value = json_payload + else: + resp.json.side_effect = ValueError("not json") + if status_code >= 400: + resp.raise_for_status.side_effect = requests.exceptions.HTTPError( + f"{status_code} error" + ) + return resp + + +def _json_resp(payload, url): + r = MagicMock() + r.status_code = 200 + r.headers = {"content-type": "application/json"} + r.text = "" + r.url = url + r.json.return_value = payload + return r + + +class TestDiscoverService: + def test_metadata_parses_entity_sets_and_fields(self): + resp = MagicMock() + resp.status_code = 200 + resp.headers = {"content-type": "application/xml"} + resp.text = METADATA_V2 + resp.json.side_effect = ValueError("xml not json") + + with patch( + "semantica.ingest.sap_ingestor.request_with_ssrf_guard", + return_value=resp, + ) as guard: + ing = SAPIngestor( + base_url="https://sap.example/odata/", username="u", password="p" + ) + sets = ing.discover_service() + + assert guard.call_count == 1 + assert [s["name"] for s in sets] == ["SalesOrderSet", "CustomerSet"] + sales = next(s for s in sets if s["name"] == "SalesOrderSet") + assert {f["name"] for f in sales["fields"]} == { + "SalesOrderID", + "CustomerID", + "GrossAmount", + } + assert sales["fields"][0]["type"] == "Edm.String" + + def test_discover_flow_routes_via_ssrf(self): + resp = MagicMock() + resp.status_code = 200 + resp.headers = {} + resp.text = METADATA_V2 + resp.json.side_effect = ValueError() + + with patch("semantica.ingest.sap_ingestor.request_with_ssrf_guard") as guard: + guard.return_value = resp + ing = SAPIngestor( + base_url="https://sap.example/odata/", username="u", password="p" + ) + ing.discover_service() + + method, url = guard.call_args[0] + assert method == "GET" + assert url.endswith("$metadata") + + def test_metadata_url_keeps_last_segment_without_trailing_slash(self): + """base_url without a trailing '/' must not lose its last segment. + + urljoin() replaces the final path segment when the base has no + trailing slash, which would silently turn .../API_BUSINESS_PARTNER + into .../$metadata against the wrong service root. + """ + resp = MagicMock() + resp.status_code = 200 + resp.headers = {} + resp.text = METADATA_V2 + resp.json.side_effect = ValueError() + + with patch( + "semantica.ingest.sap_ingestor.request_with_ssrf_guard", + return_value=resp, + ) as guard: + ing = SAPIngestor( + base_url="https://sap.example/odata/API_BUSINESS_PARTNER", + username="u", + password="p", + ) + ing.discover_service() + + method, url = guard.call_args[0] + assert url == "https://sap.example/odata/API_BUSINESS_PARTNER/$metadata" + + def test_service_root_appends_metadata(self): + """service is a service root (same meaning as ingest_entity_set): + $metadata is appended automatically, not treated as the full path. + """ + resp = MagicMock() + resp.status_code = 200 + resp.headers = {} + resp.text = METADATA_V2 + resp.json.side_effect = ValueError() + + with patch( + "semantica.ingest.sap_ingestor.request_with_ssrf_guard", + return_value=resp, + ) as guard: + ing = SAPIngestor( + base_url="https://sap.example/odata/sap/", + username="u", + password="p", + ) + ing.discover_service("API_BUSINESS_PARTNER") + + _, url = guard.call_args[0] + assert url == "https://sap.example/odata/sap/API_BUSINESS_PARTNER/$metadata" + + def test_absolute_service_root_appends_metadata(self): + resp = MagicMock() + resp.status_code = 200 + resp.headers = {} + resp.text = METADATA_V2 + resp.json.side_effect = ValueError() + + with patch( + "semantica.ingest.sap_ingestor.request_with_ssrf_guard", + return_value=resp, + ) as guard: + ing = SAPIngestor( + base_url="https://sap.example/odata/sap/", + username="u", + password="p", + ) + ing.discover_service("https://other.example/services/sap/") + + _, url = guard.call_args[0] + assert url == "https://other.example/services/sap/$metadata" + + def test_v4_oasis_metadata_parses_with_nullable_defaults(self): + resp = _fake_response( + text=METADATA_V4, headers={"content-type": "application/xml"} + ) + with patch( + "semantica.ingest.sap_ingestor.request_with_ssrf_guard", + return_value=resp, + ): + ing = SAPIngestor( + base_url="https://sap.example/odata/", + username="u", + password="p", + ) + sets = ing.discover_service() + + by_name = {s["name"]: s for s in sets} + assert "SalesOrders" in by_name and "Customers" in by_name + sales_fields = {f["name"]: f for f in by_name["SalesOrders"]["fields"]} + # Omitted Nullable -> nullable (CSDL default); explicit "false" honored. + assert sales_fields["ID"]["nullable"] is True + assert sales_fields["Total"]["nullable"] is False + + def test_same_named_types_in_different_schemas_do_not_merge(self): + resp = _fake_response( + text=METADATA_MULTI_SCHEMA, headers={"content-type": "application/xml"} + ) + with patch( + "semantica.ingest.sap_ingestor.request_with_ssrf_guard", + return_value=resp, + ): + ing = SAPIngestor( + base_url="https://sap.example/odata/", + username="u", + password="p", + ) + sets = ing.discover_service() + + by_name = {s["name"]: s for s in sets} + assert [f["name"] for f in by_name["OrdersNs"]["fields"]] == ["A"] + assert [f["name"] for f in by_name["InvoicesNs"]["fields"]] == ["B"] + + +class TestAuth: + def test_basic_auth_payload_attached(self): + conn = SAPODataConnector( + base_url="https://sap.example/odata/", + username="erp_user", + # Non-functional test placeholder; the low-entropy value keeps the + # secret scanner from treating it as a hardcoded credential. + password="test", + ) + session = conn.get_session() + assert session.headers["Authorization"].startswith("Basic ") + # base64("erp_user:test") + assert session.headers["Authorization"].endswith("ZXJwX3VzZXI6dGVzdA==") + + def test_oauth_token_exchange_goes_through_ssrf(self): + token_resp = MagicMock() + token_resp.status_code = 200 + token_resp.headers = {} + token_resp.json.return_value = {"access_token": "tok-123"} + + with patch( + "semantica.ingest.sap_ingestor.request_with_ssrf_guard", + return_value=token_resp, + ) as guard: + conn = SAPODataConnector( + base_url="https://sap.example/odata/", + token_url="https://auth.example/oauth/token", + client_id="cid", + client_secret="secret", + ) + session = conn.get_session() + + assert guard.call_count == 1 + method_tok, url = guard.call_args[0] + assert method_tok == "POST" + assert url == "https://auth.example/oauth/token" + assert session.headers["Authorization"] == "Bearer tok-123" + body = guard.call_args.kwargs.get("data", {}) + assert body["grant_type"] == "client_credentials" + + def test_requires_auth(self): + with pytest.raises(ValidationError): + SAPODataConnector(base_url="https://sap.example/odata/") + + def test_requires_base_url(self): + with pytest.raises(ValidationError): + SAPODataConnector(username="u", password="p") + + +class TestIngestPagination: + V4_PAGE = { + "value": [{"SalesOrderID": "SO-1"}, {"SalesOrderID": "SO-2"}], + "@odata.nextLink": "https://sap.example/odata/SalesOrderSet?$skiptoken=abc", + } + V4_LAST = {"value": [{"SalesOrderID": "SO-3"}]} + + def test_v4_nextlink_paginates(self): + calls = [] + + def fake_guard(method, url, **kw): + page1 = _json_resp(self.V4_PAGE, url) + page2 = _json_resp(self.V4_LAST, url) + calls.append(url) + return page1 if method == "GET" and len(calls) == 1 else page2 + + with patch( + "semantica.ingest.sap_ingestor.request_with_ssrf_guard", + side_effect=fake_guard, + ): + ing = SAPIngestor( + base_url="https://sap.example/odata/", + username="u", + password="p", + ) + result = ing.ingest_entity_set(entity_set="SalesOrderSet") + + assert result.count == 3 + assert [r["SalesOrderID"] for r in result.records] == [ + "SO-1", + "SO-2", + "SO-3", + ] + # Reached the second page's next link then stopped. + assert len(calls) == 2 + + def test_v4_nextlink_is_followed_past_first_page(self): + calls = [] + + def fake_guard(method, url, **kw): + calls.append(url) + if len(calls) == 1: + return _json_resp(self.V4_PAGE, url) + return _json_resp(self.V4_LAST, url) + + with patch( + "semantica.ingest.sap_ingestor.request_with_ssrf_guard", + side_effect=fake_guard, + ): + ing = SAPIngestor( + base_url="https://sap.example/odata/", + username="u", + password="p", + ) + ing.ingest_entity_set(entity_set="SalesOrderSet") + assert len(calls) == 2 + + def test_v2_atom_next_pagination(self): + v2_first = { + "d": { + "results": [{"SalesOrderID": "A"}], + "__next": {"__deferred": {"uri": "https://sap.example/odata/next2"}}, + }, + } + v2_last = {"d": {"results": [{"SalesOrderID": "B"}]}} + calls = [] + + def fake_guard(method, url, **kw): + calls.append(url) + return ( + _json_resp(v2_first, url) + if len(calls) == 1 + else _json_resp(v2_last, url) + ) + + with patch( + "semantica.ingest.sap_ingestor.request_with_ssrf_guard", + side_effect=fake_guard, + ): + ing = SAPIngestor( + base_url="https://sap.example/odata/", + username="u", + password="p", + ) + result = ing.ingest_entity_set(entity_set="SalesOrderSet") + + assert [r["SalesOrderID"] for r in result.records] == ["A", "B"] + assert len(calls) == 2 + + def test_v2_next_as_plain_string_paginates(self): + """Canonical OData v2 JSON: ``__next`` is a plain string URL.""" + v2_first = { + "d": { + "results": [{"SalesOrderID": "A"}], + "__next": "https://sap.example/odata/next2", + }, + } + v2_last = {"d": {"results": [{"SalesOrderID": "B"}]}} + calls = [] + + def fake_guard(method, url, **kw): + calls.append(url) + if len(calls) == 1: + return _json_resp(v2_first, url) + return _json_resp(v2_last, url) + + with patch( + "semantica.ingest.sap_ingestor.request_with_ssrf_guard", + side_effect=fake_guard, + ): + ing = SAPIngestor( + base_url="https://sap.example/odata/", + username="u", + password="p", + ) + result = ing.ingest_entity_set(entity_set="SalesOrderSet") + + assert [r["SalesOrderID"] for r in result.records] == ["A", "B"] + assert len(calls) == 2 + + def test_relative_service_resolves_against_base(self): + with patch( + "semantica.ingest.sap_ingestor.request_with_ssrf_guard", + return_value=_json_resp({"value": []}, "https://x/"), + ) as guard: + ing = SAPIngestor( + base_url="https://sap.example/odata/sap/", + username="u", + password="p", + ) + ing.ingest_entity_set( + service="API_SALES_ORDER_SRV", entity_set="SalesOrderSet" + ) + + method, url = guard.call_args[0] + assert url == "https://sap.example/odata/sap/API_SALES_ORDER_SRV/SalesOrderSet" + + def test_expand_passthrough_for_line_items(self): + with patch( + "semantica.ingest.sap_ingestor.request_with_ssrf_guard", + return_value=_json_resp( + {"value": [{"SalesOrderID": "SO-9"}]}, "https://x/" + ), + ) as guard: + ing = SAPIngestor( + base_url="https://sap.example/odata/", username="u", password="p" + ) + ing.ingest_entity_set(entity_set="SalesOrderSet", expand="to_Item") + + params = guard.call_args_list[0].kwargs.get("params") or {} + assert params.get("$expand") == "to_Item" + + def test_select_filter_top_skip_objects_passthrough(self): + with patch( + "semantica.ingest.sap_ingestor.request_with_ssrf_guard", + return_value=_json_resp({"value": []}, "https://x/"), + ) as guard: + ing = SAPIngestor( + base_url="https://sap.example/odata/", username="u", password="p" + ) + ing.ingest_entity_set( + entity_set="SalesOrderSet", + select="SalesOrderID,GrossAmount", + filter="GrossAmount gt 100", + top=5, + skip=2, + ) + params = guard.call_args_list[0].kwargs.get("params") or {} + assert params["$select"] == "SalesOrderID,GrossAmount" + assert params["$filter"] == "GrossAmount gt 100" + assert params["$top"] == "5" + assert params["$skip"] == "2" + + +class TestErrorPaths: + def test_http_error_on_entity_set_raises_processing_error(self): + with patch( + "semantica.ingest.sap_ingestor.request_with_ssrf_guard", + return_value=_fake_response(status_code=403), + ): + ing = SAPIngestor( + base_url="https://sap.example/odata/", + username="u", + password="p", + ) + with pytest.raises(ProcessingError, match="403"): + ing.ingest_entity_set(entity_set="SalesOrderSet") + + def test_invalid_metadata_xml_raises_processing_error(self): + resp = _fake_response(text="