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: axonius_api_client.api.asset_callbacks.base.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.

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()

Return type

dict

_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
APIOBJ = None
CB_NAME: str = 'csv'

name for this callback

CUSTOM_CB_EXC: List[dict] = None
FIND_KEYS: List[str] = ['name', 'name_qual', 'column_title', 'name_base']
GETARGS: dict = None
STATE: dict = None
STORE: dict = None
TAG_ROWS_ADD: List[dict] = None
TAG_ROWS_REMOVE: List[dict] = None
__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_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) –

_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

Build a map of adapters that have connections.

Return type

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]

classmethod args_map()

Get all of the argument names and their defaults for this callbacks object.

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

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 if 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()

Return type

dict

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.

See also

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

args_map() for the arguments for all callback objects.

Return type

dict

property args_strs

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

Return type

typing.List[str]

property callbacks

Get order of callbacks to run.

Return type

list

close_fd()

Close a file descriptor.

property custom_schemas

Get the custom schemas based on GETARGS.

Return type

typing.List[dict]

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_field(rows)

Explode a field into multiple rows.

Parameters
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, error=False, warning=False, level='info', level_error='error', level_warning='warning', abort=True)

Echo a message to console or log it.

Parameters
  • msg (str) – message to echo

  • error (bool) – message is an error

  • warning (bool) – message is a warning

  • level (str) – logging level for non error/non warning 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

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 of all schemas that should be excluded.

Return type

typing.List[dict]

property field_replacements

Parse the supplied list of field name replacements.

Return type

typing.List[typing.Tuple[str, str]]

property fields_selected

Get the names of the fields that were selected.

Return type

typing.List[str]

property final_columns

Get the columns that will be returned.

Return type

typing.List[str]

property final_schemas

Get the schemas that will be returned.

Return type

typing.List[dict]

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.Union[str, list, bool, int]

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.

open_fd_arg()

Open a file descriptor supplied in GETARGS.

open_fd_path()

Open a file descriptor for a path.

open_fd_stdout()

Open a file descriptor to STDOUT.

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

Get the schema of the field that should be exploded.

Return type

dict

property schemas_selected

Get the schemas of the fields that were selected.

Return type

typing.List[dict]

set_arg_value(arg, value)

Set an argument value.

Parameters
CURRENT_ROWS: None