def get_contacts()

in ees_microsoft_outlook/microsoft_outlook_contacts.py [0:0]
57 lines of code
8 McCabe index (conditional complexity)

    def get_contacts(self, ids_list_contacts, accounts, start_time, end_time):
        """This method is used to fetches contacts from the Microsoft Outlook
        :param ids_list_contacts: List of documents which is already fetched
        :param accounts: List of user accounts
        :param start_time: Start time for fetching the contacts
        :param end_time: End time for fetching the contacts
        Returns:
            documents: List of contact documents
        """
        documents = []
        start_time = convert_datetime_to_ews_format(start_time)
        end_time = convert_datetime_to_ews_format(end_time)
        contact_schema = get_schema_fields(
            constant.CONTACTS_OBJECT.lower(), self.config.get_value("objects")
        )
        for account in accounts:

            # Logic to set time zone according to user account
            self.time_zone = account.default_timezone

            try:
                # Logic to fetch contacts
                folder = account.root / "Top of Information Store" / "Contacts"
                for contact in (
                    folder.all()
                    .filter(
                        last_modified_time__gt=start_time,
                        last_modified_time__lt=end_time,
                    )
                    .only(
                        "email_addresses",
                        "phone_numbers",
                        "last_modified_time",
                        "display_name",
                        "company_name",
                        "birthday",
                    )
                ):
                    if isinstance(contact, exchangelib.items.contact.Contact):

                        # Logic to insert contact into global_keys object
                        insert_document_into_doc_id_storage(
                            ids_list_contacts,
                            contact.id,
                            "",
                            constant.CONTACTS_OBJECT.lower(),
                            self.config.get_value("connector_platform_type"),
                        )
                        contact_obj = (
                            self.convert_contacts_to_workplace_search_documents(contact)
                        )
                        contact_map = {}
                        contact_map["_allow_permissions"] = []
                        if self.config.get_value("enable_document_permission"):
                            contact_map["_allow_permissions"] = [
                                account.primary_smtp_address
                            ]
                        contact_map["type"] = constant.CONTACTS_OBJECT
                        for ws_field, ms_field in contact_schema.items():
                            contact_map[ws_field] = contact_obj[ms_field]
                        documents.append(contact_map)
            except requests.exceptions.RequestException as request_error:
                raise requests.exceptions.RequestException(
                    f"Error while fetching contacts data for {account.primary_smtp_address}. Error: {request_error}"
                )
            except Exception as exception:
                self.logger.info(
                    f"Error while fetching contacts data for {account.primary_smtp_address}. Error: {exception}"
                )
                pass
        return list(unique_everseen(documents))