`module` `oaaclient.client`

Classes for calling Veza APIs and managing OAA providers and data sources.

Use of this source code is governed by the MIT license that can be found in the LICENSE file or at https://opensource.org/licenses/MIT.

Global Variables

OAACLIENT_VERSION
PROVIDER_ICON_MAX_SIZE

`function` `report_builder_entrypoint`

python

report_builder_entrypoint() → None

Entry point for oaaclient-report-builder command

Reads a JSON file and passes it to the oaaclient.utils.build_report method

`function` `main`

python

main()

`class` `OAAClientError`

Error raised by OAAClient.

Raised for issues connecting to the OAA API and when the API returns an error.

Args:

error (str): short string for error message
message (str): detailed error message
status_code (int, optional): status code for HTTP related errors. Defaults to None.
details (list, optional): list of additional details for error. Defaults to None.

`method` `init`

python

__init__(
    error: 'str',
    message: 'str',
    status_code: 'int' = None,
    details: 'list' = None
) → None

`class` `OAAResponseError`

Error returned from API Call

`method` `init`

python

__init__(*args, **kwargs)

`class` `OAAConnectionError`

Error with API Connection

`method` `init`

python

__init__(
    error: 'str',
    message: 'str',
    status_code: 'int' = None,
    details: 'list' = None
) → None

`class` `OAAClient`

Class for OAA API Connection and Management.

Utilities for OAA-related operations with Veza API calls. Manages custom providers and data sources, and can push OAA payloads from JSON or template objects.

Connection url and API key can be automatically loaded from OS environment values if set. To utilize environment variables initialize OAAClient without providing a URL or API key value and set the VEZA_URL and VEZA_API_KEY OS environment variables.

Args:

url (str, optional): URL for Veza instance.
api_key (str, optional): Veza API key.
username (str, optional): Not used (legacy). Defaults to None.
token (str, optional): Legacy parameter name for API key. Defaults to None.

Attributes:

url (str): URL of the Veza instance to connect to
api_key (str): Veza API key
enable_compression (bool): Enable or disable compression of the OAA payload during push, defaults to enabled (True)

Raises:

OAAClientError: For errors connecting to API and if API returns errors

`method` `init`

python

__init__(
    url: 'str' = None,
    api_key: 'str' = None,
    username: 'str' = None,
    token: 'str' = None
)

`method` `add_query_report`

python

add_query_report(report_id: 'str', query_id: 'str') → dict

Add a Query to a Report

Adds a Query to an existing Report by ID

Args:

report_id (str): Report UUID to add Query to
query_id (str): Query UUID to add

Returns:

dict: API Response

`method` `api_delete`

python

api_delete(api_path: 'str', params: 'dict' = None) → dict

Perform REST API DELETE operation.

Args:

api_path (str): API Path API path relative to Veza URL

Raises:

OAAResponseError: API returned an error
OAAConnectionError: Connection error during HTTP operation

Returns:

dict: API response from call

`method` `api_get`

python

api_get(api_path: 'str', params: 'dict' = None) → list | dict

Perform a Veza API GET operation.

Makes the GET API call to the given path and processes the API response. Returns the value or values result returned from the API.

For API endpoints that return a list like /api/v1/providers/custom function will return a list of entities or an empty list if the API returns no results.
For API endpoints that are a specific ID such as /api/v1/providers/custom/<uuid> function will return the dictionary result of the JSON returned by the API.

Args:

api_path (str): API path relative to Veza URL (example /api/v1/providers).

Raises:

OAAResponseError: API returned an error
OAAConnectionError: Connection error during HTTP operation

Returns:

list|dict: Returns list or dict based on API destination

`method` `api_patch`

python

api_patch(api_path: 'str', data: 'dict', params: 'dict' = None) → dict

Perform REST API PATCH operation.

Args:

api_path (str): API Path API path relative to Veza URL

Raises:

OAAResponseError: API returned an error
OAAConnectionError: Connection error during HTTP operation

Returns:

dict: API response from call

`method` `api_post`

python

api_post(api_path: 'str', data: 'dict', params: 'dict' = None) → list | dict

Perform a Veza API POST operation.

