Source code for deephaven_enterprise.edge_acl
#
# Copyright (c) 2016-2024 Deephaven Data Labs and Patent Pending
#
"""This module supports the creation of access control lists (ACLs) for the results of Persistent Queries.
For more information on ACLs and filter generators, see the Deephaven documentation
https://deephaven.io/enterprise/docs/sys-admin/permissions/acls/#persistent-query-acls.
"""
from __future__ import annotations
from typing import Any, Union
import jpy
from deephaven import DHError
from deephaven._wrapper import JObjectWrapper, unwrap
from deephaven.jcompat import j_array_list, to_sequence
from deephaven.table import RollupTable, Table, TreeTable
from deephaven_enterprise.pivot.pivot_table import PivotTable
_JEdgeAclProvider = jpy.get_type("io.deephaven.enterprise.acl.EdgeAclProvider")
_JEdgeAclProviderBuilder: Any = jpy.get_type(
"io.deephaven.enterprise.acl.EdgeAclProvider$Builder"
)
JAclFilterGenerator: Any = jpy.get_type(
"io.deephaven.enterprise.acl.AclFilterGenerator"
)
[docs]
class EdgeAclProvider:
"""An EdgeAclProvider is used to provide access control for the results of Persistent Queries based on the
requesting user at request time.
Note that it should not be instantiated directly, but rather through the :meth:`EdgeAclProviderBuilder.build` method.
"""
[docs]
@staticmethod
def builder() -> EdgeAclProviderBuilder:
"""Creates a builder to construct an EdgeAclProvider
Returns:
EdgeAclProviderBuilder
"""
return EdgeAclProviderBuilder(_JEdgeAclProvider.builder())
[docs]
@staticmethod
def add_object_acl(group: str, value: Union[jpy.JType, Any]) -> None:
"""Adds visibility for the specified object to the specified group.
Args:
group (str): the group to add visibility for
value (Union[jpy.JType, Any]): the object to add visibility for
"""
_JEdgeAclProvider.addAcl(group, unwrap(value))
def __init__(self, j_provider: jpy.JType):
self.j_provider = jpy.cast(j_provider, _JEdgeAclProvider)
if self.j_provider is None:
raise DHError(
message="j_provider type is not io.deephaven.enterprise.database.Database"
)
[docs]
def apply_to(self, table: Union[Table, RollupTable, TreeTable, PivotTable]):
"""Attaches this ACL Provider to a table
Args:
table (Table): the table to attach to
Returns:
a new table with the ACLs attached
"""
if isinstance(table, RollupTable):
return RollupTable(
self.j_provider.applyTo(table.j_rollup_table),
aggs=table.aggs,
by=table.by,
include_constituents=table.include_constituents,
)
elif isinstance(table, TreeTable):
return TreeTable(
self.j_provider.applyTo(table.j_tree_table),
id_col=table.id_col,
parent_col=table.parent_col,
)
elif isinstance(table, PivotTable):
return PivotTable(self.j_provider.applyTo(table._j_pivot))
return Table(self.j_provider.applyTo(table.j_table))
[docs]
class EdgeAclProviderBuilder(JObjectWrapper):
"""A builder for constructing an EdgeAclProvider for access control on the result tables of Persistent Queries.
Note that it should not be instantiated directly, but rather through the static :meth:`EdgeAclProvider.builder`.
"""
j_object_type = _JEdgeAclProviderBuilder
@property
def j_object(self) -> jpy.JType:
return self.j_builder
def __init__(self, j_builder: _JEdgeAclProviderBuilder):
self.j_builder = j_builder
[docs]
def row_acl(self, group: str, acl: JAclFilterGenerator) -> EdgeAclProviderBuilder:
"""Adds a Row ACL for the specified group to the table with a filter generator. The filter generator is created
by calling one of the factory functions in :mod:`.acl_generator`. The Row Acl controls access to rows of a table.
Args:
group (str): the group the ACL applies to
acl (JAclFilterGenerator): the generator that will produce a filter for a given user
Returns:
self
"""
self.j_builder.rowAcl(group, acl)
return self
[docs]
def column_acl(
self, group: str, columns: Union[str, list[str]], acl: JAclFilterGenerator
) -> EdgeAclProviderBuilder:
"""Adds a Column ACL for the specified group and column(s) to the table with a filter generator. The filter
generator is created by calling one of the factory functions in :mod:`.acl_generator`. The Column ACL controls
access to specific values of columns of a table.
Args:
group (str): the group the ACL applies to
columns (Union[str, list[str]]): the column(s) the acl applies to
acl (JAclFilterGenerator): the generator that will produce a filter for a given user
Returns:
self
"""
columns = list(to_sequence(columns))
self.j_builder.columnAcl(group, j_array_list(columns), acl)
return self
[docs]
def build(self) -> EdgeAclProvider:
"""Constructs the completed :class:`EdgeAclProvider` object from the current state of this builder.
Returns:
EdgeAclProvider
"""
return EdgeAclProvider(self.j_builder.build())