config

config

DomoAccount_Config dataclass

DomoAccount_Config(
    data_provider_type: str,
    is_oauth: bool,
    allow_external_use: bool = True,
    parent: Any = None,
    raw: dict = None,
)

Bases: DomoBase

DomoAccount Config abstract base class

from_dict classmethod

from_dict(
    obj: dict[str, Any], parent: Any = None, **kwargs
)

Create instance from dict, handling camelCase, snake_case, and _field_map. Robust to missing or None _field_map. Uses _field_map for both directions.

Source code in src/crew_dcs/classes/DomoAccount/account_configs/_base.py

@classmethod
def from_dict(cls, obj: dict[str, Any], parent: Any = None, **kwargs):
    """
    Create instance from dict, handling camelCase, snake_case, and _field_map.
    Robust to missing or None _field_map. Uses _field_map for both directions.
    """
    # Get _field_map from dataclass field's default_factory if it exists
    field_map = {}
    if (
        hasattr(cls, "__dataclass_fields__")
        and "_field_map" in cls.__dataclass_fields__
    ):
        field_def = cls.__dataclass_fields__["_field_map"]
        if field_def.default_factory:
            field_map = field_def.default_factory()

    # Build a reverse map for input keys that are snake_case but need to be mapped to camelCase
    reverse_map = {
        v: k for k, v in field_map.items()
    }  # {"passphrase" : "passPhrase"}

    init_kwargs = {}

    init_kwargs.update(kwargs)

    for k, v in obj.items():
        if k in ["_search_metadata"]:
            continue

        # Try direct field_map, then reverse_map, then camel_to_snake
        if k in field_map:
            attr = field_map[k]

        elif k in reverse_map:
            attr = reverse_map[k]

        else:
            attr = cls._camel_to_snake(k)

        init_kwargs[attr] = v

    return cls(parent=parent, raw=obj, **init_kwargs)

to_dict

to_dict(
    export_fields: list[str] | None = None,
    override_fn: Callable | None = None,
    return_snake_case: bool = False,
    **kwargs
) -> dict

Convert config to dictionary using provided fields or all _fields_for_serialization.

Source code in src/crew_dcs/classes/DomoAccount/account_configs/_base.py

def to_dict(
    self,
    export_fields: list[str] | None = None,
    override_fn: Callable | None = None,
    return_snake_case: bool = False,
    **kwargs,
) -> dict:
    """
    Convert config to dictionary using provided fields or all _fields_for_serialization.
    """
    result = {"allowExternalUse": self.allow_external_use}

    # Use all fields_for_serialization if not provided
    export_fields = [
        key
        for key in (
            export_fields
            or self._fields_for_serialization
            or self.__dataclass_fields__.keys()
        )
        if key
        not in [
            "raw",
            "_field_map",
            "_fields_for_serialization",
            "is_oauth",
            "data_provider_type",
            "__serialize_properties__",
        ]
    ]

    reverse_map = {v: k for k, v in self._field_map.items()}

    for attr in export_fields:
        val = getattr(self, attr, None)
        if val is not None:
            key = reverse_map.get(attr, self._snake_to_camel(attr))
            result[key] = val

    result.update(kwargs)

    if return_snake_case:
        result = {self._camel_to_snake(k): v for k, v in result.items()}

    if override_fn:
        result = override_fn(self)

    return result

validate_compare_serialization

validate_compare_serialization(
    raise_on_mismatch: bool = True,
) -> bool

Validate that to_dict() output matches raw data keys (ignoring values).

Parameters:

Name	Type	Description	Default
raise_on_mismatch	bool	If True, raises AccountConfig_SerializationMismatchError on mismatch. If False, returns False on mismatch.	True

Returns:

Type	Description
bool	True if keys match, False otherwise (only if raise_on_mismatch=False)

Raises:

Type	Description
AccountConfig_SerializationMismatchError	If keys don't match and raise_on_mismatch=True

Source code in src/crew_dcs/classes/DomoAccount/account_configs/_base.py

