# -*- coding: utf-8 -*-
"""API for working with adapters."""
import pathlib
from typing import List, Optional, Union
from ...exceptions import ApiError, NotFoundError
from ...parsers.config import config_build, config_unchanged, config_unknown
from ...parsers.tables import tablize_adapters
from ...tools import path_read
from .. import json_api
from ..api_endpoints import ApiEndpoints
from ..mixins import ModelMixins
[docs]class Adapters(ModelMixins):
"""API model for working with adapters.
Examples:
Create a ``client`` using :obj:`axonius_api_client.connect.Connect`
* Get metadata of all adapters: :meth:`get`
* Get an adapter by name: :meth:`get_by_name`
* Get the advanced settings for an adapter: :meth:`config_get`
* Update the advanced settings for an adapter: :meth:`config_update`
* Upload a file to an adapter: :meth:`file_upload`
* Work with adapter connections :obj:`axonius_api_client.api.adapters.cnx.Cnx`
Notes:
All methods use the Core instance by default, but you can work with another instance by
passing the name of the instance to ``node``.
Supplying unknown keys/values for configurations will throw an error showing the
valid keys/values.
"""
[docs] def get(self, get_clients: bool = False) -> List[dict]:
"""Get all adapters on all nodes.
Examples:
Create a ``client`` using :obj:`axonius_api_client.connect.Connect`.
Get all adapters
>>> adapters = client.adapters.get()
Get details of each adapter
>>> for adapter in adapters:
... print(adapter["name"]) # name of adapter
... print(adapter["node_name"]) # name of node adapter is running on
"""
basic_data = self._get_basic()
return [
adapter_node.to_dict_old(basic_data=basic_data.adapters)
for adapter in self._get(get_clients=get_clients)
for adapter_node in adapter.adapter_nodes
]
[docs] def get_by_name_basic(self, value: str) -> dict:
"""Get an adapter by name."""
data = self._get_basic()
return data.find_by_name(value=value)
[docs] def get_by_name(self, name: str, node: Optional[str] = None, get_clients: bool = False) -> dict:
"""Get an adapter by name on a single node.
Examples:
Create a ``client`` using :obj:`axonius_api_client.connect.Connect`.
Get an adapter by name
>>> adapter = client.adapters.get_by_name(name="aws")
Get details of adapter
>>> adapter['status'] # overall adapter status
'success'
>>> adapter['cnx_count_total'] # total connection count
1
>>> adapter['cnx_count_broken'] # broken connection count
0
>>> adapter['cnx_count_working'] # working connection count
1
Get details of each connection of the adapter
>>> for cnx in adapter["cnx"]:
... print(cnx["working"]) # bool if connection is working or not
... print(cnx["error"]) # error from last fetch attempt
... print(cnx["config"]) # configuration of connection
... print(cnx["id"]) # ID of connection
... print(cnx["uuid"]) # UUID of connection
Args:
name: name of adapter to get
node: name of node to get adapter from
Raises:
:exc:`NotFoundError`: when no node found or when no adapter found on node
"""
node_meta = self.instances.get_by_name_id_core(value=node)
adapters = self.get(get_clients=get_clients)
node_name = node_meta["name"]
adapters = [adapter for adapter in adapters if adapter["node_name"] == node_name]
keys = ["name", "name_raw", "name_plugin"]
for adapter in adapters:
if any([adapter[k].lower() == name.lower() for k in keys]):
adapter["node_meta"] = node_meta
return adapter
err = f"No adapter named {name!r} found on instance {node_name!r}"
raise NotFoundError(tablize_adapters(adapters=adapters, err=err))
[docs] def config_get(
self, name: str, node: Optional[str] = None, config_type: str = "generic"
) -> dict:
"""Get the advanced settings for an adapter.
Examples:
Create a ``client`` using :obj:`axonius_api_client.connect.Connect`.
Get the generic advanced settings for an adapter
>>> config = client.adapters.config_get(name="aws")
Get the adapter specific advanced settings for an adapter
>>> config = client.adapters.config_get(name="aws", config_type="specific")
Get the discovery advanced settings for an adapter
>>> config = client.adapters.config_get(name="aws", config_type="discovery")
See the current values of a configuration
>>> import pprint
>>> pprint.pprint(config['config'])
{'connect_client_timeout': 300,
'fetching_timeout': 43200,
'last_fetched_threshold_hours': 48,
'last_seen_prioritized': False,
'last_seen_threshold_hours': 24,
'minimum_time_until_next_fetch': None,
'realtime_adapter': False,
'user_last_fetched_threshold_hours': 48,
'user_last_seen_threshold_hours': None}
Investigate the schema and current values of a configuration
>>> for setting, info in config['schema'].items():
... current_value = config['config'][setting]
... title = info['title']
... description = info.get('description')
... print(f"name of setting: {setting}")
... print(f" title of setting in GUI: {title}")
... print(f" description of setting: {description}")
... print(f" current value of setting: {current_value}")
Args:
name: name of adapter to get advanced settings of
node: name of node to get adapter from [NO LONGER USED]
config_type: One of generic, specific, or discovery
"""
adapter = self.get_by_name(name=name, node=node, get_clients=False)
adapters = self._config_get(adapter_name=adapter["name_raw"])
type_map = adapters.type_map
if config_type not in type_map:
valid = ", ".join(list(type_map))
raise ApiError(f"Adapter {name} has no config type {config_type!r}, valids: {valid}!")
adapter_config = type_map[config_type]
adapter_config["adapter"] = adapter
return adapter_config
[docs] def config_update(
self, name: str, node: Optional[str] = None, config_type: str = "generic", **kwargs
) -> dict:
"""Update the advanced settings for an adapter.
Examples:
Create a ``client`` using :obj:`axonius_api_client.connect.Connect`.
Update the generic advanced settings for the adapter
>>> updated_config = client.adapters.config_update(
... name="aws", last_seen_threshold_hours=24
... )
Update the adapter specific advanced settings
>>> updated_config = client.adapters.config_update(
... name="aws", config_type="specific", fetch_s3=True
... )
Update the discovery advanced settings
>>> # XXX currently broken!
Args:
name: name of adapter to update advanced settings of
node: name of node to get adapter from
config_type: One of generic, specific, or discovery
**kwargs: configuration to update advanced settings of config_type
"""
kwargs_config = kwargs.pop("kwargs_config", {})
kwargs.update(kwargs_config)
config_map = self.config_get(name=name, config_type=config_type)
adapter_meta = config_map["adapter"]
old_config = config_map["config"]
schemas = config_map["schema"]
source = f"adapter {name!r} {config_type} 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(
adapter_name=adapter_meta["name_raw"],
config_name=config_map["config_name"],
config=new_config,
)
return self.config_get(name=name, config_type=config_type)
[docs] def file_upload(
self,
name: str,
field_name: str,
file_name: str,
file_content: Union[str, bytes],
file_content_type: Optional[str] = None,
node: Optional[str] = None,
) -> dict:
"""Upload a file to a specific adapter on a specific node.
Examples:
Create a ``client`` using :obj:`axonius_api_client.connect.Connect`.
Upload content as a file for use in a connection later
>>> content = "content of file to upload"
>>> file_uuid = client.adapters.file_upload(
... name="aws",
... file_name="name_of_file",
... file_content=content,
... field_name="name_of_field",
... )
>>> file_uuid
{'uuid': '5f78b7dee33f0a113700a6fc', 'filename': 'name_of_file'}
Args:
name: name of adapter to upload file to
node: name of node to to upload file to
field_name: name of field (should match configuration schema key name)
file_name: name of file to upload
file_content: content of file to upload
file_content_type: mime type of file to upload
"""
adapter = self.get_by_name(name=name, node=node, get_clients=False)
return self._file_upload(
adapter_name=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: Union[str, pathlib.Path], **kwargs):
"""Upload the contents of a file to a specific adapter on a specific node.
Examples:
Create a ``client`` using :obj:`axonius_api_client.connect.Connect`.
Upload a file for use in a connection later
>>> file_uuid = client.adapters.file_upload_path(name="aws", path="test.csv")
>>> file_uuid
{'uuid': '5f78b674e33f0a113700a6fa', 'filename': 'test.csv'}
Args:
path: path to file containing contents to upload
**kwargs: passed to :meth:`file_upload`
"""
path, file_content = path_read(obj=path, binary=True, is_json=False)
if path.suffix == ".csv":
kwargs.setdefault("file_content_type", "text/csv")
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, **kwargs):
"""Post init method for subclasses to use for extra setup."""
from ..system.instances import Instances
from .cnx import Cnx
self.cnx: Cnx = Cnx(parent=self)
"""Work with adapter connections"""
self.instances: Instances = Instances(auth=self.auth)
"""Work with instances"""
[docs] def _get(
self, get_clients: bool = True, filter: Optional[str] = None
) -> List[json_api.adapters.Adapter]:
"""Private API method to get all adapters."""
api_endpoint = ApiEndpoints.adapters.get
request_obj = api_endpoint.load_request(get_clients=get_clients, filter=filter)
return api_endpoint.perform_request(http=self.auth.http, request_obj=request_obj)
[docs] def _config_update(self, adapter_name: str, config_name: str, config: dict) -> str:
"""Private API method to set advanced settings for an adapter.
Args:
adapter_name: raw name of the adapter i.e. ``aws_adapter``
config_name: name of advanced settings to set
* ``AdapterBase`` for generic advanced settings
* ``AwsSettings`` for adapter specific advanced settings (name changes per adapter)
* ``DiscoverySchema`` for discovery advanced settings
new_config: the advanced configuration key value pairs to set
"""
api_endpoint = ApiEndpoints.adapters.settings_update
request_obj = api_endpoint.load_request(
pluginId=adapter_name, configName=config_name, config=config
)
return api_endpoint.perform_request(
http=self.auth.http,
request_obj=request_obj,
adapter_name=adapter_name,
config_name=config_name,
)
[docs] def _get_basic(self) -> json_api.generic.Metadata:
"""Pass."""
api_endpoint = ApiEndpoints.adapters.get_basic
return api_endpoint.perform_request(http=self.auth.http)
[docs] def _config_get(self, adapter_name: str) -> json_api.adapters.AdapterSettings:
"""Private API method to set advanced settings for an adapter.
Args:
adapter_name: raw name of the adapter, i.e. 'aws_adapter'
"""
api_endpoint = ApiEndpoints.adapters.settings_get
return api_endpoint.perform_request(http=self.auth.http, adapter_name=adapter_name)
[docs] def _file_upload(
self,
adapter_name: str,
node_id: str,
field_name: str,
file_name: str,
file_content: Union[bytes, str],
file_content_type: Optional[str] = None,
file_headers: Optional[dict] = None,
) -> dict:
"""Private API method to upload a file to a specific adapter on a specifc node.
Args:
adapter_name: raw name of the adapter i.e. ``aws_adapter``
node_id: ID of node running adapter
field_name: name of field (should match configuration schema key name)
file_name: name of file to upload
file_content: content of file to upload
file_content_type: mime type of file to upload
file_headers: headers to use for file
"""
api_endpoint = ApiEndpoints.adapters.file_upload
data = {"field_name": field_name}
files = {"userfile": (file_name, file_content, file_content_type, file_headers)}
http_args = {"files": files, "data": data}
response = api_endpoint.perform_request(
http=self.auth.http, http_args=http_args, adapter_name=adapter_name, node_id=node_id
)
parsed = {"filename": file_name, "uuid": response["data"]["id"]}
return parsed
[docs] def _get_labels(self) -> json_api.adapters.CnxLabels:
"""Pass."""
api_endpoint = ApiEndpoints.adapters.labels_get
response = api_endpoint.perform_request(http=self.http)
return response