| 34 | |
| 35 | |
| 36 | class InfobloxExtattrManager(object): |
| 37 | # Make database connections |
| 38 | MC = MongoConnector.MongoConnector() |
| 39 | APIH = APIHelper.APIHelper() |
| 40 | IH = InfobloxHelper.InfobloxHelper() |
| 41 | |
| 42 | iblox_extattr_collection = MC.get_infoblox_extattr_connection() |
| 43 | zone_queried = None |
| 44 | record_type = None |
| 45 | next_page_id = None |
| 46 | _logger = None |
| 47 | |
| 48 | def _log(self): |
| 49 | """ |
| 50 | Get the log |
| 51 | """ |
| 52 | return logging.getLogger(__name__) |
| 53 | |
| 54 | def __get_record_type_url(self): |
| 55 | """ |
| 56 | Returns the url to be queried at infoblox to return the extattr information. |
| 57 | Paging information is appended to the URL as per the condition satisfied. |
| 58 | :return: string: URL to be queried |
| 59 | """ |
| 60 | paging_info = self.IH.get_pagination_params(self.next_page_id) |
| 61 | |
| 62 | return_fields = "&_return_fields=extattrs,zone" |
| 63 | if self.record_type == "a": |
| 64 | return_fields += ",ipv4addr" |
| 65 | elif self.record_type == "aaaa": |
| 66 | return_fields += ",ipv6addr" |
| 67 | elif self.record_type == "zone": |
| 68 | return_fields = "&_return_fields=extattrs" |
| 69 | |
| 70 | url = self.IH.get_infoblox_base_url( |
| 71 | self.zone_queried, |
| 72 | self.record_type, |
| 73 | ).format( |
| 74 | return_fields=return_fields, |
| 75 | paging_info=paging_info, |
| 76 | ) |
| 77 | return url |
| 78 | |
| 79 | def __insert_extattrs(self, insert_object): |
| 80 | """ |
| 81 | Inserts/Updates the extattr information in the database. '_ref' uniquely identifies the |
| 82 | resource. |
| 83 | :param insert_object: Dictionary containing the details of the resource. |
| 84 | """ |
| 85 | if not insert_object["_ref"] in self.previous_records: |
| 86 | insert_object["created"] = datetime.now() |
| 87 | insert_object["updated"] = datetime.now() |
| 88 | self.MC.perform_insert(self.iblox_extattr_collection, insert_object) |
| 89 | else: |
| 90 | self.previous_records.remove(insert_object["_ref"]) |
| 91 | self.iblox_extattr_collection.update_one( |
| 92 | {"_ref": insert_object["_ref"]}, |
| 93 | { |
nothing calls this directly
no test coverage detected