NexusPythonWrapper 1.0.0

REST API Wrapper library for NEXUS Integrity Centre

NEXUS Python Wrapper

---

NexusPythonWrapper is a Python library for interacting with the asset integrity database solution from Wood PLC, NEXUS Integrity Centre (IC).

The library is built around the NexusWrapper class, which handles authentication and session management, and exposes a set of dedicated clients — one per REST resource area (rows, reports, imports, jobs, etc.) — as attributes on the connection object. Every call returns a normalised Result object, and failed HTTP responses are translated into typed exceptions instead of raw requests errors.

NEXUS IC REST API Documentation: https://docs.nexusic.com/6.9/ic-web.rest.v2.html

Features

---

• Simplified session management (login/logout, hash handling)
• A dedicated client for each REST resource area
• Consistent, typed response handling via Result
• Typed exceptions for non-2xx responses
• Powerful filtering and sorting helper classes for row queries
• Automatic pagination for large row sets

Installation

---

pip install NexusPythonWrapper

Authentication

---

The NexusWrapper class has two authentication methods available, either BASIC or APIKEY. Regardless of authentication method, the uri (IC-Web address) must be provided — the domain name or IP address of the server hosting the REST service, e.g. database.nexusic.com.

NexusWrapper accepts the following parameters:

uri:                 str                # 'https://database.nexusic.com/'
authentication_type: str | AuthenticationType   # 'BASIC' or 'APIKEY'
api_key:             str = None         # required when authentication_type == 'APIKEY'
username:            str = None         # required when authentication_type == 'BASIC', e.g. '[e-mail]'
password:            str = None         # required when authentication_type == 'BASIC'
logger:              logging.Logger = None
ssl_verify:          bool = True

from nexus_python_wrapper import NexusWrapper

conn = NexusWrapper(
    uri="https://database.nexusic.com/",
    authentication_type="BASIC",
    username="[e-mail]",
    password="goodPassword!"
)

On instantiation, NexusWrapper:

1. Validates authentication_type and the associated credentials.
2. Base64-encodes the credentials and POSTs to the login endpoint, storing the returned hash (used as an authentication token on every subsequent request) along with the user's id, username, display name, and license type.
3. Sets up every resource client (rows, approvals, blobs, dashboards, functions, imports, jobs, license, reports, settings, versions).
4. GETs the version endpoint to record the database version and schema.

NexusWrapper also supports the context manager protocol, which automatically logs out and closes the session on exit:

with NexusWrapper(uri=..., authentication_type="APIKEY", api_key="...") as conn:
    result = conn.rows.get_row(table_name="Integrity_Assessment", key_value=123)

Print a summary of the current connection:

conn.connection_details()

Database:         https://database.nexusic.com
Database Version: 6.9.XXX
Database Schema:  8.125
User:             Joe Bloggs
License:          write

The hash is valid for 60 minutes from the last request. To invalidate it explicitly:

conn.logout()

The Result Object

---

Every client method returns a Result instance — a single, consistent shape for JSON, plain text, and binary responses.

