Database connections, tables and mappers

Tables

Phenex provides certain table types on which it knows how to operate. For instance, Phenex implements a CodeTable, which is an event table containing codes. Phenex has abstracted operations for each table type. For instance, given a CodeTable, Phenex knows how to filter this table based on the presence of codes within that table. Phenex doesn't care if the code table is actually a diagnosis code table or a procedure code table or a medication code table.

In onboarding a new data model to Phenex, the tables must be mapped into Phenex table types by subclassing the appropriate PhenexTable. When subclassing a PhenexTable, you must define:

1. COLUMN_MAPPING: a mapping of the input table columns to the fields on the chosen PhenexTable type (e.g. 'CD' maps to 'CODE' in a CodeTable).
2. JOIN_KEYS: if you want to use the autojoin functionality of PhenexTable, you must specify what keys to use to join pairs of tables
3. PATHS: if you want to use the autojoin functionality of PhenexTable for more complex joins, you must specify join paths to take to get from one table to another

Note that for each table type, there are REQUIRED_FIELDS, i.e., fields that MUST be defined for Phenex to work with such a table and KNOWN_FIELDS, i.e., fields that Phenex internally understands what to do with (there is a Phenotype that knows how to work with that field). For instance, in a PhenexPersonTable, one MUST define PERSON_ID, but DATE_OF_BIRTH is an optional field that PhenEx can process if given and transform into AGE. These are fixed for each table type and should not be overridden.

DEFAULT_MAPPING can map to either a single column (string) or multiple columns (list): - String value: Creates a column as a direct copy/rename Example: {"PERSON_ID": "PERSON_ID"} copies PERSON_ID column - List value: Creates a column as a coalesce of multiple columns (first non-null value) Example: {"EVENT_DATE": ["STARTDATETIME", "RECORDEDDATETIME"]} creates EVENT_DATE using STARTDATETIME if available, falling back to RECORDEDDATETIME if STARTDATETIME is null

DATE_FORMAT is a dictionary mapping source (original) column names to date format strings. When a source column appears in DATE_FORMAT, its string values are parsed into timestamps using the specified format before any further processing (coalescing, casting, etc.). This is useful when date columns are stored as strings in the source data.

IMPORTANT: The format string must use the syntax of your database backend, not Python strftime. Common formats by backend: - Snowflake: "YYYYMMDD", "YYYYMM", "YYYY", "YYYY-MM-DD" - DuckDB: "%Y%m%d", "%Y%m", "%Y", "%Y-%m-%d" - BigQuery: "%Y%m%d", "%Y%m", "%Y", "%Y-%m-%d"

Each DATE_FORMAT value can be either a plain format string or a two-element list [format, position]. The list form is meaningful for year-only (YYYY / %Y) and year-month (YYYYMM / %Y%m) formats, which do not encode a specific day. position resolves the ambiguity:

• 'first' — first day of the year (Jan 1) or month (the 1st)
• 'middle' — mid-year (Jul 2) or mid-month (the 15th)
• 'last' — last day of the year (Dec 31) or last day of the month

Examples:

class MyCodeTable(CodeTable):
    DATE_FORMAT = {"EVENTDATE": "YYYYMMDD"}  # parse "20240115" -> 2024-01-15
    DEFAULT_MAPPING = {
        "EVENT_DATE": "EVENTDATE",
    }

class MyCodeTableCoalesce(CodeTable):
    DATE_FORMAT = {
        "STARTDATE": "YYYYMMDD",       # parse "20240115" -> 2024-01-15
        "RECORDEDDATE": "DD/MM/YYYY",  # parse "15/01/2024" -> 2024-01-15
    }
    DEFAULT_MAPPING = {
        "EVENT_DATE": ["STARTDATE", "RECORDEDDATE"],  # coalesce after formatting
    }

class MyYearMonthTable(CodeTable):  # Snowflake, year-month columns
    DATE_FORMAT = {
        "STUDY_MONTH": ["YYYYMM", "first"],   # "202401" -> 2024-01-01
        "BIRTH_YEAR":  ["YYYY",   "middle"],  # "1990"   -> 1990-07-02
        "ENROL_MONTH": ["YYYYMM", "last"],    # "202401" -> 2024-01-31
    }
    DEFAULT_MAPPING = {
        "EVENT_DATE": "STUDY_MONTH",
    }

JOIN_KEYS and PATHS Documentation:

JOIN_KEYS defines direct relationships between tables. The key is the CLASS NAME of the target table, and the value is a list of join keys. Each join key can be: - A string: symmetric join (column has same name in both tables) - A 2-element tuple/list: asymmetric join (left_col, right_col) with different names

PATHS defines multi-hop join paths. The key is the CLASS NAME of the final target table, and the value is a list of CLASS NAMES for intermediate tables to traverse.

IMPORTANT: JOIN_KEYS should be defined symmetrically - if TableA can join to TableB, then TableB should also define how to join back to TableA.

Example 1: Symmetric joins (same column names)

class DummyConditionOccurrenceTable(CodeTable):
    NAME_TABLE = "DIAGNOSIS"
    JOIN_KEYS = {
        "DummyPersonTable": ["PERSON_ID"],  # Join using PERSON_ID in both tables
        "DummyEncounterTable": ["PERSON_ID", "ENCID"],  # Compound join: both keys must match
    }
    PATHS = {
        "DummyVisitDetailTable": ["DummyEncounterTable"]  # To reach VisitDetail, go through Encounter
    }

