Source code for operators.superset.disable_superset_user

from typing import Any, Dict, Optional

from airflow.exceptions import AirflowException
from airflow.models import BaseOperator
from airflow.utils.decorators import apply_defaults




class DisableSupersetUserOperator(BaseOperator):
    """Disable a Superset user (sets active=False) via the Superset REST API.

    :param username: The Superset username to disable.
    :type username: str
    :param conn_id: Airflow connection id pointing at the Superset instance.
        Defaults to ``superset_default``.
    :type conn_id: str
    :param base_url: Optional full URL override for the Superset API (e.g.
        ``"https://dap-tx.bead.example.com"``). When provided, overrides the
        connection's host/schema.
    :type base_url: str
    :param timeout: HTTP request timeout in seconds.
    :type timeout: int
    """

    template_fields = ("username", "conn_id", "base_url")
    ui_color = "#fce4ec"

    @apply_defaults
    def __init__(
        self,
        username: str,
        conn_id: str = "superset_default",
        base_url: Optional[str] = None,
        timeout: int = 30,
        *args,
        **kwargs,
    ):
        super().__init__(*args, **kwargs)
        self.username = username
        self.conn_id = conn_id
        self.base_url = base_url
        self.timeout = timeout



    def execute(self, context) -> Dict[str, Any]:
        from hooks.superset_hook import SupersetHook

        if not self.username:
            raise AirflowException("Missing required field: username")

        hook = SupersetHook(
            conn_id=self.conn_id, base_url=self.base_url, timeout=self.timeout
        )
        self.log.info("Disabling Superset user '%s'", self.username)
        result = hook.disable_user(self.username)
        self.log.info("Disable user result: %s", result)
        return result