Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def get_single_cluster_as_obj(self, cluster) -> Union[ClusterConfig, ShardedClusterConfig]:
"""Get a Single Cluster as data
url: https://docs.atlas.mongodb.com/reference/api/clusters-get-one/
Args:
cluster (str): The cluster name
Returns:
ClusterConfig: Response payload
"""
cluster_data = self.get_single_cluster(cluster=cluster)
try:
if cluster_data.get('clusterType', None) == 'SHARDED':
logger.info("Cluster Type is SHARDED, Returning a ShardedClusterConfig")
out_obj = ShardedClusterConfig.fill_from_dict(data_dict=cluster_data)
elif cluster_data.get('clusterType', None) == 'REPLICASET':
logger.info("Cluster Type is REPLICASET, Returning a ClusterConfig")
out_obj = ClusterConfig.fill_from_dict(data_dict=cluster_data)
else:
logger.info("Cluster Type is not recognized, Returning a REPLICASET")
out_obj = ClusterConfig.fill_from_dict(data_dict=cluster_data)
except Exception as e:
raise e
return out_obj