class DummyEncounterTable(PhenexTable):
    NAME_TABLE = "ENCOUNTER"
    JOIN_KEYS = {
        "DummyPersonTable": ["PERSON_ID"],
        "DummyConditionOccurrenceTable": ["PERSON_ID", "ENCID"],  # Symmetric!
        "DummyVisitDetailTable": ["PERSON_ID", "VISITID"],
    }

class DummyVisitDetailTable(PhenexTable):
    NAME_TABLE = "VISIT"
    JOIN_KEYS = {
        "DummyPersonTable": ["PERSON_ID"],
        "DummyEncounterTable": ["PERSON_ID", "VISITID"],  # Symmetric!
    }

Example 2: Asymmetric joins (different column names)

class EventTable(CodeTable):
    NAME_TABLE = "EVENT"
    JOIN_KEYS = {
        "EventMappingTable": [("ID", "EVENTID")],  # EventTable.ID joins to EventMappingTable.EVENTID
    }
    PATHS = {
        "ConceptTable": ["EventMappingTable"],
    }
    DEFAULT_MAPPING = {
        "PERSON_ID": "PERSON_ID",
        "ID": "ID",  # Must include ID in mapping for it to exist
    }

class EventMappingTable(PhenexTable):
    NAME_TABLE = "EVENT_MAPPING"
    JOIN_KEYS = {
        "EventTable": [("EVENTID", "ID")],  # Symmetric: reverse the tuple
        "ConceptTable": [("CONCEPTID", "ID")],  # Maps to ConceptTable.ID
    }
    DEFAULT_MAPPING = {
        "EVENTID": "EVENTID",
        "CONCEPTID": "CONCEPTID",
    }

class ConceptTable(CodeTable):
    NAME_TABLE = "CONCEPT"
    JOIN_KEYS = {
        "EventMappingTable": [("ID", "CONCEPTID")],  # Symmetric: reverse the tuple
    }
    DEFAULT_MAPPING = {
        "ID": "ID",
        "CODE": "CONCEPT_CODE",
        "CODE_TYPE": "VOCABULARY_ID",
    }

Example 3: Mixed symmetric and asymmetric joins

class PatientEventTable(CodeTable):
    JOIN_KEYS = {
        "EventMappingTable": [
            "PERSON_ID",  # Symmetric: PERSON_ID in both tables
            ("EVENT_ID", "EVENTID")  # Asymmetric: different column names
        ],
    }

In all examples: - Symmetric relationships use strings: ["COLUMN_NAME"] - Asymmetric relationships use tuples: [("LEFT_COL", "RIGHT_COL")] - Compound joins use multiple elements: ["COL1", "COL2"] or [("L1", "R1"), ("L2", "R2")] - All relationships should be symmetric (both tables define the join) - ALL join columns must be in DEFAULT_MAPPING for them to exist in the mapped table

Source code in phenex/tables.py

