Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __init__(self, username: str, password: str, databaseName=Settings.databaseName) -> None:
self.username = username
self.password = password
self.databaseName = databaseName
self.roles = []
ErrAtlasServerErrors
"""
if c in [Settings.SUCCESS, Settings.CREATED, Settings.ACCEPTED, Settings.NO_CONTENT]:
return details
elif c == Settings.BAD_REQUEST:
raise ErrAtlasBadRequest(c, details)
elif c == Settings.UNAUTHORIZED:
raise ErrAtlasUnauthorized(c, details)
elif c == Settings.FORBIDDEN:
raise ErrAtlasForbidden(c, details)
elif c == Settings.NOTFOUND:
raise ErrAtlasNotFound(c, details)
elif c == Settings.METHOD_NOT_ALLOWED:
raise ErrAtlasMethodNotAllowed(c, details)
elif c == Settings.CONFLICT:
raise ErrAtlasConflict(c, details)
else:
# Settings.SERVER_ERRORS
raise ErrAtlasServerErrors(c, details)
def _get_all_hosts(self, pageNum=Settings.pageNum,
itemsPerPage=Settings.itemsPerPage,
iterable=False):
"""Get All Hosts (actually processes)
Internal use only, actual data retrieval comes from properties host_list and host_names
url: https://docs.atlas.mongodb.com/reference/api/alerts-get-all-alerts/
Keyword Args:
pageNum (int): Page number
itemsPerPage (int): Number of Users per Page
iterable (bool): To return an iterable high level object instead of a low level API response
Returns:
ListOfHosts or dict: Iterable object representing this function OR Response payload
Raises:
def create_cluster(self, cluster: ClusterConfig) -> dict:
"""Create a cluster
url: POST /api/atlas/v1.0/groups/{GROUP-ID}/clusters
Args:
cluster (ClusterConfig): A Cluster Config Object
Returns:
dict: Response payload
"""
uri = Settings.api_resources["Clusters"]["Create a Cluster"].format(GROUP_ID=self.atlas.group)
logger.info("Initiating Call to Atlas API at {}".format(Settings.BASE_URL + uri))
logger.info("Cluster Config = {}".format(cluster))
cluster_data = self.atlas.network.post(Settings.BASE_URL + uri, payload=cluster.as_create_dict())
return cluster_data
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:
host_obj (Host): the host
def get_single_cluster(self, cluster: str) -> dict:
"""Get a Single Cluster
url: https://docs.atlas.mongodb.com/reference/api/clusters-get-one/
Args:
cluster (str): The cluster name
Returns:
dict: Response payload
"""
uri = Settings.api_resources["Clusters"]["Get a Single Cluster"] % (self.atlas.group, cluster)
cluster_data = self.atlas.network.get(Settings.BASE_URL + uri)
return cluster_data
def get_all_clusters(self, pageNum=Settings.pageNum, itemsPerPage=Settings.itemsPerPage, iterable=False):
"""Get All Clusters
url: https://docs.atlas.mongodb.com/reference/api/clusters-get-all/
Keyword Args:
pageNum (int): Page number
itemsPerPage (int): Number of Users per Page
iterable (bool): To return an iterable high level object instead of a low level API response
Returns:
AtlasPagination or dict: Iterable object representing this function OR Response payload
Raises:
ErrPaginationLimits: Out of limits
"""
def get_single_cluster(self, cluster: str) -> dict:
"""Get a Single Cluster
url: https://docs.atlas.mongodb.com/reference/api/clusters-get-one/
Args:
cluster (str): The cluster name
Returns:
dict: Response payload
"""
uri = Settings.api_resources["Clusters"]["Get a Single Cluster"] % (self.atlas.group, cluster)
cluster_data = self.atlas.network.get(Settings.BASE_URL + uri)
return cluster_data