4.4.6.2. CSV

CSV export callbacks.

class axonius_api_client.api.asset_callbacks.base_csv.Csv(apiobj, store, state=None, getargs=None)[source]

Bases: ExportMixins

Callbacks for formatting asset data and exporting it in CSV format.

Examples

Create a client using axonius_api_client.connect.Connect and assume apiobj is either client.devices or client.users

>>> apiobj = client.devices  # or client.users
  • args_map() for callback generic arguments to format assets.

  • args_map_custom() for callback specific arguments to format and export data.

Parameters:
classmethod args_map_custom()[source]

Get the custom argument names and their defaults for this callbacks object. :rtype: dict

Examples

Export the output to STDOUT. If export_file is not supplied, the default is to print the output to STDOUT.

>>> assets = apiobj.get(export="csv")

Export the output to a file in the default path axonius_api_client.setup_env.DEFAULT_PATH.

>>> assets = apiobj.get(export="csv", export_file="test.csv")

Export the output to an absolute path file (ignoring export_path) and overwrite the file if it exists.

>>> assets = apiobj.get(
...     export="csv",
...     export_file="/tmp/output.csv",
...     export_overwrite=True,
... )

Export the output to a file in a specific dir.

>>> assets = apiobj.get(export="csv", export_file="output.csv", export_path="/tmp")

Include the schema of all selected fields in the output.

>>> assets = apiobj.get(export="csv", export_schema=True)

Export the output to a specific file descriptor and do not close the file descriptor when finished.

>>> fd = io.StringIO()
>>> assets = apiobj.get(export="csv", export_fd=fd, export_fd_close=False)

Use ‘missing’ instead of None for rows that are missing a column.

>>> assets = apiobj.get(export="csv", export_file="test.csv", csv_key_miss='missing')

Throw an error if an unknown column is found in a row. Must be one of: ‘ignore’ or ‘raise’.

>>> assets = apiobj.get(export="csv", export_file="test.csv", csv_key_extras='raise')

Use ‘excel-tab’ format instead of ‘excel’. Must be one of ‘excel’, ‘excel-tab’, or ‘unix’.

>>> assets = apiobj.get(export="csv", export_file="test.csv", csv_dialect='excel-tab')

Quote all items instead of just non-numeric. Must be one of ‘all’, ‘minimal’, ‘nonnumeric’, or ‘none’.

>>> assets = apiobj.get(export="csv", export_file="test.csv", csv_quoting='all')

See also

  • args_map() for callback generic arguments to format assets.

Notes

This callbacks object forces the following arguments to True in order to make the output usable in the exported format: field_null, field_flatten, field_join, export_schema

These arguments can be supplied as extra kwargs passed to axonius_api_client.api.assets.users.Users.get() or axonius_api_client.api.assets.devices.Devices.get()

_init(**kwargs)[source]

Override arguments to make export readable.

start(**kwargs)[source]

Start this callbacks object.

do_start(**kwargs)[source]

Create the CSV writer and write the columns.

stop(**kwargs)[source]

Stop this callbacks object.

do_stop(**kwargs)[source]

Close the file descriptor.

write_rows(rows)[source]

Write rows to the file descriptor.

Parameters:

rows (typing.Union[typing.List[dict], dict]) – rows to process

process_row(row)[source]

Process the callbacks for current row.

Parameters:

row (typing.Union[typing.List[dict], dict]) – row to process

Return type:

typing.List[dict]

do_export_schema()[source]

Add schema rows to the output.

ALL_SCHEMAS: dict = None

Map of adapter -> field schemas.

APIOBJ = None

assets object.

Type:

axonius_api_client.api.assets.asset_mixin.AssetMixin

CB_NAME: str = 'csv'

name for this callback

CURRENT_ROWS: t.Optional[t.List[dict]] = None

current rows being processed

CUSTOM_CB_EXC: List[dict] = None

tracker of custom callbacks that have been executed by do_custom_cbs()

FIND_KEYS: List[str] = ['name', 'name_qual', 'column_title']

field schema keys to use when finding a fields schema

GETARGS: dict = None

original kwargs supplied to get assets method.

STATE: dict = None

state dict used by get assets method to track paging.

STORE: dict = None

store dict used by get assets method to track arguments.

TAG_ROWS_ADD: List[dict] = None

tracker of assets to add tags to in do_tagging().

TAG_ROWS_REMOVE: List[dict] = None

tracker of assets to remove tags from in do_tagging().

__init__(apiobj, store, state=None, getargs=None)

Callbacks base class for assets.

Parameters:
__repr__()

Show info for this object.

Return type:

str

__str__()

Show info for this object.

Return type:

str

_add_report_adapters_missing(row)

Process report: Missing adapters.

Parameters:

row (dict) – row being processed

_add_report_software_whitelist(row)

Process report: Software whitelist.

Parameters:

row (dict) – row being processed

_do_add_null_values(row, schema, key='name_qual')

Null out missing fields.

Parameters:
  • row (dict) – row being processed

  • schema (dict) – field schema to add null values for

  • key (str) – key of field schema to add null value for in row

_do_change_field_titles(row)

Asset callback to change qual name to title.

Parameters:

row (dict) – row being processed

_do_excludes(row)

Asset callback to remove fields from row.

Parameters:

row (dict) – row being processed

_do_explode_entities(row)

Explode a row into multiple rows for each asset entity.

Parameters:

row (dict) – row being processed

Return type:

typing.List[dict]

_do_explode_field(row)

Explode a field into multiple rows.

Parameters:

row (dict) – row being processed

Return type:

typing.List[dict]

_do_flatten_fields(row, schema)

Asset callback to flatten complex fields.

Parameters:
  • row (dict) – row being processed

  • schema (dict) – schema to use when flattening

_do_join_values(row)

Join values.

Parameters:

row (dict) – row being processed

_field_compress(key)

Parse fields into required format.

Parameters:

key (str) –

Return type:

str

_field_replace(key)

Parse fields into required format.

Parameters:

key (str) –

Return type:

str

property adapter_map: dict

Build a map of adapters that have connections.

add_include_dates(rows)

Process report: Add dates (history and current)

Parameters:

rows (typing.Union[typing.List[dict], dict]) – rows to process

Return type:

typing.List[dict]

add_report_adapters_missing(rows)

Process report: Missing adapters.

Parameters:

rows (typing.Union[typing.List[dict], dict]) – rows to process

Return type:

typing.List[dict]

add_report_software_whitelist(rows)

Process report: Software whitelist.

Parameters:

rows (typing.Union[typing.List[dict], dict]) – rows to process

Return type:

typing.List[dict]

property arg_export_backup: bool

Pass.

property arg_export_fd: Optional[IO]

Pass.

property arg_export_fd_close: bool

Pass.

property arg_export_file: Optional[PathLike]

Pass.

property arg_export_overwrite: bool

Pass.

property arg_export_path: Path

Pass.

classmethod args_map()

Get all the argument names and their defaults for this callbacks object. :rtype: dict

Examples

Create a client using axonius_api_client.connect.Connect and assume apiobj is either client.devices or client.users

>>> import axonius_api_client as axonapi
>>> connect_args: dict = axonapi.get_env_connect()
>>> client: axonapi.Connect = axonapi.Connect(**connect_args)
>>> apiobj: axonapi.api.assets.AssetMixin = client.devices
>>>       # or client.users or client.vulnerabilities

Flatten complex fields - Will take all sub-fields of complex fields and put them on the root level with their values index correlated to each other.

>>> assets = apiobj.get(fields=["network_interfaces"], field_flatten=True)

Explode a single field - will take that field and create new rows for list item.

>>> assets = apiobj.get(field_explode="hostname")

Exclude fields - Will remove fields from the final output.

>>> assets = apiobj.get(field_excludes=["internal_axon_id", "adapter_list_length"])

Use field titles - Will change internal field names to their titles.

>>> assets = apiobj.get(field_titles=True)

Join fields - Will join multi value fields with carriage returns.

>>> assets = apiobj.get(field_join=True)

Join fields with no trim and custom join value - Will join multi value fields with ;; and do not trim the joined value to excels maximum cell length.

>>> assets = apiobj.get(field_join=True, field_join_value=";;", field_join_trim=0)

Add fields as empty values for fields that did not return.

>>> assets = apiobj.get(field_null=True)

Add fields as empty values for fields that did not return with a custom null value.

>>> assets = apiobj.get(
...     field_null=True,
...     field_null_value="EMPTY",
...     field_null_value_complex="EMPTY LIST",
... )

Add and remove tags to all assets returned.

>>> assets = apiobj.get(tags_add=["tag1", "tag2"], tags_remove=["tag3", "tag4"])

Generate a report of adapters that are missing from each asset.

>>> assets = apiobj.get(report_adapters_missing=True)

Generate a report of installed software that does not match a list of regex for each asset.

>>> assets = apiobj.get(report_software_whitelist=["chrome", "^adobe.*acrobat"])

Echo to STDERR progress messages.

>>> assets = apiobj.get(do_echo=True)

Change the amount of assets that echo page progress when do_echo is true.

>>> assets = apiobj.get(do_echo=True, page_progress=100)

Supply a set of custom callbacks to process each row before all builtin callbacks are run. Custom callbacks receive two arguments: self (the current callback object) and rows (the current rows being processed). Custom callbacks must return a list of rows.

>>> def custom_cb1(self, rows):
...     for row in rows:
...         row["internal_axon_id"] = row["internal_axon_id"].upper()
...     return rows
...
>>> assets = apiobj.get(custom_cbs=[custom_cb1])

See also

Notes

These arguments can be supplied as extra kwargs passed to axonius_api_client.api.assets.users.Users.get() or axonius_api_client.api.assets.devices.Devices.get()

classmethod args_map_base()

Get the map of arguments that can be supplied to GETARGS.

Return type:

dict

classmethod args_map_export()

Get the export argument names and their defaults for this callbacks object. :rtype: dict

See also

args_map_custom() for the arguments specific to this callback object.

args_map() for the arguments for all callback objects.

property args_strs: List[str]

Get a list of strings that describe each arg in args_map().

property callbacks: list

