Skip to content

Data Sources

The config module provides a registry of supported data sources with their column mappings and table patterns. This enables writing source-agnostic code that works across IQVIA, Optum, Komodo, and other databases.

When to Use

Use the config module when you need to:

  • Get column mappings for a data source
  • Write source-agnostic analysis code
  • Check which data sources are supported
  • Access table naming patterns

Usually Automatic

Most functions accept a source parameter and handle configuration internally. Use this module directly when building custom queries or debugging.

Quick Example

from alx_heor.config import get_source_config, list_sources

# List available sources
print(list_sources())  # ['iqvia', 'komodo', 'optum']

# Get configuration for IQVIA
config = get_source_config("iqvia")

# Access column mappings
patient_col = config["columns"]["patient_id"]  # 'pat_id'
date_col = config["columns"]["service_date"]   # 'from_dt'

# Access table patterns
claims_pattern = config["claims_table_pattern"]  # 'claims_{year}'

Column Mappings by Source

Logical Column IQVIA Optum Komodo
patient_id pat_id patid patient_id
service_date from_dt svcdate service_date
service_end_date to_dt tsvcdat service_end_date
month_id month_id eligeff month_id
ndc ndc ndc ndc
procedure_code proc1 proc1 procedure_code
place_of_service pos pos pos
year_of_birth der_yob yrdob birth_year
sex der_sex gdr_cd gender
payer_type pay_type bus payer_type

Table Patterns

Source Claims Table Enrollment Table
IQVIA claims_{year} enroll2_{year}
Optum medical_{year} member_{year}
Komodo claims_{year} enrollment_{year}

Common Patterns

Source-Agnostic Code

from alx_heor.config import get_source_config

def get_patient_count(conn, source, schema):
    """Count patients in a source-agnostic way."""
    config = get_source_config(source)
    patient_col = config["columns"]["patient_id"]
    table_pattern = config["claims_table_pattern"]
    table = table_pattern.format(year=2024)

    sql = f"SELECT COUNT(DISTINCT {patient_col}) FROM {schema}.{table}"
    return conn.query(sql).iloc[0, 0]

# Works with any source
count_iqvia = get_patient_count(conn, "iqvia", "iqvia_schema")
count_optum = get_patient_count(conn, "optum", "optum_schema")

Checking Supported Sources

from alx_heor.config import list_sources, get_source_config

if "iqvia" in list_sources():
    config = get_source_config("iqvia")
    print(f"IQVIA patient ID column: {config['columns']['patient_id']}")
  • claims - Uses config for column mapping
  • enrollment - Uses config for enrollment tables
  • cohort - Uses config throughout

sources

Data source registry and column mappings for multi-database support.

This module is the heart of the library's multi-data-source architecture. It defines column name mappings and table patterns for each supported healthcare claims database, enabling the same analysis code to work across IQVIA, Optum, Komodo, and other data sources.

Why Column Mapping Matters:

Different data vendors use different column names for the same concepts: - Patient ID: pat_id (IQVIA) vs patid (Optum) vs patient_id (Komodo) - Service date: from_dt (IQVIA) vs svcdate (Optum) - Diagnosis fields: 12 fields in IQVIA, 5 in Optum

Without a mapping layer, you'd need separate code for each database. This module provides a single source of truth for column names, allowing functions like get_claims() to work identically across all data sources.

Supported Data Sources:

  • IQVIA Pharmetrics: US commercial claims (Redshift)
  • Optum DOD: US commercial + Medicare claims (Redshift)
  • Komodo: US all-payer claims (Snowflake)
  • MDV: Japanese claims (planned)

Configuration Structure:

Each data source config includes: - name: Human-readable name - claims_table_pattern: Pattern for yearly claims tables (e.g., 'claims_{year}') - enrollment_table_pattern: Pattern for enrollment tables - columns: Mapping of logical names to actual column names - default_claims_columns: Default columns for claims queries

Usage:

Most users won't call this module directly - the higher-level functions (get_claims, get_cohort, etc.) use it internally via the source parameter.

