Source code for axonius_api_client.api.asset_callbacks.base_json

# -*- coding: utf-8 -*-
"""JSON export callbacks."""
import json
import textwrap
from typing import List, Union

from ...tools import listify
from .base import ExportMixins


[docs]class Json(ExportMixins): """Callbacks for formatting asset data and exporting it in JSON format. Examples: Create a ``client`` using :obj:`axonius_api_client.connect.Connect` and assume ``apiobj`` is either ``client.devices`` or ``client.users`` >>> apiobj = client.devices # or client.users * :meth:`args_map` for callback generic arguments to format assets. * :meth:`args_map_custom` for callback specific arguments to format and export data. """
[docs] @classmethod def args_map_custom(cls) -> dict: """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="json") Export the output to a file in the default path :attr:`axonius_api_client.setup_env.DEFAULT_PATH`. >>> assets = apiobj.get(export="json", export_file="test.json") Export the output to an absolute path file (ignoring ``export_path``) and overwrite the file if it exists. >>> assets = apiobj.get( ... export="json", ... export_file="/tmp/output.json", ... export_overwrite=True, ... ) Export the output to a file in a specific dir. >>> assets = apiobj.get(export="json", export_file="output.json", export_path="/tmp") Return each asset as one dictionary per line (with no indenting) instead of a list of dictionaries. >>> assets = apiobj.get(export="json", export_file="test.json", json_flat=True) See Also: * :meth:`args_map` for callback generic arguments to format assets. Notes: These arguments can be supplied as extra kwargs passed to :meth:`axonius_api_client.api.assets.users.Users.get` or :meth:`axonius_api_client.api.assets.devices.Devices.get` """ args = {} args.update(cls.args_map_export()) args.update({"json_flat": False}) return args
[docs] def start(self, **kwargs): """Start this callbacks object.""" super(Json, self).start(**kwargs) flat = self.get_arg_value("json_flat") self._first_row = True self.open_fd() begin = "" if flat else "[" self._fd.write(begin)
[docs] def stop(self, **kwargs): """Stop this callbacks object.""" super(Json, self).stop(**kwargs) flat = self.get_arg_value("json_flat") self.do_export_schema() end = "" if flat else "\n]" self._fd.write(end) self.close_fd()
[docs] def process_row(self, row: Union[List[dict], dict]) -> List[dict]: """Process the callbacks for current row. Args: row: row to process """ rows = listify(row) rows = self.do_pre_row(rows=rows) row_return = [{"internal_axon_id": row["internal_axon_id"]} for row in rows] rows = self.do_row(rows=rows) self.write_rows(rows=rows) del rows, row return row_return
[docs] def write_rows(self, rows: Union[List[dict], dict]): """Write rows to the file descriptor. Args: rows: rows to process """ rows = listify(rows) flat = self.get_arg_value("json_flat") indent = None if flat else 2 prefix = " " * indent if indent else "" for row in rows: if self._first_row: pre = "" if flat else "\n" else: pre = "\n" if flat else ",\n" self._first_row = False self._fd.write(pre) value = json.dumps(row, indent=indent) value = textwrap.indent(value, prefix=prefix) if indent else value self._fd.write(value) del value, row
[docs] def do_export_schema(self): """Add schema rows to the output.""" export_schema = self.get_arg_value("export_schema") if export_schema: row = {"schemas": self.final_schemas} self.write_rows(rows=row) del row
CB_NAME: str = "json" """name for this callback"""