Source code for laceworksdk.api.v2.datasources

# -*- coding: utf-8 -*-
"""Lacework Datasources API wrapper."""

from laceworksdk.api.base_endpoint import BaseEndpoint


[docs] class DatasourcesAPI(BaseEndpoint): """A class used to represent the `Datasources API endpoint <https://docs.lacework.net/api/v2/docs/#tag/Datasources>`_ Get schema details for all datasources that you can query using LQL. """ _DEFAULT_DESCRIPTION = "No description available." def __init__(self, session): """Initializes the Datasources object. Args: session (HttpSession): An instance of the HttpSession class Returns: DatasourcesAPI: An instance of this class """ super().__init__(session, "Datasources")
[docs] def get(self): """A method to get Datasources. Returns: dict: All datasources """ response = self._session.get(self._build_url()) return response.json()
[docs] def get_datasource(self, datasource): """A method to get the schema for a particular datasource. Args: datasource (str): The name of the datasource schema get. Returns: dict: The datasource schema. """ return self._session.get(self._build_url(resource=datasource)).json()
[docs] def list_data_sources(self): """A method to list the datasources that are available. Returns: list of tuples: Each tuple has two entries, source name and description. """ response_json = self.get() return_sources = [] data_sources = response_json.get("data", []) for data_source in data_sources: description = data_source.get("description", self._DEFAULT_DESCRIPTION) if description == "None": description = self._DEFAULT_DESCRIPTION return_sources.append((data_source.get("name", "No name"), description)) return return_sources