class PhenexTable:
    """
    Phenex provides certain table types on which it knows how to operate. For instance, Phenex implements a CodeTable, which is an event table containing codes. Phenex has abstracted operations for each table type. For instance, given a CodeTable, Phenex knows how to filter this table based on the presence of codes within that table. Phenex doesn't care if the code table is actually a diagnosis code table or a procedure code table or a medication code table.

    In onboarding a new data model to Phenex, the tables must be mapped into Phenex table types by subclassing the appropriate PhenexTable. When subclassing a PhenexTable, you must define:

        1. COLUMN_MAPPING: a mapping of the input table columns to the fields on the chosen PhenexTable type (e.g. 'CD' maps to 'CODE' in a CodeTable).
        2. JOIN_KEYS: if you want to use the autojoin functionality of PhenexTable, you must specify what keys to use to join pairs of tables
        3. PATHS: if you want to use the autojoin functionality of PhenexTable for more complex joins, you must specify join paths to take to get from one table to another

    Note that for each table type, there are REQUIRED_FIELDS, i.e., fields that MUST be defined for Phenex to work with such a table and KNOWN_FIELDS, i.e., fields that Phenex internally understands what to do with (there is a Phenotype that knows how to work with that field). For instance, in a PhenexPersonTable, one MUST define PERSON_ID, but DATE_OF_BIRTH is an optional field that PhenEx can process if given and transform into AGE. These are fixed for each table type and should not be overridden.

    DEFAULT_MAPPING can map to either a single column (string) or multiple columns (list):
    - String value: Creates a column as a direct copy/rename
      Example: {"PERSON_ID": "PERSON_ID"} copies PERSON_ID column
    - List value: Creates a column as a coalesce of multiple columns (first non-null value)
      Example: {"EVENT_DATE": ["STARTDATETIME", "RECORDEDDATETIME"]} creates EVENT_DATE using
      STARTDATETIME if available, falling back to RECORDEDDATETIME if STARTDATETIME is null

    DATE_FORMAT is a dictionary mapping source (original) column names to date format strings.
    When a source column appears in DATE_FORMAT, its string values are parsed into timestamps
    using the specified format before any further processing (coalescing, casting, etc.).
    This is useful when date columns are stored as strings in the source data.

    IMPORTANT: The format string must use the syntax of your database backend, not Python strftime.
    Common formats by backend:
    - Snowflake: "YYYYMMDD", "YYYYMM", "YYYY", "YYYY-MM-DD"
    - DuckDB:    "%Y%m%d",  "%Y%m",  "%Y",  "%Y-%m-%d"
    - BigQuery:  "%Y%m%d",  "%Y%m",  "%Y",  "%Y-%m-%d"

    Each DATE_FORMAT value can be either a plain format string or a two-element list
    [format, position]. The list form is meaningful for year-only (YYYY / %Y) and
    year-month (YYYYMM / %Y%m) formats, which do not encode a specific day. position
    resolves the ambiguity:

    - 'first'  — first day of the year (Jan 1) or month (the 1st)
    - 'middle' — mid-year (Jul 2) or mid-month (the 15th)
    - 'last'   — last day of the year (Dec 31) or last day of the month

    Examples:
    ```python
    class MyCodeTable(CodeTable):
        DATE_FORMAT = {"EVENTDATE": "YYYYMMDD"}  # parse "20240115" -> 2024-01-15
        DEFAULT_MAPPING = {
            "EVENT_DATE": "EVENTDATE",
        }

    class MyCodeTableCoalesce(CodeTable):
        DATE_FORMAT = {
            "STARTDATE": "YYYYMMDD",       # parse "20240115" -> 2024-01-15
            "RECORDEDDATE": "DD/MM/YYYY",  # parse "15/01/2024" -> 2024-01-15
        }
        DEFAULT_MAPPING = {
            "EVENT_DATE": ["STARTDATE", "RECORDEDDATE"],  # coalesce after formatting
        }

    class MyYearMonthTable(CodeTable):  # Snowflake, year-month columns
        DATE_FORMAT = {
            "STUDY_MONTH": ["YYYYMM", "first"],   # "202401" -> 2024-01-01
            "BIRTH_YEAR":  ["YYYY",   "middle"],  # "1990"   -> 1990-07-02
            "ENROL_MONTH": ["YYYYMM", "last"],    # "202401" -> 2024-01-31
        }
        DEFAULT_MAPPING = {
            "EVENT_DATE": "STUDY_MONTH",
        }
    ```

    JOIN_KEYS and PATHS Documentation:

    JOIN_KEYS defines direct relationships between tables. The key is the CLASS NAME of the target table,
    and the value is a list of join keys. Each join key can be:
    - A string: symmetric join (column has same name in both tables)
    - A 2-element tuple/list: asymmetric join (left_col, right_col) with different names

    PATHS defines multi-hop join paths. The key is the CLASS NAME of the final target table,
    and the value is a list of CLASS NAMES for intermediate tables to traverse.

    IMPORTANT: JOIN_KEYS should be defined symmetrically - if TableA can join to TableB,
    then TableB should also define how to join back to TableA.

    Example 1: Symmetric joins (same column names)
    ```python
    class DummyConditionOccurrenceTable(CodeTable):
        NAME_TABLE = "DIAGNOSIS"
        JOIN_KEYS = {
            "DummyPersonTable": ["PERSON_ID"],  # Join using PERSON_ID in both tables
            "DummyEncounterTable": ["PERSON_ID", "ENCID"],  # Compound join: both keys must match
        }
        PATHS = {
            "DummyVisitDetailTable": ["DummyEncounterTable"]  # To reach VisitDetail, go through Encounter
        }

    class DummyEncounterTable(PhenexTable):
        NAME_TABLE = "ENCOUNTER"
        JOIN_KEYS = {
            "DummyPersonTable": ["PERSON_ID"],
            "DummyConditionOccurrenceTable": ["PERSON_ID", "ENCID"],  # Symmetric!
            "DummyVisitDetailTable": ["PERSON_ID", "VISITID"],
        }

    class DummyVisitDetailTable(PhenexTable):
        NAME_TABLE = "VISIT"
        JOIN_KEYS = {
            "DummyPersonTable": ["PERSON_ID"],
            "DummyEncounterTable": ["PERSON_ID", "VISITID"],  # Symmetric!
        }
    ```

    Example 2: Asymmetric joins (different column names)
    ```python
    class EventTable(CodeTable):
        NAME_TABLE = "EVENT"
        JOIN_KEYS = {
            "EventMappingTable": [("ID", "EVENTID")],  # EventTable.ID joins to EventMappingTable.EVENTID
        }
        PATHS = {
            "ConceptTable": ["EventMappingTable"],
        }
        DEFAULT_MAPPING = {
            "PERSON_ID": "PERSON_ID",
            "ID": "ID",  # Must include ID in mapping for it to exist
        }

    class EventMappingTable(PhenexTable):
        NAME_TABLE = "EVENT_MAPPING"
        JOIN_KEYS = {
            "EventTable": [("EVENTID", "ID")],  # Symmetric: reverse the tuple
            "ConceptTable": [("CONCEPTID", "ID")],  # Maps to ConceptTable.ID
        }
        DEFAULT_MAPPING = {
            "EVENTID": "EVENTID",
            "CONCEPTID": "CONCEPTID",
        }

    class ConceptTable(CodeTable):
        NAME_TABLE = "CONCEPT"
        JOIN_KEYS = {
            "EventMappingTable": [("ID", "CONCEPTID")],  # Symmetric: reverse the tuple
        }
        DEFAULT_MAPPING = {
            "ID": "ID",
            "CODE": "CONCEPT_CODE",
            "CODE_TYPE": "VOCABULARY_ID",
        }
    ```

    Example 3: Mixed symmetric and asymmetric joins
    ```python
    class PatientEventTable(CodeTable):
        JOIN_KEYS = {
            "EventMappingTable": [
                "PERSON_ID",  # Symmetric: PERSON_ID in both tables
                ("EVENT_ID", "EVENTID")  # Asymmetric: different column names
            ],
        }
    ```

    In all examples:
    - Symmetric relationships use strings: ["COLUMN_NAME"]
    - Asymmetric relationships use tuples: [("LEFT_COL", "RIGHT_COL")]
    - Compound joins use multiple elements: ["COL1", "COL2"] or [("L1", "R1"), ("L2", "R2")]
    - All relationships should be symmetric (both tables define the join)
    - ALL join columns must be in DEFAULT_MAPPING for them to exist in the mapped table
    """

    NAME_TABLE = "PHENEX_TABLE"  # name of table in the database
    JOIN_KEYS = {}  # dict: class name -> List[phenex column names]
    KNOWN_FIELDS = []  # List[phenex column names]
    DEFAULT_MAPPING = {}  # dict: input column name -> phenex column name
    PATHS = {}  # dict: table class name -> List[other table class names]
    DATE_FORMAT = {}  # dict: source column name -> backend-native date format string
    REQUIRED_FIELDS = list(DEFAULT_MAPPING.keys())

    def __init__(self, table, name=None, column_mapping={}):
        """
        Instantiate a PhenexTable, possibly overriding NAME_TABLE and COLUMN_MAPPING.
        """

        if not isinstance(table, Table):
            raise TypeError(
                f"Cannot instantiatiate {self.__class__.__name__} from {type(table)}. Must be ibis Table."
            )

        self.NAME_TABLE = name or self.NAME_TABLE

        self.column_mapping = self._get_column_mapping(column_mapping)
        self._table = table.mutate(
            **self._resolve_column_mapping(table, self.column_mapping)
        )

        for key in self.REQUIRED_FIELDS:
            try:
                getattr(self._table, key)
            except AttributeError:
                raise ValueError(f"Required field {key} not defined in COLUMN_MAPPING.")

        self._add_phenotype_table_relationship()

    def _add_phenotype_table_relationship(self):
        self.JOIN_KEYS["PhenotypeTable"] = ["PERSON_ID"]

    def _get_column_mapping(self, column_mapping=None):
        column_mapping = column_mapping or {}
        # Only validate fields explicitly passed in column_mapping parameter
        # DEFAULT_MAPPING is defined by the class itself and should be trusted
        # This allows join keys and other auxiliary fields to be in DEFAULT_MAPPING
        # without requiring them to be in KNOWN_FIELDS
        for key in column_mapping.keys():
            if key not in self.KNOWN_FIELDS:
                raise ValueError(
                    f"Unknown mapped field {key} --> {column_mapping[key]} for f{type(self)}."
                )
        default_mapping = copy.deepcopy(self.DEFAULT_MAPPING)
        default_mapping.update(column_mapping)
        return default_mapping

    def _format_column(self, col_ref, col_name):
        """
        Apply date formatting if the source column has a DATE_FORMAT entry.

        DATE_FORMAT values can be:
        - A format string: the column is parsed directly via to_timestamp, then cast to date.
        - A [format, position] list: for year-only (YYYY / %Y) or year-month
          (YYYYMM / %Y%m) formats, position resolves the ambiguous day:
            'first'  — Jan 1 or the 1st of the month
            'middle' — Jul 2 or the 15th of the month
            'last'   — Dec 31 or the last day of the month

        All paths return a date (not timestamp) so that downstream date arithmetic
        (e.g. DateDelta) works consistently across backends.
        Blank/empty strings are nullified before parsing to avoid format errors.
        Columns already typed as date are returned as-is; timestamp columns (e.g.
        timestamp('UTC') from Snowflake on re-instantiation) are cast to date.
        """
        if col_name not in self.DATE_FORMAT:
            return col_ref

        # If the column is already a date, no further processing is needed.
        # If it's a timestamp (e.g. timestamp('UTC') from Snowflake after a prior
        # instantiation), cast down to date so downstream date arithmetic stays consistent.
        col_type = str(col_ref.type())
        if col_type.startswith("date") and not col_type.startswith("timestamp"):
            return col_ref
        if col_type.startswith("timestamp"):
            return col_ref.cast("date")

        fmt_spec = self.DATE_FORMAT[col_name]
        fmt, position = (
            (fmt_spec[0], fmt_spec[1])
            if isinstance(fmt_spec, list)
            else (fmt_spec, None)
        )

        col_ref = col_ref.nullif("")

        if position is None:
            return col_ref.to_timestamp(fmt).cast("date")

        # Detect backend style: Snowflake has no '%'; DuckDB/BigQuery use '%' prefixes.
        full_date_fmt = "YYYYMMDD" if "%" not in fmt else "%Y%m%d"

        if fmt in ("YYYY", "%Y"):  # year-only
            suffix = {"first": "0101", "middle": "0702", "last": "1231"}[position]
            return col_ref.concat(suffix).to_timestamp(full_date_fmt).cast("date")

        if fmt in ("YYYYMM", "%Y%m"):  # year-month
            if position == "last":
                # Parse as 1st of month, then advance to the true last day.
                first_of_month = col_ref.concat("01").to_timestamp(full_date_fmt)
                return (
                    first_of_month + ibis.interval(months=1) - ibis.interval(days=1)
                ).cast("date")
            suffix = {"first": "01", "middle": "15"}[position]
            return col_ref.concat(suffix).to_timestamp(full_date_fmt).cast("date")

        raise ValueError(
            f"DATE_FORMAT position '{position}' is only supported for year-only "
            f"(YYYY / %Y) or year-month (YYYYMM / %Y%m) formats, got '{fmt}'."
        )

    def _resolve_column_mapping(self, table, column_mapping):
        """
        Convert raw column mapping (strings/lists) to ibis expressions for use in mutate().

        String values become direct column references: table[col].
        List values become coalesce expressions over the listed columns.
        Date columns in a coalesce list are cast to timestamp for consistent typing.
        Date formatting via DATE_FORMAT is applied before coalescing.
        """
        processed_mapping = {}
        for key, value in column_mapping.items():
            if isinstance(value, list):
                # Coalesce multiple columns - first non-null value wins
                # Apply date formatting, then cast dates to timestamp for consistent typing
                cols = []
                for col in value:
                    col_ref = self._format_column(table[col], col)
                    col_type = str(col_ref.type())
                    if col_type.startswith("date") and not col_type.startswith(
                        "timestamp"
                    ):
                        col_ref = col_ref.cast("timestamp")
                    cols.append(col_ref)
                processed_mapping[key] = ibis.coalesce(*cols)
            else:
                # Single column mapping - apply date formatting if specified
                processed_mapping[key] = self._format_column(table[value], value)
        return processed_mapping

    def __getattr__(self, name):
        # pass all attributes on to underlying table
        return getattr(self._table, name)

    def __getitem__(self, key):
        return self._table[key]

    @property
    def table(self):
        return self._table

    def join(self, other: "PhenexTable", *args, domains=None, **kwargs):
        """
        The join method performs a join of PhenexTables, using autojoin functionality if Phenex is able to find the table types specified in PATHS.
        """
        if isinstance(other, Table):
            return type(self)(self.table.join(other, *args, **kwargs))

        if not isinstance(other, PhenexTable):
            raise TypeError(f"Expected a PhenexTable instance, got {type(other)}")
        if len(args):
            # if user specifies join keys and join type, simply perform join as specified
            return type(self)(self.table.join(other.table, *args, **kwargs))

        # Do an autojoin by finding a path from the left to the right table and sequentially joining as necessary
        # joined table is the sequentially joined table
        # current table is the table for the left join in the current iteration
        joined_table = current_left_table = self
        logger.debug(
            f"Starting autojoin from {self.__class__.__name__} to {other.__class__.__name__}"
        )

        for right_table_class_name in self._find_path(other):
            # get the next right table
            right_table_search_results = [
                v
                for k, v in domains.items()
                if v.__class__.__name__ == right_table_class_name
            ]
            logger.debug(
                f"Searching for {right_table_class_name} in domains: {list(domains.keys())}"
            )
            logger.debug(
                f"Found {len(right_table_search_results)} matches for {right_table_class_name}"
            )

            if len(right_table_search_results) != 1:
                raise ValueError(
                    f"Unable to find unqiue {right_table_class_name} required to join {other.__class__.__name__}"
                )
            right_table = right_table_search_results[0]
            print(
                f"\tJoining : {current_left_table.__class__.__name__} to {right_table.__class__.__name__}"
            )

            # join keys are defined by the left table; in theory should enforce symmetry
            join_keys = current_left_table.JOIN_KEYS[right_table_class_name]

            # Build join predicate(s) - supports symmetric and asymmetric joins
            # Symmetric: ["COLUMN"] or ["COL1", "COL2"] - same column names in both tables
            # Asymmetric: [("LEFT_COL", "RIGHT_COL")] - different column names
            # Mixed: ["COL1", ("LEFT_COL", "RIGHT_COL")]
            predicates = []
            for join_key in join_keys:
                if isinstance(join_key, str):
                    # Symmetric: column exists in both tables with same name
                    predicates.append(joined_table[join_key] == right_table[join_key])
                elif isinstance(join_key, (tuple, list)) and len(join_key) == 2:
                    # Asymmetric: (left_col, right_col) - different column names
                    left_col, right_col = join_key
                    predicates.append(joined_table[left_col] == right_table[right_col])
                else:
                    raise ValueError(
                        f"Invalid join key format: {join_key}. Must be either a string or a 2-element tuple/list."
                    )

            # Combine all predicates with AND
            if len(predicates) == 1:
                join_predicate = predicates[0]
            else:
                join_predicate = predicates[0]
                for pred in predicates[1:]:
                    join_predicate = join_predicate & pred

            columns = list(set(joined_table.columns + right_table.columns))
            # subset columns, making sure to set type of table to the very left table (self)
            joined_table = type(self)(
                joined_table.join(right_table, join_predicate, **kwargs).select(columns)
            )
            current_left_table = right_table
        return joined_table

    def mutate(self, *args, **kwargs):
        return type(self)(self.table.mutate(*args, **kwargs), name=self.NAME_TABLE)

    def _find_path(self, other):
        start_name = self.__class__.__name__
        end_name = other.__class__.__name__

        logger.debug(f"Finding path from {start_name} to {end_name}")

        # first see if direct connection
        try:
            join_keys = self.JOIN_KEYS[end_name]
            logger.debug(
                f"Found direct connection: {start_name} -> {end_name} using keys {join_keys}"
            )
            return [end_name]
        except KeyError:
            logger.debug(
                f"No direct connection found in JOIN_KEYS for {start_name} -> {end_name}"
            )
            try:
                path = self.PATHS[end_name]
                full_path = path + [end_name]
                logger.debug(
                    f"Found path in PATHS: {start_name} -> {' -> '.join(full_path)}"
                )
                return full_path
            except KeyError:
                logger.error(f"No path found for {start_name} -> {end_name}")
                logger.debug(
                    f"Available JOIN_KEYS for {start_name}: {list(self.JOIN_KEYS.keys())}"
                )
                logger.debug(
                    f"Available PATHS for {start_name}: {list(self.PATHS.keys())}"
                )
                raise ValueError(
                    f"Cannot autojoin {start_name} --> {end_name}. Please specify join path in PATHS."
                )

    def filter(self, expr):
        """
        Filter the table by an Ibis Expression or using a PhenExFilter.
        """
        input_columns = self.columns
        if isinstance(expr, ibis.expr.types.Expr) or isinstance(expr, list):
            filtered_table = self.table.filter(expr)
        else:
            filtered_table = expr.filter(self)

        return type(self)(
            filtered_table.select(input_columns),
            name=self.NAME_TABLE,
            column_mapping=self.column_mapping,
        )

    @staticmethod
    def find_table_in_domains(name: str, tables: dict) -> "PhenexTable":
        """
        Find a table in a domains dictionary by mapper class name or NAME_TABLE.

        Public mapper configuration should prefer mapper class names for
        CODES_DEFINED_IN / EVENT_DATE_DEFINED_IN to stay aligned with JOIN_KEYS
        and PATHS. NAME_TABLE matching is kept as a compatibility fallback.
        """
        for domain_table in tables.values():
            if domain_table is None:
                continue
            table_name = getattr(domain_table, "NAME_TABLE", None)
            class_name = domain_table.__class__.__name__
            if table_name == name or class_name == name:
                return domain_table

        available = [
            f"{t.__class__.__name__} (NAME_TABLE={getattr(t, 'NAME_TABLE', 'N/A')})"
            for t in tables.values()
            if t is not None
        ]
        raise ValueError(
            f"Table '{name}' not found. Searched by NAME_TABLE and class name. "
            f"Available tables: {', '.join(available)}"
        )

    def resolve_event_date(self, tables: dict) -> "PhenexTable":
        """
        Ensure EVENT_DATE is present, autojoining via EVENT_DATE_DEFINED_IN if needed.

        If EVENT_DATE already exists on this table, returns self unchanged.
        Otherwise joins to the table named by EVENT_DATE_DEFINED_IN and keeps
        the original columns plus EVENT_DATE.
        """
        if "EVENT_DATE" in self.columns:
            return self

        event_date_domain = getattr(self, "EVENT_DATE_DEFINED_IN", None)
        if event_date_domain is None:
            raise ValueError(
                f"{self.__class__.__name__} does not have an EVENT_DATE column "
                "and EVENT_DATE_DEFINED_IN is not set. Either add EVENT_DATE to the "
                "table mapping or set EVENT_DATE_DEFINED_IN to specify which table "
                "contains the event date."
            )
        if tables is None:
            raise ValueError(
                f"Table required for EVENT_DATE ({event_date_domain}) but 'tables' "
                "is None. Pass the domains dictionary via the 'tables' parameter."
            )

        target_table = self.find_table_in_domains(event_date_domain, tables)
        if "EVENT_DATE" not in target_table.columns:
            raise ValueError(
                f"EVENT_DATE_DEFINED_IN='{event_date_domain}' points to "
                f"{target_table.__class__.__name__}, which does not have an "
                "EVENT_DATE column."
            )

        original_columns = list(self.columns)
        joined = self.join(target_table, domains=tables)
        columns_to_keep = [
            c
            for c in list(dict.fromkeys(original_columns + ["EVENT_DATE"]))
            if c in joined.columns
        ]
        return type(self)(
            joined.select(columns_to_keep),
            name=self.NAME_TABLE,
            column_mapping=self.column_mapping,
        )

    @classmethod
    def to_dict(cls) -> dict:
        """
        Serialize the PhenexTable class configuration (not the data).

        This serializes the class-level attributes that define the table mapping,
        but not the actual ibis table data which cannot be serialized.

        Returns:
            dict: Class configuration including NAME_TABLE, JOIN_KEYS, DEFAULT_MAPPING, etc.
        """
        return {
            "__table_class__": cls.__name__,
            "__module__": cls.__module__,
            "NAME_TABLE": cls.NAME_TABLE,
            "JOIN_KEYS": cls.JOIN_KEYS,
            "KNOWN_FIELDS": cls.KNOWN_FIELDS,
            "DEFAULT_MAPPING": cls.DEFAULT_MAPPING,
            "PATHS": cls.PATHS,
            "DATE_FORMAT": cls.DATE_FORMAT,
            "REQUIRED_FIELDS": cls.REQUIRED_FIELDS,
        }

    @classmethod
    def from_dict(cls, data: dict):
        """
        Reconstruct a PhenexTable class reference from serialized data.

        Note: This returns the class itself, not an instance, since we cannot
        reconstruct the actual table data without a database connection.

        Args:
            data: Serialized class configuration

        Returns:
            The PhenexTable subclass
        """
        # The class should already exist in the module, just return it
        return cls