Call POST on the supplied Veza instance API path, submitting a data payload.

Returns value or values response from API result. Paginated responses are automatically processed to collect all responses a single list.

Args:

api_path (str): API path relative to Veza URL example /api/v1/providers
data (dict): dictionary object included as JSON in body of POST operation
params (dict, optional): Optional HTTP query parameters. Defaults to empty dictionary.

Raises:

OAAResponseError: API returned an error
OAAConnectionError: Connection error during HTTP operation

Returns:

dict: API response as dictionary

`method` `api_put`

python

api_put(api_path: 'str', data: 'dict', params: 'dict' = None) → list | dict

Perform Veza API PUT operation.

Call PUT on the supplied Veza instance API path, including the data payload.

Returns value or values response from API result. Paginated responses are automatically processed to collect all responses a single list.

Args:

api_path (str): API path relative to Veza URL example /api/v1/providers
data (dict): dictionary object included as JSON in body of PUT operation
params (dict, optional): Optional HTTP query parameters. Defaults to empty dictionary.

Raises:

OAAResponseError: API returned an error
OAAConnectionError: Connection error during HTTP operation

Returns:

dict: API response as dictionary

`method` `create_data_source`

python

create_data_source(
    name: 'str',
    provider_id: 'str',
    options: 'dict | None' = None
) → dict

Create a new Data Source for the given Provider ID.

Args:

name (str): Name for new Data Source
provider_id (str): Unique identifier for the Provider
options: (dict, optional): Additional arguments to be included with data source create call to Veza. Defaults to None.

Raises:

ValueError: Data source name contains invalid characters

Returns:

dict: dictionary of new Data Source

`method` `create_datasource`

python

create_datasource(name, provider_id)

Deprecated Legacy function for backward-compatibility.

`method` `create_provider`

python

create_provider(
    name: 'str',
    custom_template: 'str',
    base64_icon: 'str' = '',
    options: 'dict | None' = None
) → dict

Create a new Provider.

Creates a new Provider with the given name. An error will be raised for naming conflicts.

Args:

name (str): new Provider name
custom_template (str): the OAA template to use for the Provider ("application" or "identity_provider")
base64_icon (str, optional): Base64 encoded string of icon to set for Provider. Defaults to "".
options: (dict, optional): Additional arguments to be included with provider create call to Veza. Defaults to None.

Raises:

ValueError: Provider name contains invalid characters

Returns:

dict: dictionary representing the created Provider

`method` `create_query`

python

create_query(query: 'dict') → dict

Create a new Assessment Query

For details on how to define an Assessment Query see the Veza docs.

Args:

query (dict): Query definition

Returns:

dict: API response including ID of created Query

`method` `create_report`

python

create_report(report: 'dict') → dict

Create a new Report

For details on how to define a new Report see the Veza docs.

Args:

report (dict): Report definition

Returns:

dict: API response including ID of created Report

`method` `delete_data_source`

python

delete_data_source(data_source_id: 'str', provider_id: 'str') → dict

Delete an existing Data Source by ID.

Removes a Data Source and all entity data.

Args:

data_source_id (str): ID of Data Source to delete
provider_id (str): ID of Data Source Provider

Returns:

dict: API response

`method` `delete_provider`

python

delete_provider(provider_id: 'str') → dict

Delete an existing provider by ID.

Deleting a provider removes all datasources and historical data. Fully deleting the provider is a background operation that will complete after API response is returned.

Args:

provider_id (str): ID of provider to delete

Returns:

dict: API response

`method` `delete_query`

python

delete_query(id: 'str', force: 'bool' = False) → dict

Delete an Assessment Query by ID

Args:

id (str): UUID of Query to delete
force (bool): Force deletion of query that may be part of a report. Defaults to False

Returns:

dict: API response from delete

`method` `delete_report`

python

delete_report(id: 'str') → dict

Delete Report by ID

Args:

id (str): UUID of Report to delete

Returns:

dict: API response

`method` `get_data_source`

python

get_data_source(name: 'str', provider_id: 'str') → dict

Get Provider's Data Source by name.

Find a Data Source from a specific provider based on the name of the Data Source

