How to ingest Oracle Fusion data into Foundry with BI Cloud Connector (BICC)

Show & Tell
,
1

We recently had to get a fairly large amount of data out of Oracle Fusion and into Foundry, and BI Cloud Connector (BICC) ended up being the cleanest way to do it. This post covers the approach we settled on: one Python external transform that drives BICC end to end and lands each Fusion “datastore” in Foundry as a clean, incrementally updated table. There’s no packaged connector involved, just a transforms repo and a Data Connection source.
The transform talks to three Oracle APIs along the way: BICC’s REST API for metadata, the ESS scheduler over SOAP to run the extract, and UCM (WebCenter Content) for the actual file downloads. None of them are complicated on their own; most of the work is just wiring them together in the right order.

What you’ll build

For each datastore you configure, one incremental transform that:

  1. Fetches datastore metadata from the BICC REST API (/biacm/rest)
  2. Creates/updates a BICC extract Job scoped to that datastore
  3. Triggers a VO_AND_PK_EXTRACT via the ESS SOAP API and polls until it finishes
  4. Downloads the output CSV + primary-key files from UCM (SOAP + MTOM), discovered via the job’s manifest
  5. Parses them, filters to currently-active rows using the PK file, deduplicates by primary key, and writes a clean table

How it works

The pipeline touches three Oracle API surfaces. Here’s what each one is responsible for:

Purpose Endpoint Auth that works
Trigger + poll extract (ESS) /bi/ess/esswebservice Basic Auth (WSSE UsernameToken)
Datastore metadata + jobs /biacm/rest HTTP Basic Auth
Download files (UCM) /idcws/GenericSoapPort HTTP Basic Auth (SOAP)

One tip that saved us a lot of time: do the UCM downloads through the SOAP GenericSoapPort endpoint rather than the older /cs/idcplg servlet. On stacks that use IDCS, the servlet expects a browser SSO session and won’t hand back your files over Basic Auth, whereas GenericSoapPort works with Basic Auth directly.

Prerequisites

  • An Oracle Fusion environment with BICC enabled.
  • An Oracle user whose roles allow BICC extraction and UCM download (see Step 1).
  • UCM configured as the BICC external storage (see Step 2).
  • A Foundry Python transforms repository.
  • A Foundry Data Connection source (REST/webhooks) that can reach your Fusion host, with an egress policy (see Step 3).

Step 1 — Oracle user & roles

Use (or create) an Oracle user and grant it a role that includes the following duty roles. Creating users/roles requires the Security Manager role in the Fusion Security Console (Tools → Security Console → Roles → Create Role, category “BI - Abstract Role”).

Duty role Grants
ESS Administrator Role Create and manage schedules for global data extracts / jobs
BIA_ADMINISTRATOR_DUTY Get job information, describe datastores, and perform BICC operations
OBIA_EXTRACTTRANSFORMLOAD_RWD View and download extracted files from UCM

Keep this user’s username and password handy — you’ll store them on the Foundry source in Step 3.

Step 2 — Configure UCM as BICC external storage

  1. Open the BI Applications Configuration Manager: https://<your-fusion-host>/biacm.
  2. Click Configure External Storage.
  3. On the UCM Connection tab, click Test UCM Connection and confirm you get “External Storage Connection Succeeded”.

When you trigger extracts with EXTERNAL_STORAGE_LIST = "UCM", BICC writes the output files (and manifest documents) into UCM under the FAFusionImportExport security group.