__init__(table, name=None, column_mapping={})

Instantiate a PhenexTable, possibly overriding NAME_TABLE and COLUMN_MAPPING.

Source code in phenex/tables.py

def __init__(self, table, name=None, column_mapping={}):
    """
    Instantiate a PhenexTable, possibly overriding NAME_TABLE and COLUMN_MAPPING.
    """

    if not isinstance(table, Table):
        raise TypeError(
            f"Cannot instantiatiate {self.__class__.__name__} from {type(table)}. Must be ibis Table."
        )

    self.NAME_TABLE = name or self.NAME_TABLE

    self.column_mapping = self._get_column_mapping(column_mapping)
    self._table = table.mutate(
        **self._resolve_column_mapping(table, self.column_mapping)
    )

    for key in self.REQUIRED_FIELDS:
        try:
            getattr(self._table, key)
        except AttributeError:
            raise ValueError(f"Required field {key} not defined in COLUMN_MAPPING.")

    self._add_phenotype_table_relationship()

filter(expr)

Filter the table by an Ibis Expression or using a PhenExFilter.

Source code in phenex/tables.py

def filter(self, expr):
    """
    Filter the table by an Ibis Expression or using a PhenExFilter.
    """
    input_columns = self.columns
    if isinstance(expr, ibis.expr.types.Expr) or isinstance(expr, list):
        filtered_table = self.table.filter(expr)
    else:
        filtered_table = expr.filter(self)

    return type(self)(
        filtered_table.select(input_columns),
        name=self.NAME_TABLE,
        column_mapping=self.column_mapping,
    )

