Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
USER = getenv('ATLAS_USER', 'JKKIDWUA')
API_KEY = getenv('ATLAS_KEY', '4e09ab9f-bf2c-41da-90bf-7d7974c330d2')
GROUP_ID = getenv('ATLAS_GROUP', '5b1e92c13b34b93b0230e6e1')
from atlasapi.lib import AtlasPeriods, AtlasUnits, AtlasGranularities
from atlasapi.measurements import AtlasMeasurementTypes
if not USER or not API_KEY or not GROUP_ID:
raise EnvironmentError('In order to run this smoke test you need ATLAS_USER, AND ATLAS_KEY env variables'
'your env variables are {}'.format(environ.__str__()))
a = Atlas(USER,API_KEY,GROUP_ID)
pprint('----------MeasureMents')
output = a.Hosts._get_measurement_for_host(a.Hosts.host_list[0]
,measurement=AtlasMeasurementTypes.Memory.resident,iterable=True
,period=AtlasPeriods.WEEKS_4,granularity=AtlasGranularities.MINUTE)
for each in a.Hosts.host_list:
pprint(each.__dict__)
# for cluster in cluster_list:
# print('Cluster name {}'.format(cluster))
pprint('------------Test get hosts by cluster-----------------')
# hosts = a.Hosts.host_list_by_cluster('monitoringtest')
print('-----------Test get metrics for a clusters hosts---------------')
a.Hosts.fill_host_list(for_cluster='monitoringtest')
#a.Hosts.get_measurement_for_hosts(measurement=AtlasMeasurementTypes.Network.bytes_out
# , period=AtlasPeriods.HOURS_1, granularity=AtlasGranularities.FIVE_MINUTE)
for hostObj in a.Hosts.host_list:
hostObj.get_measurement_for_host(measurement=AtlasMeasurementTypes.Network.bytes_out,
period=AtlasPeriods.HOURS_1, granularity=AtlasGranularities.FIVE_MINUTE)
the = list()
def _get_measurement_for_host(self, host_obj: Host, granularity: AtlasGranularities = AtlasGranularities.HOUR,
period: AtlasPeriods = AtlasPeriods.WEEKS_1,
measurement: AtlasMeasurementTypes = AtlasMeasurementTypes.Cache.dirty,
pageNum: int = Settings.pageNum,
itemsPerPage: int = Settings.itemsPerPage,
iterable: bool = True) -> Union[dict, Iterable[AtlasMeasurement]]:
"""Get measurement(s) for a host
Internal use only, should come from the host obj itself.
Returns measurements for the passed Host object.
url: https://docs.atlas.mongodb.com/reference/api/process-measurements/
Accepts either a single measurement, but will retrieve more than one measurement
if the measurement (using the AtlasMeasurementTypes class)
/api/atlas/v1.0/groups/{GROUP-ID}/processes/{HOST}:{PORT}/measurements
def get_measurement_for_hosts(self, granularity: AtlasGranularities = AtlasGranularities.HOUR,
period: AtlasPeriods = AtlasPeriods.WEEKS_1,
measurement=AtlasMeasurementTypes.Cache.dirty, return_data: bool = False):
"""Get measurement(s) for all hosts in the host_list
Populates all hosts in the host_list with the requested metric.
Multiple calls will append additional metrics to the same host object.
Please note that using the `return_data` param will also return the updated
host objects, which may unnecessarily consume memory.
Keyword Args:
granularity (AtlasGranularities): the desired granularity
period (AtlasPeriods): The desired period
measurement (AtlasMeasurementTypes) : The desired measurement or Measurement class
def get_measurement_for_host(self, granularity: AtlasGranularities = AtlasGranularities.HOUR,
period: AtlasPeriods = AtlasPeriods.WEEKS_1,
measurement: AtlasMeasurementTypes = AtlasMeasurementTypes.Cache.dirty,
pageNum: int = Settings.pageNum,
itemsPerPage: int = Settings.itemsPerPage,
iterable: bool = True) -> Union[dict, Iterable[AtlasMeasurement]]:
"""Get measurement(s) for a host
Returns measurements for the Host object.
url: https://docs.atlas.mongodb.com/reference/api/process-measurements/
Accepts either a single measurement, but will retrieve more than one measurement
if the measurement (using the AtlasMeasurementTypes class)
/api/atlas/v1.0/groups/{GROUP-ID}/processes/{HOST}:{PORT}/measurements
Keyword Args: