def _translate_filter()

in datastore/providers/azuresearch_datastore.py [0:0]

• 29 lines of code
• 11 McCabe index (conditional complexity)

    def _translate_filter(filter: DocumentMetadataFilter) -> str:
        """
        Translates a DocumentMetadataFilter into an Azure Search filter string
        """
        if filter is None:
            return None

        escape = lambda s: s.replace("'", "''")

        # regex to validate dates are in OData format
        date_re = re.compile(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z")

        filter_list = []
        if filter.document_id is not None:
            filter_list.append(
                f"{FIELDS_DOCUMENT_ID} eq '{escape(filter.document_id)}'"
            )
        if filter.source is not None:
            filter_list.append(f"{FIELDS_SOURCE} eq '{escape(filter.source)}'")
        if filter.source_id is not None:
            filter_list.append(f"{FIELDS_SOURCE_ID} eq '{escape(filter.source_id)}'")
        if filter.author is not None:
            filter_list.append(f"{FIELDS_AUTHOR} eq '{escape(filter.author)}'")
        if filter.start_date is not None:
            if not date_re.match(filter.start_date):
                raise ValueError(
                    f"start_date must be in OData format, got {filter.start_date}"
                )
            filter_list.append(f"{FIELDS_CREATED_AT} ge {filter.start_date}")
        if filter.end_date is not None:
            if not date_re.match(filter.end_date):
                raise ValueError(
                    f"end_date must be in OData format, got {filter.end_date}"
                )
            filter_list.append(f"{FIELDS_CREATED_AT} le {filter.end_date}")
        return " and ".join(filter_list) if len(filter_list) > 0 else None