find_table_in_domains(name, tables) staticmethod

Find a table in a domains dictionary by mapper class name or NAME_TABLE.

Public mapper configuration should prefer mapper class names for CODES_DEFINED_IN / EVENT_DATE_DEFINED_IN to stay aligned with JOIN_KEYS and PATHS. NAME_TABLE matching is kept as a compatibility fallback.

Source code in phenex/tables.py

@staticmethod
def find_table_in_domains(name: str, tables: dict) -> "PhenexTable":
    """
    Find a table in a domains dictionary by mapper class name or NAME_TABLE.

    Public mapper configuration should prefer mapper class names for
    CODES_DEFINED_IN / EVENT_DATE_DEFINED_IN to stay aligned with JOIN_KEYS
    and PATHS. NAME_TABLE matching is kept as a compatibility fallback.
    """
    for domain_table in tables.values():
        if domain_table is None:
            continue
        table_name = getattr(domain_table, "NAME_TABLE", None)
        class_name = domain_table.__class__.__name__
        if table_name == name or class_name == name:
            return domain_table

    available = [
        f"{t.__class__.__name__} (NAME_TABLE={getattr(t, 'NAME_TABLE', 'N/A')})"
        for t in tables.values()
        if t is not None
    ]
    raise ValueError(
        f"Table '{name}' not found. Searched by NAME_TABLE and class name. "
        f"Available tables: {', '.join(available)}"
    )

