Skip to content

CodelistFilter

Bases: Filter

CodelistFilter is a class designed to filter a CodeTable based on a specified codelist.

The filter automatically detects where codes are defined by checking the CodeTable's CODES_DEFINED_IN property. If CODES_DEFINED_IN is set to a mapper class name (e.g., "ConceptTable"), the filter will use autojoin to reach that table. If CODES_DEFINED_IN is None, codes are assumed to be in the current table.

Event dates defined on another table are handled separately via EVENT_DATE_DEFINED_IN (resolved by EventPhenotype / PhenexTable.resolve_event_date).

Attributes:

Name Type Description
codelist Codelist

The codelist used for filtering the CodeTable.
name str

The name of the filter. Defaults to the name of the codelist if not provided.
Source code in phenex/filters/codelist_filter.py 
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
class CodelistFilter(Filter):
    """
    CodelistFilter is a class designed to filter a CodeTable based on a specified codelist.

    The filter automatically detects where codes are defined by checking the CodeTable's
    CODES_DEFINED_IN property. If CODES_DEFINED_IN is set to a mapper class name
    (e.g., "ConceptTable"),
    the filter will use autojoin to reach that table. If CODES_DEFINED_IN is None, codes
    are assumed to be in the current table.

    Event dates defined on another table are handled separately via EVENT_DATE_DEFINED_IN
    (resolved by EventPhenotype / PhenexTable.resolve_event_date).

    Attributes:
        codelist (Codelist): The codelist used for filtering the CodeTable.
        name (str): The name of the filter. Defaults to the name of the codelist if not provided.
    """

    def __init__(self, codelist: Codelist, name=None):
        self.codelist = codelist
        self.name = name or self.codelist.name
        self.codelist_as_tuples = self._convert_codelist_to_tuples()
        super(CodelistFilter, self).__init__()

    def _convert_codelist_to_tuples(self) -> List[Tuple[str, str]]:
        if self.codelist is not None:
            if not isinstance(self.codelist, Codelist):
                raise ValueError("Codelist must be an instance of Codelist")
            return [
                (ct, c)
                for ct, codes in self.codelist.resolved_codelist.items()
                for c in codes
            ]
        return []

    def _filter(self, code_table: CodeTable) -> CodeTable:
        assert is_phenex_code_table(code_table)

        if self.codelist.fuzzy_match:
            return self._filter_fuzzy_codelist(code_table)
        else:
            return self._filter_literal_codelist(code_table)

    def _filter_fuzzy_codelist(self, code_table):
        filter_condition = False
        for code_type, codelist in self.codelist.resolved_codelist.items():
            codelist = [str(code) for code in codelist]
            if self.codelist.use_code_type:
                filter_condition = filter_condition | (
                    (code_table.CODE_TYPE == code_type)
                    & (code_table.CODE.cast("str").like(codelist))
                )
            else:
                filter_condition = filter_condition | code_table.CODE.cast("str").like(
                    codelist
                )

        # PhenexTable.filter() returns a wrapped PhenexTable
        return code_table.filter(filter_condition)

    def _filter_literal_codelist(self, code_table):
        # Generate the codelist table as an Ibis literal set
        codelist_df = pd.DataFrame(
            self.codelist_as_tuples, columns=["code_type", "code"]
        ).fillna("")
        codelist_table = ibis.memtable(codelist_df)

        # Create a join condition based on code and possibly code_type
        code_column = code_table.CODE
        if self.codelist.use_code_type:
            code_type_column = code_table.CODE_TYPE
            join_condition = (
                code_column.cast("str") == codelist_table.code.cast("str")
            ) & (code_type_column == codelist_table.code_type)
        else:
            join_condition = code_column.cast("str") == codelist_table.code.cast("str")

        # Perform join and wrap result in same table type as input
        filtered_table = code_table.inner_join(codelist_table, join_condition).select(
            code_table.columns
        )
        return type(code_table)(filtered_table)

    def autojoin_filter(
        self, table: CodeTable, tables: Optional[Dict[str, PhenexTable]] = None
    ) -> CodeTable:
        """
        Automatically joins the necessary tables and applies the codelist filter.

        The filter checks the table's CODES_DEFINED_IN property to determine where codes are located:
        - If CODES_DEFINED_IN is None: codes are in the current table, filter directly
        - If CODES_DEFINED_IN is set: codes are in another table, perform autojoin to reach it

        This design allows the table mapper to control where codes are located, keeping the
        phenotype and filter logic clean and unaware of database schema details.

        Parameters:
            table (CodeTable): The table containing events to be filtered.
            tables (dict): A dictionary of tables from the DomainsDictionary for joining.

        Returns:
            CodeTable: The filtered CodeTable with events matching the codelist.

        Examples:
            ```
            # Example 1: Codes in the table itself (traditional pattern)
            class ConditionOccurrenceTable(CodeTable):
                CODES_DEFINED_IN = None  # Codes are in this table
                DEFAULT_MAPPING = {
                    "CODE": "CONDITION_CONCEPT_ID",
                    "CODE_TYPE": "VOCABULARY_ID"
                }

            # Example 2: Codes in a separate concept table (autojoin pattern)
            class EventWithoutCodesTable(CodeTable):
                CODES_DEFINED_IN = "ConceptTable"
                JOIN_KEYS = {"EventMappingTable": ["EVENTMAPPINGID"]}
                PATHS = {"ConceptTable": ["EventMappingTable"]}

            # Usage is identical regardless of where codes are:
            codelist_filter = CodelistFilter(codelist)
            filtered = codelist_filter.autojoin_filter(table, tables)
            ```
        """
        # Check if table specifies where codes are defined
        codes_domain = getattr(table, "CODES_DEFINED_IN", None)

        # Check if CODE and CODE_TYPE columns exist in the current table
        has_code = "CODE" in table.columns
        has_code_type = (
            "CODE_TYPE" in table.columns if self.codelist.use_code_type else True
        )

        if not (has_code and has_code_type):
            # Need to join to get CODE/CODE_TYPE columns
            if codes_domain is None:
                raise ValueError(
                    f"CodeTable {table.__class__.__name__} does not have CODE/CODE_TYPE columns "
                    "and CODES_DEFINED_IN is not set. Either add CODE/CODE_TYPE columns to the "
                    "table mapping or set CODES_DEFINED_IN to specify which table contains the codes."
                )
            if tables is None:
                raise ValueError(
                    f"Table required for codelist filter ({codes_domain}) but 'tables' parameter is None. "
                    "Pass the domains dictionary via the 'tables' parameter."
                )

            # Find the target table and perform autojoin
            target_table = PhenexTable.find_table_in_domains(codes_domain, tables)
            original_columns = table.columns
            table = table.join(target_table, domains=tables)

            # Apply the codelist filter
            filtered_table = self._filter(table)

            # Restore original column structure (plus any columns needed from join)
            # This ensures we don't pollute the output with unnecessary joined columns
            columns_to_keep = list(set(original_columns) & set(filtered_table.columns))
            return type(table)(filtered_table.select(columns_to_keep))

        # If CODE/CODE_TYPE already exist, just apply the filter directly
        return self._filter(table)