Attribute	Description
status_code	HTTP status code
message	HTTP reason phrase
content_type	Response Content-Type header
kind	ResultKind.JSON, .TEXT, .BINARY, or .EMPTY
rows	list[dict] — populated for JSON responses
metadata	Everything from the JSON envelope other than rows (e.g. pageSize, startRow, totalRows, stamp)
text	Populated for text/* responses
content	Populated for binary responses (e.g. blobs, files)
headers	Raw response headers

Convenience helpers:

result.ok          # True for any 2xx status code
result.first        # first row, or None
result.value        # first row (JSON) / text / bytes, depending on kind
result.has_more     # True if the row-set is paginated and more rows remain
result.stamp        # pagination stamp, if present

result.save("output.pdf")     # writes .content or .text to disk
result.extend(other_result)   # merges another page of rows into this Result

len(result)         # number of rows
for row in result:  # iterate rows directly
    ...
result[0]           # index into rows directly
bool(result)        # same as result.ok

Error Handling

---

Non-2xx responses raise a typed exception (all inherit from NexusError) instead of returning a Result, so you can branch on the specific failure:

Exception	Meaning
AuthenticationError	Login/authentication with NEXUS failed
CredentialsError	Supplied credentials were invalid/incomplete for the chosen auth type
DecodingError	Response body could not be decoded as promised by its Content-Type
NotFoundError	404 — row, field, table, or resource does not exist
ForbiddenError	403 — user lacks permission for the action
ValidationError	422 — row data failed validation
PayloadTooLargeError	404.13 — uploaded blob/file exceeded the server's max request size
ServerError	5xx — NEXUS returned a server-side error
NexusResponseError	Base class for any other non-2xx response

from nexus_python_wrapper.exceptions import NotFoundError

try:
    row = conn.rows.get_row(table_name="Integrity_Assessment", key_value=123)
except NotFoundError as exc:
    print(exc.status_code, exc.message, exc.body)

Rows Client (conn.rows)

---

Handles tables, rows, fields, and raw Business Object blueprints.

Get Rows

result = conn.rows.get_rows(
    table_name="Integrity_Assessment",
    filter=my_filter,          # optional NexusFilter, sent as X-NEXUS-Filter header
    sort=my_sort,               # optional NexusSort, sent as X-NEXUS-Sort header
    paginated=True,              # automatically fetches every page (default: True)
    page_size=100,
    start_row=0,
    calculated_values=True
)

Passing business_object=True instead returns the raw Business Object blueprint (its field definitions) rather than row data.

Get a Single Row

result = conn.rows.get_row(table_name="Integrity_Assessment", key_value=123)

Get a Single Field

result = conn.rows.get_field(
    table_name="Integrity_Assessment",
    key_value=123,
    field_name="Assessment",
    format=None,   # e.g. for blob/document fields
    page=None
)

Create a Row (PUT)

body = {
    "Component_ID": 1,
    "Assessment_Date": "2025-04-26",
    "Inspection_Date": "2025-04-01",
    "Inspection_Activity": 4,
    "Assessment": "Light surface corrosion noted throughout the corrosion circuit, not considered a concern at this time"
}
result = conn.rows.put(table_name="Integrity_Assessment", request_body=body)

Update a Row (POST)

body = {
    "Inspection_Date": "2025-04-14",
    "Assessment": "Severe internal corrosion noted at the 6 o'clock position of the pipeline."
}
result = conn.rows.post(table_name="Integrity_Assessment", key_value=123, request_body=body)

Delete a Row

result = conn.rows.delete(table_name="Integrity_Assessment", key_value=123)

NEXUS returns no content for a successful delete. The wrapper fetches the row before deleting it and attaches that data to the returned Result, so you retain full visibility into what was removed.

Validate a Row

Validates a row payload against the server without saving it:

result = conn.rows.validate(table_name="Integrity_Assessment", request_body=body, key_value=0)

Filtering and Sorting

---

NexusFilter builds the X-NEXUS-Filter header used with get_rows:

from nexus_python_wrapper import NexusFilter

f = NexusFilter(operator="and")
f.where(field="Status", value="Open", method="eq")
f.where(field="Priority", method="in", items=[1, 2, 3])

result = conn.rows.get_rows(table_name="Integrity_Assessment", filter=f)

• where(field, value=None, method=None, invert=None, items=None) — add a clause. Supported methods: eq, lt, gt, le, ge, like, in, pa, ch. Use items (not value) when method="in". Set invert=True to negate the clause.
• nested(other_filter) — embed another NexusFilter as a nested condition group.
• and_() / or_() — switch the top-level operator.

NexusSort builds the X-NEXUS-Sort header:

from nexus_python_wrapper import NexusSort

s = NexusSort()
s.sort("Assessment_Date", ascending=False)

result = conn.rows.get_rows(table_name="Integrity_Assessment", sort=s)

Approvals Client (conn.approvals)

---

Operations on NEXUS IC change requests.

conn.approvals.get()                                  # all pending/approved/actioned requests
conn.approvals.approve(request_id=101, comments="Looks good")
conn.approvals.reject(request_id=102, comments="Needs revision")

Blobs Client (conn.blobs)

---

Uploads binary content (used ahead of imports, file fields, etc.) and returns a blob GUID.

with open("data.zip", "rb") as f:
    result = conn.blobs.upload(data=f.read())

Dashboards Client (conn.dashboards)

---

Retrieves the layout and data for a dashboard by key, optionally passing parameter values.

result = conn.dashboards.get(key=5)
result = conn.dashboards.get(key=5, parameters={"Region": "North Sea"})

Functions Client (conn.functions)

---

Inspects and executes NEXUS server-side functions.

from nexus_python_wrapper.clients.functions import FunctionFormat

details = conn.functions.get_details(function_id="CorrosionRate")

result = conn.functions.execute(
    function_id="CorrosionRate",
    parameters={"ComponentId": 1},
    format=FunctionFormat.VALUE   # or FunctionFormat.PNG
)

Imports Client (conn.imports)

---

Imports tabulated data into NEXUS tables from a previously uploaded blob.

contents = conn.imports.get_contents(file_id=blob_guid)     # enumerate files in an uploaded zip

columns = conn.imports.detect_columns(
    file=blob_guid,
    business_object="Integrity_Assessment",
    header_rows=1,
    delimiter=",",
    qualifier="quote"
)

job = conn.imports.execute(
    file=blob_guid,
    business_object="Integrity_Assessment",
    date_format="yyyy/MM/dd",
    time_format="HH:nn:ss",
    test=False   # set True to run a validation-only test job
)

execute starts an asynchronous job — track it with the Jobs client (below).

Jobs Client (conn.jobs)

---

Tracks asynchronous server-side jobs (imports, report generation, etc.).

conn.jobs.status(job_id="...")   # omit job_id to list all jobs
conn.jobs.content(job_id="...")  # retrieve job output/content
conn.jobs.dismiss(job_id="...")  # dismiss a completed job

License Client (conn.license)

---

result = conn.license.get()

Reports Client (conn.reports)

---

from nexus_python_wrapper.clients.reports import OutputTypes

details = conn.reports.get_report_details(key=42)

result = conn.reports.generate_report(
    report_template=42,
    output_type=OutputTypes.PDF,
    parameters=[{"parameterName": "StartDate", "value": "2025-01-01"}],
    excluded_groups=[3]
)

recipients, embed_html, and subject are only valid when output_type=OutputTypes.HTML (used for emailing the generated report):

result = conn.reports.generate_report(
    report_template=42,
    output_type=OutputTypes.HTML,
    recipients="[e-mail]",
    subject="Monthly Integrity Report",
    embed_html=True
)

Settings Client (conn.settings)

---

CRUD operations for user UI preferences and saved Grid/Tree layouts.

from nexus_python_wrapper.clients.settings import SettingCategory

result = conn.settings.get(SettingCategory.GRID, sub_category="Anomaly")

conn.settings.update([
    {"category": "Grid", "subCategory": "Anomaly", "identifier": "MyView", "value": "..."}
])

# NEXUS often stores setting values as stringified JSON — unpack them safely:
value = conn.settings.unpack_json(result.first)

Versions Client (conn.versions)

---

result = conn.versions.get()