Step 3 — Create the Foundry Data Connection source

  1. Create a REST / webhooks source whose host is your Fusion instance (https://<your-fusion-host>).
  2. Attach an egress policy allowing that host on port 443.
  3. Configure Basic Auth with your Oracle user’s username + password. This is the only credential the pipeline needs — no OAuth/JWT/IDCS.
  4. Enable “Allow this source to be imported into code repositories”, then import the source into your transforms repo.
  5. Note the source’s RID — you’ll reference it in code as SOURCE_RID.

Step 4 — The transform code

The transform needs a few libraries beyond the defaults: transforms-external-systems (to use the Data Connection source), zeep (a SOAP client, for the ESS calls), and polars. The easiest way to add them is the Libraries panel in the code repository — search for each one and click Add.
4a. Config + credentials. Read the username/password back out of the source’s pre-configured Basic Auth client. (The SOAP WSSE token needs the plaintext password, so we decode it rather than relying on a pre-built header.)

import base64
import datetime, os, time, re, io, zipfile, logging
from requests import Session
from requests.auth import HTTPBasicAuth
from zeep import Client
from zeep.wsse.username import UsernameToken
from zeep.transports import Transport
from zeep.wsse.utils import WSU, get_timestamp

INSTANCE_URL = "https://<your-fusion-host>"
SOAP_URL     = INSTANCE_URL + "/idcws/GenericSoapPort"   # UCM SOAP (Basic Auth works here)
OUTPUT_BASE  = "/<Your Project>/clean/"
SOURCE_RID   = "ri.magritte..source.<your-source-rid>"
SEMANTIC_VERSION = 1

def get_basic_auth(source):
    """Recover (username, password) from the source's Basic Auth header."""
    client = source.get_https_connection().get_client()
    auth = client.headers.get("Authorization", "")
    if not auth.startswith("Basic "):
        raise RuntimeError("Source has no Basic Auth configured.")
    user, pwd = base64.b64decode(auth[6:]).decode("utf-8").split(":", 1)
    return user, pwd

4b. Trigger the extract over SOAP (ESS). Note the del ... User-Agent line — the Oracle endpoint hangs if a User-Agent header is present.

def soap_client(user, pwd):
    session = Session()
    transport = Transport(session=session, timeout=60)
    del transport.session.headers["User-Agent"]   # Oracle hangs otherwise
    created = datetime.datetime.now()
    ts = WSU("Timestamp")
    ts.append(WSU("Created", get_timestamp(zulu_timestamp=created)))
    ts.append(WSU("Expires", get_timestamp(zulu_timestamp=created + datetime.timedelta(minutes=10))))
    token = UsernameToken(username=user, password=pwd, created=created,
                          nonce=base64.b64encode(os.urandom(16)).decode(),
                          timestamp_token=ts, use_digest=False, hash_password=False)
    return Client(wsdl=INSTANCE_URL + "/bi/ess/esswebservice?wsdl", wsse=token, transport=transport)

def trigger_extraction(user, pwd, job_id):
    svc = soap_client(user, pwd).bind("ESSWebService", "SchedulerServiceImplPort")
    body = {
        "jobDefinitionId": {"name": "BICloudConnectorJobDefinition",
                            "packageName": "oracle.apps.ess.biccc", "type": "JOB_DEFINITION"},
        "application": "oracle.biacm", "requestedStartTime": "",
        "requestParameters": {"parameter": [
            {"dataType": "STRING", "name": "EXTRACT_JOB_TYPE", "value": "VO_AND_PK_EXTRACT", "scope": None},
            {"dataType": "LONG",   "name": "JOB_ID", "value": str(job_id), "scope": None},
            {"dataType": "STRING", "name": "EXTERNAL_STORAGE_LIST", "value": "UCM", "scope": None},
        ]},
    }
    return str(svc.submitRequest(**body))   # returns an ESS request id you then poll

4c. Download from UCM over SOAP. Build a raw SOAP envelope against GenericSoapPort using a session that carries Basic Auth. UCM returns the file as an MTOM multipart attachment, so you pull the octet-stream part out yourself.

def _soap_ucm(session, idc_service, fields):
    field_xml = "\n".join(f'<ucm:Field name="{k}">{v}</ucm:Field>' for k, v in fields.items())
    envelope = f'''<?xml version="1.0" encoding="UTF-8"?>
<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/"
                  xmlns:ucm="http://www.oracle.com/UCM">
  <soapenv:Body>
    <ucm:GenericRequest webKey="cs">
      <ucm:Service IdcService="{idc_service}">
        <ucm:Document>{field_xml}</ucm:Document>
      </ucm:Service>
    </ucm:GenericRequest>
  </soapenv:Body>
</soapenv:Envelope>'''
    return session.post(SOAP_URL, data=envelope.encode(),
                        headers={"Content-Type": "text/xml; charset=utf-8", "SOAPAction": ""})

def _mtom_bytes(response):
    """Return the binary attachment from an MTOM multipart SOAP response."""
    m = re.search(r'boundary="([^"]+)"', response.headers.get("Content-Type", ""))
    if not m:
        return response.content
    for part in response.content.split(b"--" + m.group(1).encode()):
        sep = part.find(b"\r\n\r\n")
        if sep == -1:
            continue
        head, body = part[:sep].decode(errors="replace").lower(), part[sep + 4:].rstrip(b"\r\n")
        if "application/octet-stream" in head and body:
            return body
    return None

def ucm_session(user, pwd):
    s = Session(); s.auth = HTTPBasicAuth(user, pwd); return s   # Basic Auth is all you need

4d. Find the files via the job manifest. Don’t try to find extract files by filename search — instead the extract produces manifest documents named MANIFEST_DATA_{job_id} and MANIFEST_PRIMARY_KEYS_{job_id}. Search for the manifest, parse it to get the exact document IDs, then GET_FILE each.

def _search_ucm(session, query):
    resp = _soap_ucm(session, "GET_SEARCH_RESULTS",
                     {"QueryText": f"<qsch>{query}</qsch>", "ResultCount": "20",
                      "SortField": "dInDate", "SortOrder": "Desc", "IsJson": "1"})
    xml = resp.text  # (parse MTOM xml part in practice)
    return re.findall(r'name="dID">(\d+)<', xml)

def _download(session, doc_id):
    return _mtom_bytes(_soap_ucm(session, "GET_FILE", {"dID": doc_id, "allowInterrupt": "1"}))

def download_extraction(session, job_id):
    # manifest lists the data-file doc IDs; each downloaded file is a zip of CSVs
    manifest_ids = _search_ucm(session, f"MANIFEST_DATA_{job_id}")
    data_doc_ids = []
    for mid in manifest_ids:
        text = _download(session, mid).decode(errors="replace")
        for line in text.strip().splitlines():
            parts = line.split(";")
            if len(parts) >= 2 and "=" not in parts[0]:
                data_doc_ids.append(parts[1])
    csvs = []
    for did in data_doc_ids:
        blob = _download(session, did)
        if blob[:2] == b"PK":                       # it's a zip
            with zipfile.ZipFile(io.BytesIO(blob)) as z:
                csvs += [z.read(f) for f in z.namelist() if f.endswith(".csv")]
        else:
            csvs.append(blob)
    return csvs   # (do the same for MANIFEST_PRIMARY_KEYS_{job_id} to get the active-key file)

4e. Generate one incremental transform per datastore. Tie it together: metadata → job → trigger → poll → download → parse → PK-dedup → write. It’s @lightweight (single-node Polars) and @incremental.

import polars as pl
from transforms.api import transform, lightweight, incremental, Output, LightweightOutput
from transforms.external.systems import external_systems, Source

def create_extract_transform(datastore):
    output_path = OUTPUT_BASE + datastore

    @lightweight(memory_gb=16)
    @incremental(semantic_version=SEMANTIC_VERSION)
    @external_systems(oracle_source=Source(SOURCE_RID))
    @transform(output=Output(output_path))
    def extract(ctx, oracle_source: Source, output: LightweightOutput, datastore=datastore):
        user, pwd = get_basic_auth(oracle_source)
        rest = Session(); rest.auth = HTTPBasicAuth(user, pwd)

        meta   = rest.get(f"{INSTANCE_URL}/biacm/rest/meta/datastores/{datastore}").json()
        pk_cols = [c["name"].upper() for c in meta["columns"]
                   if c.get("isPopulate") and c.get("isPrimaryKey")]
        job_id  = create_or_update_job(rest, datastore, meta)      # PUT /biacm/rest/meta/jobs/
        req_id  = trigger_extraction(user, pwd, job_id)
        poll_extraction(user, pwd, req_id)                          # poll getRequestState until SUCCEEDED

        csvs = download_extraction(ucm_session(user, pwd), job_id)
        new = pl.concat([pl.read_csv(io.BytesIO(c), infer_schema=False) for c in csvs])
        new = new.rename({c: c.upper() for c in new.columns})
        new = new.with_columns(pl.lit(float(time.time())).alias("ingest_timestamp"))

        if ctx.is_incremental:
            prev = output.polars(mode="previous")
            if prev is not None and prev.height:
                new = (pl.concat([prev, new], how="diagonal")
                         .sort("ingest_timestamp", descending=True)
                         .unique(subset=pk_cols, keep="first"))
        output.set_mode("replace")
        output.write_table(new)

    extract.__name__ = f"bicc_extract_{datastore.split('.')[-1]}"
    return extract

TRANSFORMS = [create_extract_transform(ds) for ds in DATASTORES_TO_INGEST]

Step 5 — Choose which datastores to ingest

Keep a small config.py as the single knob. Each fully-qualified datastore name produces one clean output table:

# config.py
DATASTORES_TO_INGEST = [
    "FscmTopModelAM.AnalyticsServiceAM.TerritoriesTLPVO",
    "CrmAnalyticsAM.PartiesAnalyticsAM.Organization",
    "CrmAnalyticsAM.PartiesAnalyticsAM.Location",
    # ...add the datastores you want
]

To find datastore names, browse Manage Offerings and Data Stores in the BICC UI (datastores are grouped by offering). If a datastore Supports Incremental and already has a non-null Last Extract Date from another consumer, use a dedicated/duplicated datastore — the Last Extract Date is tracked on the datastore, so a shared one can cause gaps.

Step 6 — Build & schedule

  1. Commit and let CI publish the transforms.
  2. Build each /clean/<datastore> output. Each build triggers a live BICC extract, so it can take a couple of minutes.
  3. Add schedules for recurring ingestion. Stagger them so you don’t fire many concurrent BICC/ESS jobs at once.

Known behaviors & limitations

Columns are ingested as strings. CSVs are read with infer_schema=False, so every column is a string — a safe “raw” layer that avoids precision loss on large IDs and misparsed dates. To type them, map the BICC metadata (dataType/scale/precision) to Polars types and apply schema_overrides, or cast downstream.
Extracts are not deleted from UCM. The pipeline only reads; extract files accumulate in UCM. This doesn’t affect correctness (incremental runs filter manifests by creation date), but to reclaim space either rely on Oracle’s own BICC/UCM retention, or add a DELETE_DOC call over the same SOAP port after a successful download (requires delete permission; only safe if Foundry is the sole consumer).

Scaling considerations

  • Per-run cost grows with total table size — the incremental step reads, dedups, and rewrites the whole cumulative table each run. Periodically snapshot-compact, or push dedup into a downstream Spark transform.
  • Single-node memory ceiling — downloaded bytes, unzipped files, the previous output, and the full PK file must fit in one node’s RAM. For very large datastores (>~50 GB) use a distributed @transform.spark variant instead of @lightweight.
  • Verify BICC-side incremental — make sure the datastore’s incremental (lastUpdateDate) extraction is configured so Oracle ships only deltas, not a full extract every run.
  • Oracle job concurrency — stagger schedules to avoid overloading the ESS/BICC scheduler.

Once the source is set up, adding another datastore is a one-line change in config.py and it shows up in Foundry as its own incrementally maintained table.

3 Likes