autojoin_filter(table, tables=None)

Automatically joins the necessary tables and applies the codelist filter.

The filter checks the table's CODES_DEFINED_IN property to determine where codes are located: - If CODES_DEFINED_IN is None: codes are in the current table, filter directly - If CODES_DEFINED_IN is set: codes are in another table, perform autojoin to reach it

This design allows the table mapper to control where codes are located, keeping the phenotype and filter logic clean and unaware of database schema details.

Parameters:

Name Type Description Default
table CodeTable

The table containing events to be filtered.

 required
tables dict

A dictionary of tables from the DomainsDictionary for joining.

 None

Returns:

Name Type Description
CodeTable CodeTable

The filtered CodeTable with events matching the codelist.

Examples:

# Example 1: Codes in the table itself (traditional pattern)
class ConditionOccurrenceTable(CodeTable):
    CODES_DEFINED_IN = None  # Codes are in this table
    DEFAULT_MAPPING = {
        "CODE": "CONDITION_CONCEPT_ID",
        "CODE_TYPE": "VOCABULARY_ID"
    }

# Example 2: Codes in a separate concept table (autojoin pattern)
class EventWithoutCodesTable(CodeTable):
    CODES_DEFINED_IN = "ConceptTable"
    JOIN_KEYS = {"EventMappingTable": ["EVENTMAPPINGID"]}
    PATHS = {"ConceptTable": ["EventMappingTable"]}