def validate_compare_serialization(self, raise_on_mismatch: bool = True) -> bool:
    """
    Validate that to_dict() output matches raw data keys (ignoring values).

    Args:
        raise_on_mismatch: If True, raises AccountConfig_SerializationMismatchError on mismatch.
                         If False, returns False on mismatch.

    Returns:
        True if keys match, False otherwise (only if raise_on_mismatch=False)

    Raises:
        AccountConfig_SerializationMismatchError: If keys don't match and raise_on_mismatch=True
    """
    if not self.raw:
        return True

    to_dict_keys = set(self.to_dict(return_snake_case=False).keys())
    raw_keys = set(self.raw.keys()) - {"_search_metadata", "data_provider_type"}

    only_in_to_dict = to_dict_keys - raw_keys
    only_in_raw = raw_keys - to_dict_keys

    has_mismatch = bool(only_in_to_dict or only_in_raw)

    if has_mismatch and raise_on_mismatch:
        account_id = self.raw.get("_search_metadata", {}).get(
            "account_id", "unknown"
        )
        raise AccountConfig_SerializationMismatchError(
            cls_instance=self,
            data_provider_type=self.data_provider_type,
            account_id=str(account_id),
            only_in_to_dict=dict.fromkeys(only_in_to_dict),
            only_in_raw=dict.fromkeys(only_in_raw),
            value_mismatches={},
        )

    return not has_mismatch

DomoAccount_Config_AbstractCredential dataclass

DomoAccount_Config_AbstractCredential(
    data_provider_type: str = "abstract-credential-store",
    is_oauth: bool = False,
    allow_external_use: bool = True,
    parent: Any = None,
    raw: dict = None,
    credentials: dict = None,
)

Bases: DomoAccount_Config

Abstract credential store — stores arbitrary key-value credentials.

The Domo API expects the credentials field as a JSON string, not a dict. This class handles the serialization/deserialization transparently so callers can work with a plain Python dict.

from_dict classmethod

from_dict(obj, parent=None, **kwargs)

Parse the API response, converting credentials string to dict.

Source code in src/crew_dcs/classes/DomoAccount/account_configs/domo.py

@classmethod
def from_dict(cls, obj, parent=None, **kwargs):
    """Parse the API response, converting credentials string to dict."""
    raw_creds = obj.get("credentials")
    if isinstance(raw_creds, str):
        # keep as-is if not parseable (e.g. masked)
        with contextlib.suppress(json.JSONDecodeError, TypeError):
            raw_creds = json.loads(raw_creds)

    return cls(
        parent=parent,
        raw=obj,
        credentials=raw_creds,
        **{
            k: v
            for k, v in kwargs.items()
            if k not in ["credentials", "raw", "parent"]
        },
    )

to_dict

to_dict(**kwargs) -> dict

Serialize to API format — credentials as JSON string.

Source code in src/crew_dcs/classes/DomoAccount/account_configs/domo.py

def to_dict(self, **kwargs) -> dict:
    """Serialize to API format — credentials as JSON string."""
    result = {"allowExternalUse": self.allow_external_use}
    if self.credentials is not None:
        result["credentials"] = json.dumps(self.credentials)
    return result

DomoAccount_Config_Bls dataclass

DomoAccount_Config_Bls(
    data_provider_type: str = "bls",
    is_oauth: bool = False,
    allow_external_use: bool = True,
    parent: Any = None,
    raw: dict = None,
)

Bases: DomoAccount_Config

Bureau of Labor Statistics — public data API, no credentials required.

The live API response only contains allowExternalUse.

DomoAccount_Config_DomoToS3 dataclass

DomoAccount_Config_DomoToS3(
    data_provider_type: str = "domo-to-s3",
    is_oauth: bool = False,
    allow_external_use: bool = True,
    parent: Any = None,
    raw: dict = None,
    client_id: str = None,
    client_secret: str = None,
    bucket: str = None,
    region: str = "us-west-2",
    _field_map: list[dict[str:str]] = (
        lambda: {
            "client_id": "access_key",
            "client_secret": "secret_Key",
        }
    )(),
)

Bases: DomoAccount_Config

Domo to S3 export connector configuration.

DomoAccount_Config_DomoToSalesforce2 dataclass

DomoAccount_Config_DomoToSalesforce2(
    data_provider_type: str = "domo-to-salesforce-2",
    is_oauth: bool = True,
    allow_external_use: bool = True,
    parent: Any = None,
    raw: dict = None,
)

Bases: DomoAccount_Config_OAuthBase