Get order of callbacks to run.

close_fd()

Close a file descriptor.

property custom_schemas: List[dict]

Get the custom schemas based on GETARGS.

do_add_null_values(rows)

Null out missing fields.

Parameters:

rows (typing.Union[typing.List[dict], dict]) – rows to process

Return type:

typing.List[dict]

do_change_field_compress(rows)

Asset callback to shorten field names.

Parameters:

rows (typing.Union[typing.List[dict], dict]) – rows to process

Return type:

typing.List[dict]

do_change_field_replace(rows)

Asset callback to replace characters.

Parameters:

rows (typing.Union[typing.List[dict], dict]) – rows to process

Return type:

typing.List[dict]

do_change_field_titles(rows)

Asset callback to change qual name to title.

Parameters:

rows (typing.Union[typing.List[dict], dict]) – rows to process

Return type:

typing.List[dict]

do_custom_cbs(rows)

Execute any custom callbacks for current row.

Parameters:

rows (typing.Union[typing.List[dict], dict]) – rows to process

Return type:

typing.List[dict]

do_excludes(rows)

Asset callback to remove fields from row.

Parameters:

rows (typing.Union[typing.List[dict], dict]) – rows to process

Return type:

typing.List[dict]

do_explode_entities(rows)

Explode a row into a row for each asset entity.

Parameters:

rows (typing.Union[typing.List[dict], dict]) – rows being processed

Return type:

typing.List[dict]

do_explode_field(rows)

Explode a field into multiple rows.

Parameters:

rows (typing.Union[typing.List[dict], dict]) – rows being processed

Return type:

typing.List[dict]

do_flatten_fields(rows)

Asset callback to flatten complex fields.

Parameters:

rows (typing.Union[typing.List[dict], dict]) – rows to process

Return type:

typing.List[dict]

do_join_values(rows)

Join values.

Parameters:

rows (typing.Union[typing.List[dict], dict]) – rows to process

Return type:

typing.List[dict]

do_pre_row(rows)

Pre-processing callbacks for current row.

Parameters:

rows (typing.Union[typing.List[dict], dict]) – rows to process

Return type:

typing.List[dict]

do_row(rows)

Execute the callbacks for current row.

Parameters:

rows (typing.Union[typing.List[dict], dict]) – rows to process

Return type:

typing.List[dict]

do_tag_add()

Add tags to assets.

do_tag_remove()

Remove tags from assets.

do_tagging()

Add or remove tags to assets.

echo(msg, debug=False, error=False, warning=False, level='info', level_debug='debug', level_error='error', level_warning='warning', abort=True)

Echo a message to console or log it.

Parameters:
  • msg (typing.Union[str, typing.List[str]]) – message to echo

  • error (typing.Union[bool, str, typing.Type[Exception]]) – message is an error

  • warning (bool) – message is a warning

  • level (str) – logging level for non error/non warning messages

  • level_debug (str) – logging level for debug messages

  • level_error (str) – logging level for error messages

  • level_warning (str) – logging level for warning messages

  • abort (bool) – sys.exit(1) if error is true

  • debug (bool) – message is a debug message

echo_columns(**kwargs)

Echo the columns of the fields selected.

echo_page_progress()

Echo progress per N rows using an echo method.

property excluded_schemas: List[dict]

List of all schemas that should be excluded.

property export_full_path: Path

Pass.

property export_templates: dict

Pass.

property field_replacements: List[Tuple[str, str]]

Parse the supplied list of field name replacements.

property fields_selected: List[str]

Get the names of the fields that were selected.

property final_columns: List[str]

Get the columns that will be returned.

property final_schemas: List[dict]

Get the schemas that will be returned.

get_arg_value(arg)

Get an argument value.

Parameters:

arg (str) – key to get from GETARGS with a default value from args_map()

Return type:

typing.Any

get_sub_schemas(schema)

Get all the schemas of sub-fields for a complex field.

Parameters:

schema (dict) – schema of complex field

Return type:

typing.Generator[dict, None, None]

is_excluded(schema)

Check if a name supplied to field_excludes matches one of FIND_KEYS.

Parameters:

schema (dict) – field schema

Return type:

bool

open_fd()

Open a file descriptor.

Return type:

typing.IO

open_fd_arg()

Open a file descriptor supplied in GETARGS.

Return type:

typing.IO

open_fd_path()

Open a file descriptor for a path.

Return type:

typing.IO

open_fd_stdout()

Open a file descriptor to STDOUT.

Return type:

typing.IO

process_tags_to_add(rows)

Add assets to tracker for adding tags.

Parameters:

rows (typing.Union[typing.List[dict], dict]) – rows to process

Return type:

typing.List[dict]

process_tags_to_remove(rows)

Add assets to tracker for removing tags.

Parameters:

rows (typing.Union[typing.List[dict], dict]) – rows to process

Return type:

typing.List[dict]

property schema_to_explode: dict

Get the schema of the field that should be exploded.

property schemas_selected: List[dict]

Get the schemas of the fields that were selected.

set_arg_value(arg, value)

Set an argument value.

Parameters:
LOG: logging.Logger

logger for this object.