# Usage is identical regardless of where codes are:
codelist_filter = CodelistFilter(codelist)
filtered = codelist_filter.autojoin_filter(table, tables)
Source code in phenex/filters/codelist_filter.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
def autojoin_filter(
    self, table: CodeTable, tables: Optional[Dict[str, PhenexTable]] = None
) -> CodeTable:
    """
    Automatically joins the necessary tables and applies the codelist filter.

    The filter checks the table's CODES_DEFINED_IN property to determine where codes are located:
    - If CODES_DEFINED_IN is None: codes are in the current table, filter directly
    - If CODES_DEFINED_IN is set: codes are in another table, perform autojoin to reach it

    This design allows the table mapper to control where codes are located, keeping the
    phenotype and filter logic clean and unaware of database schema details.

    Parameters:
        table (CodeTable): The table containing events to be filtered.
        tables (dict): A dictionary of tables from the DomainsDictionary for joining.

    Returns:
        CodeTable: The filtered CodeTable with events matching the codelist.

    Examples:
        ```
        # Example 1: Codes in the table itself (traditional pattern)
        class ConditionOccurrenceTable(CodeTable):
            CODES_DEFINED_IN = None  # Codes are in this table
            DEFAULT_MAPPING = {
                "CODE": "CONDITION_CONCEPT_ID",
                "CODE_TYPE": "VOCABULARY_ID"
            }

        # Example 2: Codes in a separate concept table (autojoin pattern)
        class EventWithoutCodesTable(CodeTable):
            CODES_DEFINED_IN = "ConceptTable"
            JOIN_KEYS = {"EventMappingTable": ["EVENTMAPPINGID"]}
            PATHS = {"ConceptTable": ["EventMappingTable"]}

        # Usage is identical regardless of where codes are:
        codelist_filter = CodelistFilter(codelist)
        filtered = codelist_filter.autojoin_filter(table, tables)
        ```
    """
    # Check if table specifies where codes are defined
    codes_domain = getattr(table, "CODES_DEFINED_IN", None)

    # Check if CODE and CODE_TYPE columns exist in the current table
    has_code = "CODE" in table.columns
    has_code_type = (
        "CODE_TYPE" in table.columns if self.codelist.use_code_type else True
    )

    if not (has_code and has_code_type):
        # Need to join to get CODE/CODE_TYPE columns
        if codes_domain is None:
            raise ValueError(
                f"CodeTable {table.__class__.__name__} does not have CODE/CODE_TYPE columns "
                "and CODES_DEFINED_IN is not set. Either add CODE/CODE_TYPE columns to the "
                "table mapping or set CODES_DEFINED_IN to specify which table contains the codes."
            )
        if tables is None:
            raise ValueError(
                f"Table required for codelist filter ({codes_domain}) but 'tables' parameter is None. "
                "Pass the domains dictionary via the 'tables' parameter."
            )

        # Find the target table and perform autojoin
        target_table = PhenexTable.find_table_in_domains(codes_domain, tables)
        original_columns = table.columns
        table = table.join(target_table, domains=tables)

        # Apply the codelist filter
        filtered_table = self._filter(table)

        # Restore original column structure (plus any columns needed from join)
        # This ensures we don't pollute the output with unnecessary joined columns
        columns_to_keep = list(set(original_columns) & set(filtered_table.columns))
        return type(table)(filtered_table.select(columns_to_keep))

    # If CODE/CODE_TYPE already exist, just apply the filter directly
    return self._filter(table)

filter(table)

Filters the given table according to the rules of the Filter.

Parameters:

Name Type Description Default
table PhenexTable

The table to be filtered.

 required

Returns:

Name Type Description
PhenexTable PhenexTable

The filtered table. The returned table has the exact same schema as the input table but has rows removed.
Source code in phenex/filters/filter.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def filter(self, table: PhenexTable) -> PhenexTable:
    """
    Filters the given table according to the rules of the Filter.

    Args:
        table (PhenexTable): The table to be filtered.

    Returns:
        PhenexTable: The filtered table. The returned table has the exact same schema as the input table but has rows removed.
    """
    input_columns = table.columns
    filtered_table = self._filter(table)
    if not set(input_columns) <= set(filtered_table.columns):
        raise ValueError(f"Filter must not remove columns.")

    filtered_table = filtered_table.select(input_columns)
    if isinstance(table, PhenexTable):
        return type(table)(filtered_table)
    else:
        return filtered_table