Example

from alx_heor.config import get_source_config, list_sources print(list_sources()) ['iqvia', 'komodo', 'optum'] config = get_source_config('iqvia') print(f"Patient ID column: {config['columns']['patient_id']}") Patient ID column: pat_id print(f"Claims table pattern: {config['claims_table_pattern']}") Claims table pattern: claims_{year}

See Also

claims.get_claims : Query claims using source configuration cohort.get_cohort : Build cohorts using source configuration

Notes
  • Column names are case-sensitive in SQL
  • Table patterns use Python format strings with {year} placeholder
  • Add new data sources by extending the SOURCES dictionary
  • IQVIA has 12 diagnosis fields; Optum has 5 - this affects query generation

ColumnMapping

Bases: TypedDict

Column name mappings for a data source.

Source code in alx_heor\config\sources.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
class ColumnMapping(TypedDict, total=False):
    """Column name mappings for a data source."""

    patient_id: str
    service_date: str
    service_end_date: str
    month_id: str
    diagnosis: list[str]
    ndc: str
    place_of_service: str
    procedure_code: str
    record_type: str
    year_of_birth: str
    sex: str
    payer_type: str

DataSourceConfig

Bases: TypedDict

Configuration for a claims data source.

Source code in alx_heor\config\sources.py 
86
87
88
89
90
91
92
93
94
95
class DataSourceConfig(TypedDict, total=False):
    """Configuration for a claims data source."""

    name: str
    claims_table_pattern: str
    enrollment_table_pattern: str
    rx_lookup_table: str
    pr_lookup_table: str
    columns: ColumnMapping
    default_claims_columns: list[str]

get_source_config 

get_source_config(source: str) -> DataSourceConfig

Get configuration for a data source by name.

This function retrieves the column mappings and table patterns for a specific data source. It's the primary interface for accessing the source registry, used internally by most library functions.

The configuration enables database-agnostic code by abstracting away vendor-specific column names. When you call get_claims(source='iqvia'), the function uses this configuration to know that patient_id is pat_id and service_date is from_dt.

Parameters:

Name Type Description Default
source str

Data source name: 'iqvia', 'optum', 'komodo', etc. Case-insensitive (converted to lowercase internally).

 required

Returns:

Type Description
DataSourceConfig

Configuration dictionary with: - name: Human-readable name (e.g., 'IQVIA Pharmetrics') - claims_table_pattern: Pattern for claims tables (e.g., 'claims_{year}') - enrollment_table_pattern: Pattern for enrollment tables - columns: Dict mapping logical names to actual column names - default_claims_columns: List of default columns for queries

Raises:

Type Description
ValueError

If the source is not recognized. Error message lists available sources.
See Also

list_sources : Get list of all available data sources SOURCES : The underlying registry dictionary

Notes
  • Source names are case-insensitive ('IQVIA' == 'iqvia')
  • The columns dict uses logical names as keys (patient_id, service_date)
  • Diagnosis columns are returned as a list (may vary in length by source)

Examples:

Get IQVIA configuration:

>>> config = get_source_config('iqvia')
>>> print(f"Patient ID: {config['columns']['patient_id']}")
Patient ID: pat_id
>>> print(f"Service date: {config['columns']['service_date']}")
Service date: from_dt
>>> print(f"Diagnosis columns: {config['columns']['diagnosis'][:3]}")
Diagnosis columns: ['diag1', 'diag2', 'diag3']

Compare IQVIA vs Optum column names:

>>> iqvia = get_source_config('iqvia')
>>> optum = get_source_config('optum')
>>> print(f"IQVIA patient ID: {iqvia['columns']['patient_id']}")
IQVIA patient ID: pat_id
>>> print(f"Optum patient ID: {optum['columns']['patient_id']}")
Optum patient ID: patid

Use in custom queries:

>>> config = get_source_config('iqvia')
>>> pat_col = config['columns']['patient_id']
>>> date_col = config['columns']['service_date']
>>> sql = f"SELECT {pat_col}, {date_col} FROM claims WHERE ..."