Args:

name (str): Data Source name
provider_id (str): Provider unique ID

Returns:

dict: Data Source as dict or None

`method` `get_data_sources`

python

get_data_sources(provider_id: 'str') → list[dict]

Get Data Sources for Provider by ID.

Get the list of existing Data Sources, filtered by Provider UUID.

Args:

provider_id (str): ID of Provider

Returns:

list[dict]: List of Data Sources as dictionaries

`method` `get_provider`

python

get_provider(name: 'str') → dict

Get Provider by name.

Args:

name (str): name of Provider

Returns:

dict: dictionary representing Provider or None

`method` `get_provider_by_id`

python

get_provider_by_id(provider_id: 'str') → dict

Get Provider by UUID identifier.

Args:

provider_id (str): Unique global identifier for provider

Returns:

dict: dictionary representation of Provider or None

`method` `get_provider_list`

python

get_provider_list() → list[dict]

Return list of Providers.

Returns:

list[dict]: Returns a list of existing Providers as dictionaries

`method` `get_queries`

python

get_queries(include_inactive_queries: 'bool' = True) → list[dict]

Get all saved Assessment Queries

Veza can filter out queries that include inactive entity types (e.g. Okta Users on a system without Okta configured). To only retrieve queries that include active entity types set include_inactive_queries to False.

Args:

include_inactive_queries (bool): Set False to exclude inactive queries from result. Defaults to True.

Returns:

list: List of assessment Queries as dictionaries

`method` `get_query_by_id`

python

get_query_by_id(id: 'str') → dict

Get Assessment Query by ID

Args:

id (str): UUID identifier for Query

Returns:

dict: Query definition

`method` `get_report_by_id`

python

get_report_by_id(id: 'str', include_inactive_queries: 'bool' = True) → dict

Get Report by ID

Veza can filter out queries from reports that only contain entity types that are not configured (e.g. Okta Users on a system without Okta configured). To only return queries configured on the report that match entity types configured on the system set include_inactive_queries to False

Args:

id (str): UUID of Report to get
include_inactive_queries (bool): Set True to include inactive queries. Default True.

Returns:

dict: Report definition

`method` `get_reports`

python

get_reports(
    include_inactive_reports: 'bool' = True,
    include_inactive_queries: 'bool' = True
) → list[dict]

Get all Reports

Gets Reports created on the system. To get all reports include_inactive_reports and include_inactive_queries must be set to True.

Args:

include_inactive_reports (bool, Optional): Set to True to include reports that contain no active providers, defaults to True.
include_inactive_queries (bool, Optional): Set to True to include reports that contain only inactive queries, defaults to True.

Returns:

list[dict]: List of Reports as dictionary objects

`method` `push_application`

python

push_application(
    provider_name: 'str',
    data_source_name: 'str',
    application_object: 'CustomApplication | CustomIdPProvider',
    save_json: 'bool' = False,
    create_provider: 'bool' = False,
    options: 'dict | None' = None
) → dict

Push an OAA Application Object (such as CustomApplication).

Extracts the OAA JSON payload from the supplied OAA class (e.g. CustomApplication, CustomIdPProvider, etc) and push to the supplied Data Source.

The Provider with provider_name must be a valid existing Provider or create_provider must be set to True. A new data source will be created automatically by default if it does not already exist.

For logging, and debugging, the optional save_json flag writes the payload to a local file (before push). Output file name is formatted with a timestamp: {data source name}-{%Y%m%d-%H%M%S}.json

Args:

provider_name (str): Name of an existing Provider.
data_source_name (str): Name for Data Source (will be created if it doesn't exist).
application_object (Class): OAA object to extract the payload from
save_json (bool, optional): Save the JSON payload to a local file before push. Defaults to False.
create_provider (bool, optional): Create a new Provider if Provider does not already exists. Defaults to False.
options (dict, optional): Additional dictionary of key/values to be included in push API call. Defaults to None.

Raises:

OAAClientError: If any API call returns an error (including errors processing the OAA payload).

Returns:

dict: API response from push, including any warnings that are returned.

`method` `push_metadata`

python

push_metadata(
    provider_name: 'str',
    data_source_name: 'str',
    metadata: 'dict',
    save_json: 'bool' = False,
    options: 'dict | None' = None
) → dict

Push an OAA payload dictionary to Veza.

Publishes the supplied metadata dictionary representing an OAA payload to the specified provider and data source. The function will create a new data source if it does not already exist, but requires the Provider be created ahead of time.

Args:

provider_name (str): Name of existing Provider
data_source_name (str): Name for Data Source, will be created if doesn't exist.
metadata (dict): Dictionary of OAA payload to push.
save_json (bool, optional): Save the OAA JSON payload to a local file before push. Defaults to False.
options (dict, optional): Additional dictionary of key/values to be included in push API call. Defaults to None.

Raises:

OAAClientError: If any API call returns an error including errors processing the OAA payload.

Returns:

dict: API response to the push request (including any warnings).

`method` `update_provider_icon`

python

update_provider_icon(provider_id: 'str', base64_icon: 'str') → None

Update an existing custom provider icon from base64 encoded string.

To load an icon from file, use utils.encode_icon_file to get the base64 encoding of the file first

Args:

provider_id (str): unique ID of existing provider
base64_icon (str): base64 encoded string of new icon

Raises:

ValueError: If icon size exceeds maximum allowed size

`method` `update_report`

python

update_report(report_id: 'str', report: 'dict') → dict

Update an existing report

Args:

report_id (str): UUID of Report to updated
report (dict): Updated Report definition

Returns:

dict: API response

`method` `update_user_agent`

python

update_user_agent(extra: 'str' = '') → None

Updates the User-Agent string passed with all API calls

Generates a User-Agent with the oaaclient version, Python version and platform information.

The optional extra string will be appended if provided.

Args:

extra (str, optional): Additional information to append to User-Agent string. Defaults to "".

`class` `OAARetry`

Super class for urllib3.util.retry

Super class to allow modifying the default max backoff time from 120 seconds to our own value

Args:

Retry (type): description

`method` `init`

python

__init__(backoff_max=30, **kwargs) → None

oaaclient modules

module oaaclient.client ​

Global Variables ​

function report_builder_entrypoint ​

function main ​

class OAAClientError ​

method __init__ ​

class OAAResponseError ​

method __init__ ​

class OAAConnectionError ​

method __init__ ​

class OAAClient ​

method __init__ ​

method add_query_report ​

method api_delete ​

method api_get ​

method api_patch ​

method api_post ​

method api_put ​

method create_data_source ​

method create_datasource ​

method create_provider ​

method create_query ​

method create_report ​

method delete_data_source ​

method delete_provider ​

method delete_query ​

method delete_report ​

method get_data_source ​

method get_data_sources ​

method get_provider ​

method get_provider_by_id ​

method get_provider_list ​

method get_queries ​

method get_query_by_id ​

method get_report_by_id ​

method get_reports ​

method push_application ​

method push_metadata ​

method update_provider_icon ​

method update_report ​

method update_user_agent ​

class OAARetry ​

method __init__ ​

`module` `oaaclient.client`

Global Variables

`function` `report_builder_entrypoint`

`function` `main`

`class` `OAAClientError`

`method` `init`

`class` `OAAResponseError`

`method` `init`

`class` `OAAConnectionError`

`method` `init`

`class` `OAAClient`

`method` `init`

`method` `add_query_report`

`method` `api_delete`

`method` `api_get`

`method` `api_patch`

`method` `api_post`

`method` `api_put`

`method` `create_data_source`

`method` `create_datasource`

`method` `create_provider`

`method` `create_query`

`method` `create_report`

`method` `delete_data_source`

`method` `delete_provider`

`method` `delete_query`

`method` `delete_report`

`method` `get_data_source`

`method` `get_data_sources`

`method` `get_provider`

`method` `get_provider_by_id`

`method` `get_provider_list`

`method` `get_queries`

`method` `get_query_by_id`

`method` `get_report_by_id`

`method` `get_reports`

`method` `push_application`

`method` `push_metadata`

`method` `update_provider_icon`

`method` `update_report`

`method` `update_user_agent`

`class` `OAARetry`

`method` `init`