Method find_all

src/wechaty/user/contact.py:222–267

Find all contacts based on query, which can be string, ContactQueryFilter, or callable function. Args: query: the query body to build filter Examples: >>> # 1. find contacts based query string, will match one of: contact_id, weixin, name and

(cls: Type[Contact],
                       query: Optional[Union[str, ContactQueryFilter, Callable[[Contact], bool]]] = None
                       )

    @classmethod
    async def find_all(cls: Type[Contact],
                       query: Optional[Union[str, ContactQueryFilter, Callable[[Contact], bool]]] = None
                       ) -> List[Contact]:
        """
        Find all contacts based on query, which can be string, ContactQueryFilter, or callable<filter> function.

        Args:
            query: the query body to build filter

        Examples:
            >>> # 1. find contacts based query string, will match one of: contact_id, weixin, name and alias
            >>> # what's more, contact_id and weixin will follow extract match, name and alias will follow fuzzy match
            >>> Contact.find_all('your-contact-id/weixin')   # find: <your-contact-id> == contact.contact_id
            >>> Contact.find_all('name/alias')     # find: <name> in contact.name

            >>> # 2. find contacts based ContactQueryFilter object, will match all fields
            >>> query = ContactQueryFilter(id='your-contact-id', weixin='weixin')     # find: <your-contact-id> == contact.contact_id
            >>> query = ContactQueryFilter(name='your-contact-name')      # find: <your-contact-name> in contact.name
            >>> Contact.find_all(query)

            >>> # 3. find contacts based on callable query function
            >>> def filter_contacts(contact: Contact) -> bool:
            >>>     if contact.contact_id == "your-contact-id":
            >>>         return True
            >>>     return False
            >>> Contact.find_all(filter_contacts)

        Returns:
            Contact: the contact object filtered by query
        """
        log.info('find_all() <%s, %s>', cls, query)

        # 1. load contacts with concurrent tasks
        contact_ids: List[str] = await cls.get_puppet().contact_list()

        contacts: List[Contact] = [cls.load(contact_id) for contact_id in contact_ids]
        tasks: List[Task] = [asyncio.create_task(contact.ready()) for contact in contacts]
        await gather_with_concurrency(PARALLEL_TASK_NUM, tasks)

        # 2. filter contacts
        if not query:
            return contacts

        contacts = cls._filter_contacts(contacts, query)

        return contacts

    async def ready(self, force_sync: bool = False) -> None:
        """