from_dict(data) classmethod

Reconstruct a PhenexTable class reference from serialized data.

Note: This returns the class itself, not an instance, since we cannot reconstruct the actual table data without a database connection.

Parameters:

Name	Type	Description	Default
data	dict	Serialized class configuration	required

Returns:

Type	Description
	The PhenexTable subclass

Source code in phenex/tables.py

@classmethod
def from_dict(cls, data: dict):
    """
    Reconstruct a PhenexTable class reference from serialized data.

    Note: This returns the class itself, not an instance, since we cannot
    reconstruct the actual table data without a database connection.

    Args:
        data: Serialized class configuration

    Returns:
        The PhenexTable subclass
    """
    # The class should already exist in the module, just return it
    return cls

join(other, *args, domains=None, **kwargs)

The join method performs a join of PhenexTables, using autojoin functionality if Phenex is able to find the table types specified in PATHS.

Source code in phenex/tables.py

def join(self, other: "PhenexTable", *args, domains=None, **kwargs):
    """
    The join method performs a join of PhenexTables, using autojoin functionality if Phenex is able to find the table types specified in PATHS.
    """
    if isinstance(other, Table):
        return type(self)(self.table.join(other, *args, **kwargs))

    if not isinstance(other, PhenexTable):
        raise TypeError(f"Expected a PhenexTable instance, got {type(other)}")
    if len(args):
        # if user specifies join keys and join type, simply perform join as specified
        return type(self)(self.table.join(other.table, *args, **kwargs))

    # Do an autojoin by finding a path from the left to the right table and sequentially joining as necessary
    # joined table is the sequentially joined table
    # current table is the table for the left join in the current iteration
    joined_table = current_left_table = self
    logger.debug(
        f"Starting autojoin from {self.__class__.__name__} to {other.__class__.__name__}"
    )

    for right_table_class_name in self._find_path(other):
        # get the next right table
        right_table_search_results = [
            v
            for k, v in domains.items()
            if v.__class__.__name__ == right_table_class_name
        ]
        logger.debug(
            f"Searching for {right_table_class_name} in domains: {list(domains.keys())}"
        )
        logger.debug(
            f"Found {len(right_table_search_results)} matches for {right_table_class_name}"
        )

        if len(right_table_search_results) != 1:
            raise ValueError(
                f"Unable to find unqiue {right_table_class_name} required to join {other.__class__.__name__}"
            )
        right_table = right_table_search_results[0]
        print(
            f"\tJoining : {current_left_table.__class__.__name__} to {right_table.__class__.__name__}"
        )

        # join keys are defined by the left table; in theory should enforce symmetry
        join_keys = current_left_table.JOIN_KEYS[right_table_class_name]

        # Build join predicate(s) - supports symmetric and asymmetric joins
        # Symmetric: ["COLUMN"] or ["COL1", "COL2"] - same column names in both tables
        # Asymmetric: [("LEFT_COL", "RIGHT_COL")] - different column names
        # Mixed: ["COL1", ("LEFT_COL", "RIGHT_COL")]
        predicates = []
        for join_key in join_keys:
            if isinstance(join_key, str):
                # Symmetric: column exists in both tables with same name
                predicates.append(joined_table[join_key] == right_table[join_key])
            elif isinstance(join_key, (tuple, list)) and len(join_key) == 2:
                # Asymmetric: (left_col, right_col) - different column names
                left_col, right_col = join_key
                predicates.append(joined_table[left_col] == right_table[right_col])
            else:
                raise ValueError(
                    f"Invalid join key format: {join_key}. Must be either a string or a 2-element tuple/list."
                )

        # Combine all predicates with AND
        if len(predicates) == 1:
            join_predicate = predicates[0]
        else:
            join_predicate = predicates[0]
            for pred in predicates[1:]:
                join_predicate = join_predicate & pred

        columns = list(set(joined_table.columns + right_table.columns))
        # subset columns, making sure to set type of table to the very left table (self)
        joined_table = type(self)(
            joined_table.join(right_table, join_predicate, **kwargs).select(columns)
        )
        current_left_table = right_table
    return joined_table

