magic_etl

magic_etl

MagicETL_Dataflow dataclass

MagicETL_Dataflow(
    auth: DomoAuth,
    id: str,
    raw: dict,
    name: str | None = None,
    owner: str | None = None,
    description: str | None = None,
    tags: list[str] | None = None,
    version_id: int | None = None,
    version_number: int | None = None,
    versions: list[dict[str, Any]] | None = None,
    jupyter_workspace_config: dict | None = None,
    Definition: DomoDataflow_Definition | None = None,
    History: DomoDataflow_History | None = None,
    JupyterWorkspace: DomoJupyterWorkspace | None = None,
)

Bases: DomoDataflow

DomoDataflow subclass for Magic ETL (databaseType == "MAGIC") dataflows.

Adds Magic ETL-specific functionality, most notably instructions which fetches the full tile/function reference from the Domo expression-docs API.

instructions async

instructions(
    return_raw: bool = False,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    pin_version: str | None = None,
    **context_kwargs
)

Retrieve the Magic ETL tile/function reference for this instance's Domo instance.

Calls v2 endpoint by default, falls back to v1 if unavailable (non-auth errors). Logs a warning if fallback is used. Raises if both fail.

Parameters:

Name	Type	Description	Default
return_raw	bool	Return the raw ResponseGetData object instead of the parsed response body.	False
debug_api	bool	Enable API debugging output.	False
debug_num_stacks_to_drop	int	Stack frames to drop in debug logging.	2
session	AsyncClient | None	Optional httpx.AsyncClient to reuse.	None
context	RouteContext | None	RouteContext for request configuration.	None
pin_version	str | None	'v2', 'v1', or None for auto-fallback.	None
**context_kwargs		Additional context parameters.	{}

Returns:

Type	Description
	Parsed tile documentation (list / dict) when return_raw=False,
	or the ResponseGetData object when return_raw=True.

Source code in src/crew_dcs/classes/DomoDataflow/magic_etl.py

async def instructions(
    self,
    return_raw: bool = False,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    pin_version: str | None = None,  # 'v2', 'v1', or None for auto
    **context_kwargs,
):
    """Retrieve the Magic ETL tile/function reference for this instance's Domo instance.

    Calls v2 endpoint by default, falls back to v1 if unavailable (non-auth errors).
    Logs a warning if fallback is used. Raises if both fail.

    Args:
        return_raw: Return the raw ``ResponseGetData`` object instead of the
            parsed response body.
        debug_api: Enable API debugging output.
        debug_num_stacks_to_drop: Stack frames to drop in debug logging.
        session: Optional ``httpx.AsyncClient`` to reuse.
        context: ``RouteContext`` for request configuration.
        pin_version: 'v2', 'v1', or None for auto-fallback.
        **context_kwargs: Additional context parameters.

    Returns:
        Parsed tile documentation (list / dict) when ``return_raw=False``,
        or the ``ResponseGetData`` object when ``return_raw=True``.
    """
    import logging

    logger = logging.getLogger("crew_dcs.magic_etl.instructions")

    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        debug_num_stacks_to_drop=debug_num_stacks_to_drop,
        **context_kwargs,
    )

    # Allow pinning version for explicit calls
    if pin_version == "v1":
        res = await dataflow_routes.get_magic_etl_tile_documentation_v1(
            auth=self.auth,
            context=context,
        )
        if return_raw:
            return res
        return res.response
    if pin_version == "v2":
        res = await dataflow_routes.get_magic_etl_tile_documentation(
            auth=self.auth,
            context=context,
        )
        if return_raw:
            return res
        return res.response

    # Default: try v2, fallback to v1 if non-auth error
    try:
        res = await dataflow_routes.get_magic_etl_tile_documentation(
            auth=self.auth,
            context=context,
        )
        if return_raw:
            return res
        return res.response
    except Exception as e:
        # Only fallback if not auth error
        if hasattr(e, "status") and e.status in {401, 403}:
            raise
        logger.warning(
            "Falling back to v1 expression-docs endpoint due to: %s", str(e)
        )
        try:
            res = await dataflow_routes.get_magic_etl_tile_documentation_v1(
                auth=self.auth,
                context=context,
            )
            if return_raw:
                return res
            return res.response
        except Exception as e2:  # noqa: BLE001
            raise RuntimeError(  # noqa: B904
                f"Both v2 and v1 expression-docs endpoints failed: v2 error: {e}, v1 error: {e2}"
            )