ODCS Generator Reference

The ODCS generator module provides integrations for generating ODCS YAML from supported catalog platforms.

Collibra Integration

CollibraClient

class wxdi.odcs_generator.generate_odcs_from_collibra.CollibraClient(base_url: str, username: str, password: str)

Bases: object

Client for interacting with Collibra REST API

Parameters:

• base_url (str)

• username (str)

• password (str)

HEADERS_JSON = {'Accept': 'application/json'}

HEADERS_CONTENT_JSON = {'Content-Type': 'application/json'}

DEFAULT_LIMIT = 1000

get_asset(asset_id: str)

Fetch asset details from Collibra

Return type:

Dict[str, Any]

Parameters:

asset_id (str)

get_asset_attributes(asset_id: str)

Fetch all attributes for an asset

Return type:

List[Dict[str, Any]]

Parameters:

asset_id (str)

get_asset_relations(asset_id: str, as_source: bool = True)

Fetch all relations for an asset

Parameters:

• asset_id (str) – The asset ID

• as_source (bool, default: True) – If True, fetch relations where asset is source; if False, where asset is target

Return type:

List[Dict[str, Any]]

get_asset_tags(asset_id: str)

Fetch tags directly assigned to an asset

Return type:

List[str]

Parameters:

asset_id (str)

get_asset_classifications(asset_id: str)

Fetch data classifications for an asset using GraphQL API

Return type:

List[str]

Parameters:

asset_id (str)

ODCSGenerator

class wxdi.odcs_generator.generate_odcs_from_collibra.ODCSGenerator(collibra_client: CollibraClient)

Bases: object

Generate ODCS YAML from Collibra asset metadata

Parameters:

collibra_client (CollibraClient)

UTC_TIMEZONE_SUFFIX = '+00:00'

EXCLUDED_ATTRIBUTES = {'Description'}

LOGICAL_TYPE_MAPPING = {'array': 'array', 'date': 'date', 'date time': 'timestamp', 'decimal number': 'number', 'geographical': 'string', 'integer': 'integer', 'n/a': None, 'number': 'number', 'object': 'object', 'string': 'string', 'text': 'string', 'time': 'time', 'true/false': 'boolean', 'whole number': 'integer'}

NUMERIC_TYPES = ['DECIMAL', 'NUMERIC', 'NUMBER']

generate_odcs(asset_id: str)

Generate ODCS structure from a single Collibra asset

Parameters:

asset_id (str) – Collibra asset ID to include in the contract

Return type:

Dict[str, Any]

Returns:

ODCS data contract dictionary with single asset in schema array

Raises:

ValueError – If asset is not a table

Informatica Integration

InformaticaClient

class wxdi.odcs_generator.generate_odcs_from_informatica.InformaticaClient(base_url, username, password)

Bases: object

Client for interacting with Informatica CDGC API.

CONTENT_TYPE_JSON = 'application/json'

HEADERS_JSON = {'Accept': 'application/json'}

HEADERS_CONTENT_JSON = {'Content-Type': 'application/json'}

get_session_id()

Return type:

Dict[str, Any]

get_auth_token()

Get authentication token with caching to avoid repeated auth calls.

Return type:

str

get_asset_details(asset_id: str)

Fetch asset details including table and column information.

Return type:

Dict[str, Any]

Parameters:

asset_id (str)

validate_asset_is_table(asset_data: Dict[str, Any])

Validate that the asset is a table and not a schema or other type.

Parameters:

asset_data (Dict[str, Any]) – The asset data returned from Informatica API

Raises:

ValueError – If the asset is not a table

Return type:

None

get_column_details(column_id: str)

Fetch detailed information for a specific column.

Return type:

Dict[str, Any]

Parameters:

column_id (str)