Superset实现动态SQL查询功能

数据运维 2023-04-15 竹子爱熊猫手机阅读

使用自定义参数方式实现 superset 实现SQL动态查询 1、启用参数：config.py 设置"ENABLE_TEMPLATE_PROCESSING": True 2、当前superset v1.2版本支持的参数包括： {{ current_username() }} 当前登录用户名 {{ c

                        <p>使用自定义参数方式实现 superset 实现SQL动态查询</p>

1、启用参数：config.py 设置"ENABLE_TEMPLATE_PROCESSING": True

2、当前superset v1.2版本支持的参数包括：

{{ current_username() }} 当前登录用户名 {{ current_username(add_to_cache_keys=False) }} 不从缓存中获取登录用户名，默认从缓存获取 {{ current_user_id()}} 当前登录用户ID {{ current_user_id(add_to_cache_keys=False) }} 不从缓存中获取登录用户ID，默认从缓存获取 {{ url_param('custom_variable') }} url 参数，比如127.0.0.1：8001dashboard?abc=123，参数就是{{ url_param('abc') }} 结果就是123 {{ cache_key_wrapper() }} 还没有弄明白啥用 {{ filter_values("字段名") }} 获取dashboard filter_box组件对某个字段的筛选结果 {{ from_dttm }} 获取dashboard filter_box组件日期筛选的开始时间 {{ to_dttm }} 获取dashboard filter_box组件日期筛选的结束时间 {{ get_filters() }} 暂时没有弄明白

除此之外，还可以自定义参数，自定义参数方法：

①修改superset/jinja_context.py文件，修改三个地方：

regex = re.compile( r"{{.(" r"current_user_id(.)|" r"current_username(.)|" r"current_userroles(.)|" r"isadmin(.)|" r"cache_key_wrapper(.)|" r"url_param(.)" r").}}" )

↑↑↑↑注意此处的 current_userroles 和 isadmin 是我自定义的，源文件没有

def current_user_id(self, add_to_cache_keys: bool = True) -> Optional[int]: """ Return the user ID of the user who is currently logged in.

    :param add_to_cache_keys: Whether the value should be included in the cache key
    :returns: The user ID
    """

    if hasattr(g, "user") and g.user:
        if add_to_cache_keys:
            self.cache_key_wrapper(g.user.get_id())
        return g.user.get_id()
    return None

def current_username(self, add_to_cache_keys: bool = True) -&gt; Optional[str]:
    """
    Return the username of the user who is currently logged in.

    :param add_to_cache_keys: Whether the value should be included in the cache key
    :returns: The username
    """

    if g.user and hasattr(g.user, "username"):
        if add_to_cache_keys:
            self.cache_key_wrapper(g.user.username)
        return g.user.username
    return None
def current_userroles(self, add_to_cache_keys: bool = True) -&gt; Optional[str]:
    """
    Return the roles of the user who is currently logged in.

    :param add_to_cache_keys: Whether the value should be included in the cache key
    :returns: The userroles
    """

    if g.user and hasattr(g.user, "roles"):
        if add_to_cache_keys:
            user_roles = "/".join([role.name.lower() for role in list(g.user.roles)])
            self.cache_key_wrapper(user_roles)
            print(user_roles)
            return user_roles
            """admin in user_roles"""
    return None

def isadmin(self, add_to_cache_keys: bool = True) -&gt; Optional[str]:
    """
    Return the roles of the user who is currently logged in.

    :param add_to_cache_keys: Whether the value should be included in the cache key
    :returns: The userroles
    """

    if g.user and hasattr(g.user, "roles"):
        if add_to_cache_keys:
            user_roles = [role.name.lower() for role in list(g.user.roles)]
            return "admin" in user_roles
    return None

↑↑↑↑仿照系统自带的 current_username 编造自己的函数，我写了current_userroles 和 isadmin

class JinjaTemplateProcessor(BaseTemplateProcessor): def set_context(self, kwargs: Any) -> None: super().set_context(kwargs) extra_cache = ExtraCache(self._extra_cache_keys) self._context.update( { "url_param": partial(safe_proxy, extra_cache.url_param), "current_user_id": partial(safe_proxy, extra_cache.current_user_id), "current_username": partial(safe_proxy, extra_cache.current_username), "current_userroles": partial(safe_proxy, extra_cache.current_userroles), "isadmin": partial(safe_proxy, extra_cache.isadmin), "cache_key_wrapper": partial(safe_proxy, extra_cache.cache_key_wrapper), "filter_values": partial(safe_proxy, filter_values), } )

↑↑↑↑仿照系统自带的 current_username 编造自己的函数，我写了current_userroles 和 isadmin

就是这3个地方，但是注意，自己在第二步早的函数，返回值必须是：

ALLOWED_TYPES = ( NONE_TYPE, "bool", "str", "unicode", "int", "long", "float", "list", "dict", "tuple", "set", )

否则会提示错误，或者自己修改这个types，我是转换，比如上面那个g.user.roles 返回的结果就不是上面类型，导致我一直不成功，最后修改了下，才可以

3、判断是否自定义成功：

在superset sql lab中执行如下代码，如果能被解析，就说明成功

4、应用案例：

在dataset里面，动态访问数据源，数据源添加where语句：select * from sales where salesname =' {{current_username()}}'

dashboard里面，通过获取筛选器的结果，然后获取其他表应当显示的数据范围：

select DATE,risktype,sum(num) as num from (SELECT date , customerid,product,risktype ,count(*) as num from v_superset_forecast_risk group by date , customerid,product,risktype ) a join (select distinct customer_code,product from v_superset_access where name='{{ current_username() }}' )access on a.customerid=access.customer_code and a.product=access.product and DATE_FORMAT(date,'%Y-%m')> DATE_FORMAT(date_sub(STR_TO_DATE(concat( {{ "'" + "', '".join(filter_values('yearmonthend')) + "'" }},'-01'), '%Y-%m-%d'), interval 12 month),'%Y-%m') and DATE_FORMAT(date,'%Y-%m')<={{ "'" + "', '".join(filter_values('yearmonthend')) + "'" }} group by DATE,risktype

因为sql里面可以使用jinja 表达式，比如判断筛选当前没有筛选的时候，获取什么数据