Source code for axonius_api_client.api.adapters.adapters

# -*- coding: utf-8 -*-
"""API models for working with adapters and connections."""
from ...constants import DEFAULT_NODE
from ...exceptions import ApiError, NotFoundError
from ...tools import path_read
from ..mixins import ModelMixins
from ..parsers.adapters import parse_adapters, parse_schema
from ..parsers.config import config_build, config_unchanged, config_unknown
from ..parsers.tables import tablize_adapters
from ..routers import API_VERSION
from .cnx import Cnx


[docs]class Adapters(ModelMixins): """API model for working with adapters.""" @property def router(self): """Router for this API model.""" return API_VERSION.adapters
[docs] def get(self): """Pass.""" parsed = parse_adapters(raw=self._get()) parsed = sorted(parsed, key=lambda x: [x["node_name"], x["name"]]) return parsed
[docs] def get_by_name(self, name, node=DEFAULT_NODE): """Pass.""" adapters = self.get() keys = ["node_name", "node_id"] nodes = [x for x in adapters if node.lower() in [x[k].lower() for k in keys]] if not nodes: err = f"No node named {node!r} found" raise NotFoundError(tablize_adapters(adapters=adapters, err=err)) keys = ["name", "name_raw", "name_plugin"] for adapter in nodes: if any([adapter[k].lower() == name.lower() for k in keys]): return adapter err = f"No adapter named {name!r} found on node {node!r}" raise NotFoundError(tablize_adapters(adapters=adapters, err=err))
[docs] def config_get(self, name, node=DEFAULT_NODE, generic=True): """Pass.""" adapter = self.get_by_name(name=name, node=node) return self.config_refetch(adapter=adapter, generic=generic)
[docs] def config_refetch(self, adapter, generic=True): """Pass.""" name_config = "generic_name" if generic else "specific_name" name_config = adapter["schemas"][name_config] name_plugin = adapter["name_plugin"] if not generic and not name_config: name = adapter["name"] raise ApiError(f"Adapter {name} has no adapter specific advanced settings!") data = self._config_get(name_plugin=name_plugin, name_config=name_config) data["schema"] = parse_schema(raw=data["schema"]) return data
[docs] def config_update(self, name, node=DEFAULT_NODE, generic=True, **kwargs): """Pass.""" kwargs_config = kwargs.pop("kwargs_config", {}) kwargs.update(kwargs_config) adapter = self.get_by_name(name=name, node=node) name_config = "generic_name" if generic else "specific_name" name_config_src = "generic" if generic else "adapter specific" name_config = adapter["schemas"][name_config] config_map = self.config_refetch(adapter=adapter, generic=generic) old_config = config_map["config"] schemas = config_map["schema"] source = f"adapter {name!r} {name_config_src} advanced settings" config_unknown(schemas=schemas, new_config=kwargs, source=source) new_config = config_build( schemas=schemas, old_config=old_config, new_config=kwargs, source=source ) config_unchanged( schemas=schemas, old_config=old_config, new_config=new_config, source=source ) self._config_update( name_raw=adapter["name_raw"], name_config=name_config, new_config=new_config, ) return self.config_refetch(adapter=adapter, generic=generic)
[docs] def file_upload( self, name, field_name, file_name, file_content, file_content_type=None, node=DEFAULT_NODE, ): """Upload a str as a file to a specific adapter on a specific node. Args: adapter (:obj:`str` or :obj:`dict`): * If :obj:`str`, the name of the adapter to get the metadata for * If :obj:`dict`, the metadata for a previously fetched adapter field (:obj:`str`): name of field to store data in name (:obj:`str`): filename to use when uploading file content (:obj:`str` or :obj:`bytes`): content of file to upload node (:obj:`str`, optional): default :data:`.DEFAULT_NODE` - node name running ``adapter`` content_type (:obj:`str`, optional): default ``None`` - mime type of **content** Returns: :obj:`dict`: dict that can be used in a connection parameter of type file that contains the keys: * uuid: UUID of uploaded file * filename: filename of uploaded file """ adapter = self.get_by_name(name=name, node=node) return self._file_upload( name_raw=adapter["name_raw"], node_id=adapter["node_id"], file_name=file_name, field_name=field_name, file_content=file_content, file_content_type=file_content_type, )
[docs] def file_upload_path(self, path, **kwargs): """Upload the contents of a file to a specific adapter on a specific node. Args: adapter (:obj:`str` or :obj:`dict`): * If :obj:`str`, the name of the adapter to get the metadata for * If :obj:`dict`, the metadata for a previously fetched adapter field (:obj:`str`): name of field to store data in name (:obj:`str`): filename to use when uploading file path (:obj:`str` or :obj:`pathlib.Path`): path to file containing contents to upload node (:obj:`str`, optional): default :data:`.DEFAULT_NODE` - node name running ``adapter`` content_type (:obj:`str`, optional): default ``None`` - mime type of **content** Returns: :obj:`dict`: dict that can be used in a connection parameter of type file that contains the keys: * uuid: UUID of uploaded file * filename: filename of uploaded file """ path, file_content = path_read(obj=path, binary=True, is_json=False) kwargs.setdefault("field_name", path.name) kwargs.setdefault("file_name", path.name) kwargs["file_content"] = file_content return self.file_upload(**kwargs)
[docs] def _init(self, auth, **kwargs): """Post init method for subclasses to use for extra setup. Args: auth (:obj:`.auth.Model`): object to use for auth and sending API requests """ self.cnx = Cnx(parent=self) super(Adapters, self)._init(auth=auth, **kwargs)
[docs] def _get(self): """Direct API method to get all adapters. Returns: :obj:`dict`: raw metadata for all adapters """ path = self.router.root return self.request(method="get", path=path)
[docs] def _config_update(self, name_raw, name_config, new_config): """Direct API method to set advanced settings for an adapter. Args: name_raw (:obj:`str`): unique plugin name of adapter to set advanced settings for. unique plugin name refers to the name of an adapter for a specific node name_config (:obj:`str`): name of advanced settings to set - either AdapterBase for generic advanced settings or AwsSettings for adapter specific settings, if applicable config (:obj:`dict`): the configuration values to set Returns: :obj:`str`: empty str """ path = self.router.config_set.format( adapter_name_raw=name_raw, adapter_config_name=name_config ) return self.request( method="post", path=path, json=new_config, error_json_invalid=False )
[docs] def _config_get(self, name_plugin, name_config): """Direct API method to set advanced settings for an adapter. Args: name_plugin (:obj:`str`): unique plugin name of adapter to set advanced settings for. unique plugin name refers to the name of an adapter for a specific node i.e. aws_adapter_0 name_config (:obj:`str`): name of advanced settings to set - either AdapterBase for generic advanced settings or AwsSettings for adapter specific settings, if applicable Returns: :obj:`dict`: current settings for name_config of name_plugin """ path = self.router.config_get.format( adapter_name_plugin=name_plugin, adapter_config_name=name_config ) return self.request(method="get", path=path)
[docs] def _file_upload( self, name_raw, node_id, field_name, file_name, file_content, file_content_type=None, file_headers=None, ): """Direct API method to upload a file to a specific adapter on a specifc node. Args: name_raw (:obj:`str`): name of the adapter to upload a file to node_id (:obj:`str`): ID of node running **adapter** field_name (:obj:`str`): name of setting this file is intended for file_name (:obj:`str`): filename to use when uploading file file_content (:obj:`str` or :obj:`bytes`): content to upload file_content_type (:obj:`str`, optional): default ``None`` - mime type of **content** file_headers (:obj:`dict`, optional): default ``None`` - headers to use for file **content** Returns: :obj:`dict`: dict that can be used in a connection parameter of type file that contains the keys: * uuid: UUID of uploaded file * filename: filename of uploaded file """ data = {"field_name": field_name} files = {"userfile": (file_name, file_content, file_content_type, file_headers)} path = self.router.file_upload.format( adapter_name_raw=name_raw, adapter_node_id=node_id ) ret = self.request(method="post", path=path, data=data, files=files) ret["filename"] = file_name return ret
# def _download_file(self, name, node_id, cnx_id, schema_key): # """Direct API method to download a file. # Notes: # WORK IN PROGRESS! # """ # data = {"uuid": cnx_id, "schema_key": schema_key} # path = self.router.download_file.format( # name=name, node_id=node_id # ) # ret = self.request(method="post", path=path, json=data, raw=True) # return ret