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="