Domo → Salesforce write-back (v2) — OAuth, no credential fields exposed via config API.

DomoAccount_Config_GoogleSheets dataclass

DomoAccount_Config_GoogleSheets(
    data_provider_type: str = "google-sheets",
    is_oauth: bool = True,
    allow_external_use: bool = True,
    parent: Any = None,
    raw: dict = None,
)

Bases: DomoAccount_Config_OAuthBase

Google Sheets — OAuth, no credential fields exposed via config API.

DomoAccount_Config_HighBandwidthConnector dataclass

DomoAccount_Config_HighBandwidthConnector(
    data_provider_type: str = "amazon-athena-high-bandwidth",
    is_oauth: bool = False,
    allow_external_use: bool = True,
    parent: Any = None,
    raw: dict = None,
    access_key: str = None,
    secret_key: str = None,
    bucket: str = None,
    region: str = "us-west-2",
    workgroup: str = None,
)

Bases: DomoAccount_Config

this connector is not enabled by default contact your CSM / AE

DomoAccount_Config_Json5 dataclass

DomoAccount_Config_Json5(
    data_provider_type: str = "json5",
    is_oauth: bool = False,
    allow_external_use: bool = True,
    parent: Any = None,
    raw: dict = None,
    api_key: str = None,
    name: str = None,
    api_type: str = None,
    authentication: str = None,
)

Bases: DomoAccount_Config

JSON5 REST/JSON connector authenticated with an API key.

Fields reverse-engineered from live API response

apiKey → api_key name — header name (e.g. "x-domo-developer-token") apiType → api_type (e.g. "header") authentication — auth scheme (e.g. "apiKey")

DomoAccount_Config_OAuthBase dataclass

DomoAccount_Config_OAuthBase(
    data_provider_type: str = "",
    is_oauth: bool = True,
    allow_external_use: bool = True,
    parent: Any = None,
    raw: dict = None,
)

Bases: DomoAccount_Config

Base for OAuth connector accounts that return null from the config API.

from_dict is safe to call with obj=None (OAuth connectors have no retrievable credential fields). to_dict returns {} so callers that check config.to_dict() != {} skip the missing-key test.

DomoAccount_Config_Salesforce dataclass

DomoAccount_Config_Salesforce(
    data_provider_type: str = "salesforce",
    is_oauth: bool = True,
    allow_external_use: bool = True,
    parent: Any = None,
    raw: dict = None,
)

Bases: DomoAccount_Config_OAuthBase

Salesforce CRM — OAuth, no credential fields exposed via config API.

DomoAccount_Config_Snowflake dataclass

DomoAccount_Config_Snowflake(
    data_provider_type: str = "snowflake",
    is_oauth: bool = False,
    allow_external_use: bool = True,
    parent: Any = None,
    raw: dict = None,
    account: str = None,
    username: str = None,
    password: str = None,
    host: str = None,
    role: str = None,
)

Bases: DomoAccount_Config

this connector is not enabled by default contact your CSM / AE

DomoAccount_Config_SnowflakeKeyPairAuthentication dataclass

DomoAccount_Config_SnowflakeKeyPairAuthentication(
    data_provider_type: str = "snowflakekeypairauthentication",
    is_oauth: bool = False,
    allow_external_use: bool = True,
    parent: Any = None,
    raw: dict = None,
    private_key: str = None,
    account: str = None,
    passphrase: str = None,
    username: str = None,
    role: str = None,
    _fields_for_serialization: list[str] = (
        lambda: [
            "private_key",
            "role",
            "account",
            "username",
            "passphrase",
        ]
    )(),
)

Bases: DomoAccount_Config

Snowflake Key Pair Authentication.

Required fields for this config (enforced at runtime): - private_key - account - username

These values are required by the Domo API for Snowflake key pair accounts and must be present on the config instance before serialization.

DomoAccount_Config_SnowflakeKeyPairInternalManagedUnload dataclass

DomoAccount_Config_SnowflakeKeyPairInternalManagedUnload(
    data_provider_type: str = "snowflake-keypair-internal-managed-unload",
    is_oauth: bool = False,
    allow_external_use: bool = True,
    parent: Any = None,
    raw: dict = None,
    private_key: str = None,
    account: str = None,
    passphrase: str = None,
    username: str = None,
    role: str = None,
    _field_map: dict = (
        lambda: {"passPhrase": "passphrase"}
    )(),
)