Handle unknown source gracefully:

>>> try:
...     config = get_source_config('unknown')
... except ValueError as e:
...     print(f"Error: {e}")
Error: Unknown data source: 'unknown'. Available sources: iqvia, komodo, optum
Source code in alx_heor\config\sources.py 
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
def get_source_config(source: str) -> DataSourceConfig:
    """Get configuration for a data source by name.

    This function retrieves the column mappings and table patterns for a
    specific data source. It's the primary interface for accessing the
    source registry, used internally by most library functions.

    The configuration enables database-agnostic code by abstracting away
    vendor-specific column names. When you call `get_claims(source='iqvia')`,
    the function uses this configuration to know that patient_id is `pat_id`
    and service_date is `from_dt`.

    Parameters
    ----------
    source : str
        Data source name: 'iqvia', 'optum', 'komodo', etc.
        Case-insensitive (converted to lowercase internally).

    Returns
    -------
    DataSourceConfig
        Configuration dictionary with:
        - name: Human-readable name (e.g., 'IQVIA Pharmetrics')
        - claims_table_pattern: Pattern for claims tables (e.g., 'claims_{year}')
        - enrollment_table_pattern: Pattern for enrollment tables
        - columns: Dict mapping logical names to actual column names
        - default_claims_columns: List of default columns for queries

    Raises
    ------
    ValueError
        If the source is not recognized. Error message lists available sources.

    See Also
    --------
    list_sources : Get list of all available data sources
    SOURCES : The underlying registry dictionary

    Notes
    -----
    - Source names are case-insensitive ('IQVIA' == 'iqvia')
    - The columns dict uses logical names as keys (patient_id, service_date)
    - Diagnosis columns are returned as a list (may vary in length by source)

    Examples
    --------
    Get IQVIA configuration:

    >>> config = get_source_config('iqvia')
    >>> print(f"Patient ID: {config['columns']['patient_id']}")
    Patient ID: pat_id
    >>> print(f"Service date: {config['columns']['service_date']}")
    Service date: from_dt
    >>> print(f"Diagnosis columns: {config['columns']['diagnosis'][:3]}")
    Diagnosis columns: ['diag1', 'diag2', 'diag3']

    Compare IQVIA vs Optum column names:

    >>> iqvia = get_source_config('iqvia')
    >>> optum = get_source_config('optum')
    >>> print(f"IQVIA patient ID: {iqvia['columns']['patient_id']}")
    IQVIA patient ID: pat_id
    >>> print(f"Optum patient ID: {optum['columns']['patient_id']}")
    Optum patient ID: patid

    Use in custom queries:

    >>> config = get_source_config('iqvia')
    >>> pat_col = config['columns']['patient_id']
    >>> date_col = config['columns']['service_date']
    >>> sql = f"SELECT {pat_col}, {date_col} FROM claims WHERE ..."

    Handle unknown source gracefully:

    >>> try:
    ...     config = get_source_config('unknown')
    ... except ValueError as e:
    ...     print(f"Error: {e}")
    Error: Unknown data source: 'unknown'. Available sources: iqvia, komodo, optum
    """
    source_lower = source.lower()
    if source_lower not in SOURCES:
        available = ", ".join(sorted(SOURCES.keys()))
        raise ValueError(
            f"Unknown data source: '{source}'. Available sources: {available}"
        )
    return SOURCES[source_lower]

list_sources 

list_sources() -> list[str]

List all available data sources.

Returns:

Type Description
list[str]

List of data source names.
Example

list_sources() ['iqvia', 'komodo', 'optum']

Source code in alx_heor\config\sources.py 
280
281
282
283
284
285
286
287
288
289
290
291
292
293
def list_sources() -> list[str]:
    """List all available data sources.

    Returns
    -------
    list[str]
        List of data source names.

    Example
    -------
    >>> list_sources()
    ['iqvia', 'komodo', 'optum']
    """
    return sorted(SOURCES.keys())