_conformed

_conformed

Conformed property mappings for stream configurations.

This module defines semantic properties that exist across different stream types (e.g., "query", "database", "warehouse") and maps them to platform-specific configuration parameter names.

This allows DomoStream to provide a consistent interface regardless of the underlying data provider:

stream.sql          # Works for Snowflake, Athena, PostgreSQL, etc.
stream.database     # Works across database connectors
stream.report_id    # Works for Adobe, Qualtrics, Google Sheets

ConformedProperty dataclass

ConformedProperty(
    name: str,
    mappings: dict[str, str],
    description: str = None,
    is_repr: bool = False,
)

Maps semantic property names to platform-specific config keys.

Defines a "conformed" or "semantic" property that exists conceptually across multiple data providers, even though each provider may use a different parameter name.

Attributes:

Name	Type	Description
name	str	Semantic name of the property (e.g., "query", "database")
mappings	dict[str, str]	Dict mapping provider_type to typed_config attribute name
description	str	Human-readable description of what this property represents
supported_providers	list[str]	List of provider types that support this property
is_repr	bool	Whether to include this property in repr output (default: False)

Example

query = ConformedProperty( name="query", description="SQL query or data selection statement", mappings={ "snowflake": "query", "aws-athena": "query", "amazon-athena-high-bandwidth": "query", "postgresql": "query", }, is_repr=True # Show in repr )

supported_providers property

supported_providers: list[str]

List of provider types that support this property.

get_key_for_provider

get_key_for_provider(provider_type: str) -> str | None

Get the typed_config attribute name for a specific provider.

Parameters:

Name	Type	Description	Default
provider_type	str	Data provider type (e.g., "snowflake", "aws-athena")	required

Returns:

Type	Description
str | None	Attribute name on the typed config class, or None if not supported

Example

query_prop.get_key_for_provider("snowflake") "query" query_prop.get_key_for_provider("unknown-provider") None

Source code in src/crew_dcs/classes/DomoDataset/stream_configs/_conformed.py

def get_key_for_provider(self, provider_type: str) -> str | None:
    """Get the typed_config attribute name for a specific provider.

    Args:
        provider_type: Data provider type (e.g., "snowflake", "aws-athena")

    Returns:
        Attribute name on the typed config class, or None if not supported

    Example:
        >>> query_prop.get_key_for_provider("snowflake")
        "query"
        >>> query_prop.get_key_for_provider("unknown-provider")
        None
    """
    return self.mappings.get(provider_type)