resolve_event_date(tables)

Ensure EVENT_DATE is present, autojoining via EVENT_DATE_DEFINED_IN if needed.

If EVENT_DATE already exists on this table, returns self unchanged. Otherwise joins to the table named by EVENT_DATE_DEFINED_IN and keeps the original columns plus EVENT_DATE.

Source code in phenex/tables.py

def resolve_event_date(self, tables: dict) -> "PhenexTable":
    """
    Ensure EVENT_DATE is present, autojoining via EVENT_DATE_DEFINED_IN if needed.

    If EVENT_DATE already exists on this table, returns self unchanged.
    Otherwise joins to the table named by EVENT_DATE_DEFINED_IN and keeps
    the original columns plus EVENT_DATE.
    """
    if "EVENT_DATE" in self.columns:
        return self

    event_date_domain = getattr(self, "EVENT_DATE_DEFINED_IN", None)
    if event_date_domain is None:
        raise ValueError(
            f"{self.__class__.__name__} does not have an EVENT_DATE column "
            "and EVENT_DATE_DEFINED_IN is not set. Either add EVENT_DATE to the "
            "table mapping or set EVENT_DATE_DEFINED_IN to specify which table "
            "contains the event date."
        )
    if tables is None:
        raise ValueError(
            f"Table required for EVENT_DATE ({event_date_domain}) but 'tables' "
            "is None. Pass the domains dictionary via the 'tables' parameter."
        )

    target_table = self.find_table_in_domains(event_date_domain, tables)
    if "EVENT_DATE" not in target_table.columns:
        raise ValueError(
            f"EVENT_DATE_DEFINED_IN='{event_date_domain}' points to "
            f"{target_table.__class__.__name__}, which does not have an "
            "EVENT_DATE column."
        )

    original_columns = list(self.columns)
    joined = self.join(target_table, domains=tables)
    columns_to_keep = [
        c
        for c in list(dict.fromkeys(original_columns + ["EVENT_DATE"]))
        if c in joined.columns
    ]
    return type(self)(
        joined.select(columns_to_keep),
        name=self.NAME_TABLE,
        column_mapping=self.column_mapping,
    )

to_dict() classmethod

Serialize the PhenexTable class configuration (not the data).

This serializes the class-level attributes that define the table mapping, but not the actual ibis table data which cannot be serialized.

Returns:

Name	Type	Description
dict	dict	Class configuration including NAME_TABLE, JOIN_KEYS, DEFAULT_MAPPING, etc.

Source code in phenex/tables.py

@classmethod
def to_dict(cls) -> dict:
    """
    Serialize the PhenexTable class configuration (not the data).

    This serializes the class-level attributes that define the table mapping,
    but not the actual ibis table data which cannot be serialized.

    Returns:
        dict: Class configuration including NAME_TABLE, JOIN_KEYS, DEFAULT_MAPPING, etc.
    """
    return {
        "__table_class__": cls.__name__,
        "__module__": cls.__module__,
        "NAME_TABLE": cls.NAME_TABLE,
        "JOIN_KEYS": cls.JOIN_KEYS,
        "KNOWN_FIELDS": cls.KNOWN_FIELDS,
        "DEFAULT_MAPPING": cls.DEFAULT_MAPPING,
        "PATHS": cls.PATHS,
        "DATE_FORMAT": cls.DATE_FORMAT,
        "REQUIRED_FIELDS": cls.REQUIRED_FIELDS,
    }