Bases: DomoAccount_Config

Snowflake Key Pair Internal Managed Unload

DomoAccount_Config_SnowflakeKeyPairUnload_V2 dataclass

DomoAccount_Config_SnowflakeKeyPairUnload_V2(
    data_provider_type: str = "snowflake-key-pair-unload-v2",
    is_oauth: bool = False,
    allow_external_use: bool = True,
    parent: Any = None,
    raw: dict = None,
    private_key: str = None,
    account: str = None,
    passphrase: str = None,
    username: str = None,
    role: str = None,
    access_key: str = None,
    secret_key: str = None,
    bucket: str = None,
    region: str = None,
    encryption_type: str = None,
    _field_map: dict = (
        lambda: {"passPhrase": "passphrase"}
    )(),
    _fields_for_serialization: list[str] = (
        lambda: [
            "private_key",
            "role",
            "account",
            "username",
            "passphrase",
            "access_key",
            "secret_key",
            "bucket",
            "region",
            "encryption_type",
        ]
    )(),
)

Bases: DomoAccount_Config

Snowflake Key Pair Authentication for Unload V2

DomoAccount_Config_SnowflakeKeyPairWriteback dataclass

DomoAccount_Config_SnowflakeKeyPairWriteback(
    data_provider_type: str = "snowflake-key-pair-writeback",
    is_oauth: bool = False,
    allow_external_use: bool = True,
    parent: Any = None,
    raw: dict = None,
    private_key: str = None,
    account: str = None,
    passphrase: str = None,
    username: str = None,
    role: str = None,
    _field_map: dict = (
        lambda: {"passPhrase": "passphrase"}
    )(),
    _fields_for_serialization: list[str] = (
        lambda: [
            "private_key",
            "role",
            "account",
            "username",
            "passphrase",
        ]
    )(),
)

Bases: DomoAccount_Config

Snowflake Key Pair Authentication for Writeback

DomoAccount_Config_SnowflakeUnload_V2 dataclass

DomoAccount_Config_SnowflakeUnload_V2(
    data_provider_type: str = "snowflake-unload-v2",
    is_oauth: bool = False,
    allow_external_use: bool = True,
    parent: Any = None,
    raw: dict = None,
    account: str = None,
    username: str = None,
    password: str = None,
    access_key: str = None,
    secret_key: str = None,
    region: str = None,
    bucket: str = None,
    role: str = None,
    _fields_for_serialization: list[str] = (
        lambda: [
            "account",
            "username",
            "password",
            "access_key",
            "secret_key",
            "bucket",
            "region",
            "role",
        ]
    )(),
)

Bases: DomoAccount_Config

this connector is not enabled by default contact your CSM / AE

DomoAccount_Config_Squareup dataclass

DomoAccount_Config_Squareup(
    data_provider_type: str = "squareup",
    is_oauth: bool = True,
    allow_external_use: bool = True,
    parent: Any = None,
    raw: dict = None,
)

Bases: DomoAccount_Config_OAuthBase

Square (Squareup) payments — OAuth, no credential fields exposed via config API.

register_account_config

register_account_config(data_provider_type: str)

Decorator to register a DomoAccount_Config subclass.

Parameters:

Name	Type	Description	Default
data_provider_type	str	The data provider type identifier (e.g., 'snowflake')	required

Example

@register_account_config('snowflake') @dataclass class DomoAccount_Config_Snowflake(DomoAccount_Config): account: str = None username: str = None password: str = field(repr=False, default=None) role: str = None

Source code in src/crew_dcs/classes/DomoAccount/config.py

def register_account_config(data_provider_type: str):
    """Decorator to register a DomoAccount_Config subclass.

    Args:
        data_provider_type: The data provider type identifier (e.g., 'snowflake')

    Example:
        @register_account_config('snowflake')
        @dataclass
        class DomoAccount_Config_Snowflake(DomoAccount_Config):
            account: str = None
            username: str = None
            password: str = field(repr=False, default=None)
            role: str = None
    """

    def decorator(cls: type[DomoAccount_Config]) -> type[DomoAccount_Config]:
        _ACCOUNT_CONFIG_REGISTRY[data_provider_type] = cls
        return cls

    return decorator