modules
Top-level package for Boltworks.
boltworks
¶
Main module.
cli
special
¶
argparse_decorator
¶
argparse_command(argparser=None, echo_back=True, do_ack=True, automagic=False)
¶
This is the primary method for this feature. It is a decorator that you put onto your Slack Command Method to parse the arguments in the command and pass them to your function.
You can either construct an argparse.ArgumentParser yourself to do the work (better) Or set automagic=True to have a simple ArgumentParser constructed for you based on your method signature (faster)
Any parameter names in your method signature which Slack would usually inject (eg 'args' or 'say' or 'logger' etc) will be passed through.
1 2 3 4 5 6 7 8 9 10 | |
can be called as /dothething cc dd --a 1 2 3 -b 5
In automagic mode, to have a parameter be called as --foo, make it KeywordOnly by placing it after a single asterisk. Otherwise the argument name is silent
1 2 3 4 | |
/dotheotherthing test 1 2 3 --d 6 --c 5
(You are allowed to use Automagic together with an ArgumentParser as well to fill in an extra arg not handled by the parser)
In both cases, the argparser library will always insert a --help parameter, though it will be better documented if you build your own ArgumentParser.
It's recommended when you configure the Command in Slack to configure the usage string as [--help] to let people know they can do that.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
argparser |
Optional[ArgumentParser] |
An optional argparse.ArgumentParser object with which to parse the arguments |
None |
echo_back |
bool |
description. Defaults to True. Calls 'say' to echo back the command executed to the user before executing it. |
True |
do_ack |
bool |
description. Defaults to True. Calls Slack's ack() method for you upon succesful parsing, so you don't have to do it yourself. |
True |
automagic |
bool |
description. Defaults to False. Enables automagic parsing of arguments based on your method signature. Only supported on Python>=3.9 |
False |
Exceptions:
| Type | Description |
|---|---|
ValueError |
various ValueErrors can be raised, generally at the time of creation of the argparse_handler |
Source code in boltworks/cli/argparse_decorator.py
def argparse_command(argparser:Optional[ArgumentParser]=None,echo_back=True,do_ack=True,automagic=False):
"""
This is the primary method for this feature. It is a decorator that you put onto your Slack Command Method to parse the arguments in the command and pass them to your function.
You can either construct an argparse.ArgumentParser yourself to do the work (better)
Or set automagic=True to have a simple ArgumentParser constructed for you based on your method signature (faster)
Any parameter names in your method signature which Slack would usually inject (eg 'args' or 'say' or 'logger' etc) will be passed through.
```
argparser = argparse.ArgumentParser()
argparser.add_argument("--a", nargs="*", type=float)
argparser.add_argument("-b", type=int)
argparser.add_argument("c")
argparser.add_argument("d", nargs="?", default="default")
@app.command("/dothething")
@argparse_command(parser)
def your_method_name(args:Args, say, a, b, c, d):
...
```
can be called as `/dothething cc dd --a 1 2 3 -b 5`
In automagic mode, to have a parameter be called as --foo, make it KeywordOnly by placing it after a single asterisk. Otherwise the argument name is silent
```
@app.command("/dotheotherthing")
@argparse_command(automagic=True)
def your_method_name(args:Args, say, a:str, b:list[int], *, c:int, d:int):
...
```
can be called as `/dotheotherthing test 1 2 3 --d 6 --c 5`
(You are allowed to use Automagic together with an ArgumentParser as well to fill in an extra arg not handled by the parser)
In both cases, the argparser library will always insert a `--help` parameter, though it will be better documented if you build your own ArgumentParser.
It's recommended when you configure the Command in Slack to configure the usage string as `[--help]` to let people know they can do that.
Args:
argparser (Optional[ArgumentParser]): An optional argparse.ArgumentParser object with which to parse the arguments
echo_back (bool, optional): _description_. Defaults to True. Calls 'say' to echo back the command executed to the user before executing it.
do_ack (bool, optional): _description_. Defaults to True. Calls Slack's ack() method for you upon succesful parsing, so you don't have to do it yourself.
automagic (bool, optional): _description_. Defaults to False. Enables automagic parsing of arguments based on your method signature. Only supported on Python>=3.9
Raises:
ValueError: various ValueErrors can be raised, generally at the time of creation of the argparse_handler
* If there are conflicts between slack-reserved argnames and your argparser names
* If you don't have automagic set to True and don't pass an ArgumentParser, or the ArgumentParser doesn't handle all your method's params
* If you use automagic, but don't sufficiently type-hint your parameters, or the types are too complicated etc.
"""
if not argparser:
if not automagic:
raise ValueError("If you don't pass in an argparser, you must set automagic to True")
argparser=ArgumentParser()
if automagic and sys.version_info < (3, 9):
raise ValueError("automagic mode is not supported on Python versions < 3.9")
argparser_dests=[a.dest for a in argparser._actions if a.dest !="help"]
argname_conflicts=[dest for dest in argparser_dests if dest in RESERVED_SLACK_ARGS]
if argname_conflicts:
raise ValueError(f"One or more dest param names in your argparser conflict with built in param names used by slack bolt: {', '.join(argname_conflicts)}\nTry setting the `dest` argument for those argparser actions to something else")
# if "args" in argparser_dests:
# raise ValueError(f"The 'args' param name is reserved for Slack, so you can't use it for an argparser argument name.\nTry setting `the` dest parameter for that action to something else")
def _mid_func(decorated_function):
midfunc_argparser=argparser
deco_sig=inspect.signature(decorated_function)
deco_argnames=[a.name for a in deco_sig.parameters.values()]
deco_varargs=[a.name for a in deco_sig.parameters.values() if a.kind==a.VAR_POSITIONAL]
if deco_varargs:
raise ValueError(f"There is a varargs `*{deco_varargs[0]}` in the method signature of the method you are decorating, which is unsupported. Please use a list instead")
deco_varkwargs=[a.name for a in deco_sig.parameters.values() if a.kind==a.VAR_KEYWORD]
if not deco_varkwargs: #if there is no **kwargs to handle extra arguments...
unhandled_arguments=[dest for dest in argparser_dests
if dest not in deco_argnames]
if unhandled_arguments:
raise ValueError(f"The argparser object you provided generates one or more arguments ({','.join(unhandled_arguments)}) which are unhandled by the method you are decorating, nor does your method does not have a **kwargs to absorb them")
deco_extra_args_in_deco=[a for a in deco_sig.parameters.values()
if a.kind is not a.VAR_KEYWORD
and a.name not in RESERVED_SLACK_ARGS
and a.name not in argparser_dests]
deco_extra_args_in_deco_without_defaults=[a for a in deco_extra_args_in_deco if a.default is a.empty]
if not automagic and deco_extra_args_in_deco_without_defaults:
raise ValueError(f"The method you are decorating has one or more parameters without defaults defined that your argparser is not set to fill:{','.join(deco_extra_args_in_deco)}")
if automagic and deco_extra_args_in_deco:
midfunc_argparser = automagically_add_args_to_argparser(decorated_function, midfunc_argparser, deco_extra_args_in_deco)
def _inner_func_to_return_to_slack(args:Args):
def fn(message,file=None):
safe_post_in_blockquotes(args.respond,message)
args.ack()
innerfunc_argparser=copy.deepcopy(midfunc_argparser)#so we can have _print_message be unique to each call
innerfunc_argparser.formatter_class=SlackArgParseFormatter
innerfunc_argparser._print_message=fn
if args.command:
command_base=args.command['command'].replace("*","") if 'command' in args.command else '<cmd>'
command_args=shlex.split(args.command['text'].replace("*","")) if 'text' in args.command else []
elif args.message:
split_message=shlex.split(args.message['text'].replace("*",""))
command_base=split_message[0]
command_args=split_message[1:]
else: raise Exception("This class is only to be used with @command or @message")
# unfortunately it's not possible to perfectly strip out the markup formatting so for now we are just stripping out asterisks
# whenever they add 'blocks' to the command response, we can parse that to get the exact plaintext representation
# see also https://stackoverflow.com/a/70627214/10773089
innerfunc_argparser.prog=command_base
#maybe add more verbose explanation if you use `--var`` when it should have been `var`, or vice versa
parsed_params=vars(innerfunc_argparser.parse_args(command_args))
if do_ack:
args.ack()
if echo_back:
args.say(text=f"*{command_base} {(shlex.join(command_args))}* \t(run by <@{args.context['user_id']}>)")
available_slackvars_to_pass={'args':args,**vars(args)}
if deco_varkwargs:#if there's a varkwargs then just pass everything
slackvars_to_pass=available_slackvars_to_pass
else:
slackvars_to_pass={k:v for k,v in available_slackvars_to_pass.items() if k in deco_argnames}
return decorated_function(**slackvars_to_pass,**parsed_params)
return _inner_func_to_return_to_slack
#on init, it will return a function that does all the above stuff when run
return _mid_func
gui
special
¶
treenodeui
¶
ButtonChildContainer (ChildNodeContainer)
¶
The most basic child node container, this represents a button
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
ChildNodeContainer |
_type_ |
description |
required |
Source code in boltworks/gui/treenodeui.py
class ButtonChildContainer(ChildNodeContainer):
"""The most basic child node container, this represents a button
Args:
ChildNodeContainer (_type_): _description_
"""
def __init__(self,child_nodes:list[TreeNode],expand_button_format_string:str=NAMELESS_FMT_STR_EXPAND,collapse_button_format_string:str=NAMELESS_FMT_STR_COLLAPSE,static_button_text:Optional[str]=None,child_pageination:int=10):
self.child_nodes=child_nodes
self.expand_button_format_string=expand_button_format_string if not static_button_text else static_button_text
self.collapse_button_format_string = collapse_button_format_string if not static_button_text else static_button_text
self.child_pageination=child_pageination
def format_container(self, rootkey:str,pointer_to_container:_ExpandPointer, child_already_selected:int=-1) -> InteractiveElement:
if child_already_selected == -1:#if not already selected then it should be an expand button
return TreeNodeUI._button_to_replace_block(rootkey=rootkey,
expandpointer=pointer_to_container.append(0) #appending 0 to point to first node contained within this button
,button_text=self.expand_button_format_string.format(len(self.child_nodes)))
else:
return TreeNodeUI._button_to_replace_block(rootkey=rootkey,expandpointer=pointer_to_container[:-1] #slicing off the last one to collapse this container and only show it's containing node
,button_text=self.collapse_button_format_string.format(len(self.child_nodes)),style="danger")
@staticmethod
def forJsonDetails(jsonlike:list|dict,name:str="details",pageination=15,optimize_blocks=True):
children,numchildren=_jsonlike_to_treenode_and_truenum_children(jsonlike,optimize_blocks=optimize_blocks,_level=0,pageination=pageination)
return ButtonChildContainer(
children,
expand_button_format_string=name.format(numchildren) if name else NAMELESS_FMT_STR_EXPAND.format(numchildren),
collapse_button_format_string=name.format(numchildren) if name else NAMELESS_FMT_STR_COLLAPSE.format(numchildren),
child_pageination=pageination)
TreeNode
¶
The basic building block of this UI library, a Node has it's own Blocks, under formatblocks, and optionally, one or more child_containers containing one or more child nodes which can be expanded
If there is only one childNodeContainer it will by default be placed on the side of the first formatblock
If there is more than one, they will by default be placed on their own row, after the formatblocks
This behavior can be overriden with the first_child_container_on_side param
Source code in boltworks/gui/treenodeui.py
class TreeNode:
"""The basic building block of this UI library, a Node has it's own Blocks, under formatblocks, and optionally, one or more child_containers containing one or more child nodes which can be expanded
If there is only one childNodeContainer it will by default be placed on the side of the first formatblock
If there is more than one, they will by default be placed on their own row, after the formatblocks
This behavior can be overriden with the `first_child_container_on_side` param
"""
def __init__(
self,
formatblocks:Union[list[Block],Block,str],
children_containers:ChildNodeContainer|ChildNodeMenuContainer|list[ChildNodeContainer|ChildNodeMenuContainer]|None=None,
first_child_container_on_side:Optional[bool]=None,
auto_expand_children_if_only_one=False):
"""_summary_
Args:
formatblocks (Union[list[Block],Block,str]): The formatting blocks for the node
children_containers (ChildNodeContainer | ChildNodeMenuContainer | list[ChildNodeContainer | ChildNodeMenuContainer], optional): The whole point of the UI: these are interactive elements which if clicked/selected, will expand their child nodes underneath the node
first_child_container_on_side (bool, optional): This overrides the default behavior, which is to put the first childNodeContainer on the side of the first formatting block IF it is the only childNodeContainer
auto_expand_children_if_only_one (bool, optional): This option is deprecated and may be removed. It forces expand for any child that itself only has a single child node. Defaults to False.
"""
self.formatblocks=formatblocks
self.children_containers=children_containers if isinstance(children_containers,list) else [children_containers] if children_containers else []
if first_child_container_on_side is not None:
self.first_child_container_on_side=first_child_container_on_side # override
else: self.first_child_container_on_side = len(self.children_containers)==1 # default behavior for side placement
self.auto_expand_children_if_only_one=auto_expand_children_if_only_one #delete?
def __repr__(self):
toreturn=f"{type(self)} "
if self.children_containers:
toreturn+=f"({len(self.children_containers)} children_containers, total grandchildren_nodes {sum(len(cc.child_nodes) for cc in self.children_containers)}) "
toreturn+=": "
if isinstance(self.formatblocks,str): toreturn+=self.formatblocks
elif isinstance(self.formatblocks,SectionBlock): toreturn+=self.formatblocks.text.text
elif isinstance(self.formatblocks,list): toreturn+="\n".join([b.text.text if isinstance(b,SectionBlock) and b.text else b.to_dict() for b in self.formatblocks])
return toreturn
def text_formatting_as_str(n): #best try basis, not guaranteed to return correctly
if isinstance(n.formatblocks,str): return n.formatblocks
elif isinstance(n.formatblocks,SectionBlock): return n.formatblocks.text.text
elif isinstance(n.formatblocks,list): return "\n".join([fb.text.text for fb in n.formatblocks if 'text' in fb.attributes])
else: return "" #or raise exception??
@staticmethod
def withSimpleSideButton(formatblocks:Union[list[Block],Block,str],children:list[TreeNode]=[],expand_button_format_string:str=NAMELESS_FMT_STR_EXPAND,collapse_button_format_string:str=NAMELESS_FMT_STR_COLLAPSE,child_pageination:int=10)->TreeNode:
"""A simple way to get a basic node with a side button expanding its children
Args:
formatblocks (Union[list[Block],Block,str]): the formatting blocks for the node itself
children (list[TreeNode], optional): any children for the node
expand_button_format_string (str, optional): a string for the side button if any, when it is not yet expanded, if {} is in the string, it will be replaced with the number of children
collapse_button_format_string (str, optional): a string for the side button if any, when it is already expanded, if {} is in the string, it will be replaced with the number of children
child_pageination (int, optional): _description_. Defaults to 10.
Returns:
TreeNode: the treenode requested
"""
return TreeNode(formatblocks,[ButtonChildContainer(children,expand_button_format_string,collapse_button_format_string,child_pageination=child_pageination)] if children else [])
@staticmethod
def fromJson(formatblocks:Union[str,Block,list[Block]],jsonlike:Union[list,dict],pageination=15,optimize_blocks=True):
return TreeNode(formatblocks,[ButtonChildContainer.forJsonDetails(jsonlike,"details",pageination,optimize_blocks)])
__init__(self, formatblocks, children_containers=None, first_child_container_on_side=None, auto_expand_children_if_only_one=False)
special
¶
summary
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
formatblocks |
Union[list[Block],Block,str] |
The formatting blocks for the node |
required |
children_containers |
ChildNodeContainer | ChildNodeMenuContainer | list[ChildNodeContainer | ChildNodeMenuContainer] |
The whole point of the UI: these are interactive elements which if clicked/selected, will expand their child nodes underneath the node |
None |
first_child_container_on_side |
bool |
This overrides the default behavior, which is to put the first childNodeContainer on the side of the first formatting block IF it is the only childNodeContainer |
None |
auto_expand_children_if_only_one |
bool |
This option is deprecated and may be removed. It forces expand for any child that itself only has a single child node. Defaults to False. |
False |
Source code in boltworks/gui/treenodeui.py
def __init__(
self,
formatblocks:Union[list[Block],Block,str],
children_containers:ChildNodeContainer|ChildNodeMenuContainer|list[ChildNodeContainer|ChildNodeMenuContainer]|None=None,
first_child_container_on_side:Optional[bool]=None,
auto_expand_children_if_only_one=False):
"""_summary_
Args:
formatblocks (Union[list[Block],Block,str]): The formatting blocks for the node
children_containers (ChildNodeContainer | ChildNodeMenuContainer | list[ChildNodeContainer | ChildNodeMenuContainer], optional): The whole point of the UI: these are interactive elements which if clicked/selected, will expand their child nodes underneath the node
first_child_container_on_side (bool, optional): This overrides the default behavior, which is to put the first childNodeContainer on the side of the first formatting block IF it is the only childNodeContainer
auto_expand_children_if_only_one (bool, optional): This option is deprecated and may be removed. It forces expand for any child that itself only has a single child node. Defaults to False.
"""
self.formatblocks=formatblocks
self.children_containers=children_containers if isinstance(children_containers,list) else [children_containers] if children_containers else []
if first_child_container_on_side is not None:
self.first_child_container_on_side=first_child_container_on_side # override
else: self.first_child_container_on_side = len(self.children_containers)==1 # default behavior for side placement
self.auto_expand_children_if_only_one=auto_expand_children_if_only_one #delete?
withSimpleSideButton(formatblocks, children=[], expand_button_format_string='expand {}', collapse_button_format_string='collapse {}', child_pageination=10)
staticmethod
¶
A simple way to get a basic node with a side button expanding its children
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
formatblocks |
Union[list[Block],Block,str] |
the formatting blocks for the node itself |
required |
children |
list[TreeNode] |
any children for the node |
[] |
expand_button_format_string |
str |
a string for the side button if any, when it is not yet expanded, if {} is in the string, it will be replaced with the number of children |
'expand {}' |
collapse_button_format_string |
str |
a string for the side button if any, when it is already expanded, if {} is in the string, it will be replaced with the number of children |
'collapse {}' |
child_pageination |
int |
description. Defaults to 10. |
10 |
Returns:
| Type | Description |
|---|---|
TreeNode |
the treenode requested |
Source code in boltworks/gui/treenodeui.py
@staticmethod
def withSimpleSideButton(formatblocks:Union[list[Block],Block,str],children:list[TreeNode]=[],expand_button_format_string:str=NAMELESS_FMT_STR_EXPAND,collapse_button_format_string:str=NAMELESS_FMT_STR_COLLAPSE,child_pageination:int=10)->TreeNode:
"""A simple way to get a basic node with a side button expanding its children
Args:
formatblocks (Union[list[Block],Block,str]): the formatting blocks for the node itself
children (list[TreeNode], optional): any children for the node
expand_button_format_string (str, optional): a string for the side button if any, when it is not yet expanded, if {} is in the string, it will be replaced with the number of children
collapse_button_format_string (str, optional): a string for the side button if any, when it is already expanded, if {} is in the string, it will be replaced with the number of children
child_pageination (int, optional): _description_. Defaults to 10.
Returns:
TreeNode: the treenode requested
"""
return TreeNode(formatblocks,[ButtonChildContainer(children,expand_button_format_string,collapse_button_format_string,child_pageination=child_pageination)] if children else [])
TreeNodeUI
¶
Source code in boltworks/gui/treenodeui.py
class TreeNodeUI:
def __init__(self,app:App,kvstore:KVStore) -> None:
"""This is the managing class for the NodeUI, which handles posting nodes and then responding to InteractiveElements to expand/contract node children
Args:
app (App): A Slack Bolt App instance, for posting and registering actionhandlers
kvstore (_type_): a KVStore instance, for storing and looking up Nodes
"""
app.action(re.compile(f"{prefix_for_callback}.*"))(self._do_callback_action)
self.expiring_root_dict=ExpiringDict(max_age_seconds=120,max_len=20)
self.kvstore=kvstore #.namespaced(prefix_for_callback)
self._slack_chat_client=app.client
def post_single_node(self,post_callable_or_channel:str|Say|Respond,node:TreeNode,alt_text:str=None,expand_first:bool=False):
"""Posts a Single Node
Args:
post_callable_or_channel: either an instance of Respond or Say, or a channelid to post to
node (TreeNode): the node to post
alt_text (str, optional): for notifications that require a simple string
expand_first (bool, optional): if set to True, the Node will post with its first child container expanded [to its first menu option]
"""
say=Say(self._slack_chat_client,post_callable_or_channel) if isinstance(post_callable_or_channel,str) else post_callable_or_channel
rootkey=self._rootkey_from_treenode(node)
return say(text=alt_text or node.text_formatting_as_str(),
blocks=self._format_tree(rootkey,expand_first=expand_first),unfurl_links=False)
def post_treenodes(self,post_callable_or_channel:str|Say|Respond,treenodes:list[TreeNode],post_all_together:bool,global_header:str=None,*,message_if_none:str=None,expand_first_if_seperate=False,**other_global_tn_kwargs):
"""Posts multiple Nodes together
Args:
post_callable_or_channel (Union[PostMethod,str]): either an instance of Respond or Say, or a channelid to post to
treenodes (list[TreeNode]): the nodes to post
post_all_together (bool): if True, will collect the nodes under a parent node so they can all be collapsed together to save space
global_header (str, optional): if posting together, this will be the text of the parent node, otherwise just a header posted before the nodes
message_if_none (str, optional): optionally, provide a string to post if there are no nodes
expand_first_if_seperate (bool, optional): like expand_first for post_single_node, only effective if posting the blocks seperately
"""
say=Say(self._slack_chat_client,post_callable_or_channel) if isinstance(post_callable_or_channel,str) else post_callable_or_channel
if not treenodes:
if message_if_none: say(message_if_none)
return
if post_all_together:
num_treenodes=f" ({len(treenodes)}) " if len(treenodes)>1 else ""
alt_text=f"{global_header}: {num_treenodes} {treenodes[0].text_formatting_as_str()} "
say(text=alt_text,
blocks=self._format_tree(
self._rootkey_from_treenode(TreeNode.withSimpleSideButton(
formatblocks=global_header if global_header else [],
children=treenodes,
**other_global_tn_kwargs
)),expand_first=True
),unfurl_links=False)
else:
if global_header:
say(text=global_header)
for node in treenodes:
rootkey=self._rootkey_from_treenode(node)
alt_text=node.text_formatting_as_str()
say(text=alt_text,blocks=self._format_tree(rootkey,expand_first=expand_first_if_seperate),unfurl_links=False)
def _rootkey_from_treenode(self,node:TreeNode):
rootkey=str(uuid1())
self.kvstore[rootkey]=node
self.expiring_root_dict[rootkey]=node
return rootkey
def _get_root(self,rootkey:str)->TreeNode:
if rootkey in self.expiring_root_dict:
return self.expiring_root_dict[rootkey]
root=self.kvstore[rootkey]
self.expiring_root_dict[rootkey]=root
return root
def _do_callback_action(self,ack,action,respond):
# if logging.root.level<=logging.DEBUG:
# self.profiler.start()
ack()#find a way of threading that maybe?
callback_data=action['action_id'][len(prefix_for_callback):]
rootkey,expandpointer=self._deserialize_callback(callback_data)
value = action['selected_option']['value'] if 'selected_option' in action and 'value' in action['selected_option'] else None
if value:
if int(value) == -1: #deselect
expandpointer=expandpointer[:-1]
else:
expandpointer=expandpointer.extend([int(value),0])
blocks=self._format_tree(rootkey,expandpointer=expandpointer)
response=respond(replace_original=True,blocks=blocks)
if isinstance(response, WebhookResponse):
if response.status_code!=200:
respond(f"error in slack handling: {response.body}",replace_original=False)
print(f"{datetime.datetime.now()}: error in slack handling: {response.body}")
return response
# if logging.root.level<=logging.DEBUG:
# self.profiler.stop()
# print(self.profiler.output_text(unicode=True, color=True))
@staticmethod
def _button_to_replace_block(button_text:str,rootkey:str,expandpointer:_ExpandPointer,**format_options):
serialized_data=TreeNodeUI._serialize_callback(rootkey,expandpointer)
button=ButtonElement(
text=button_text,
action_id=f"{serialized_data}",
**format_options
)
return button
@staticmethod
def _serialize_callback(rootkey:str,expandpointer:_ExpandPointer)->str:
serialized_pointer=','.join([str(v) for v in expandpointer])
return f"{prefix_for_callback}{rootkey}^{serialized_pointer}"
@staticmethod
def _deserialize_callback(data:str)->Tuple[str,_ExpandPointer]:
rootkey,serialized_pointer=data.rsplit('^', 1)
pointerelems=serialized_pointer.split(',')
expandpointer=_ExpandPointer([int(p) for p in pointerelems])
return rootkey,expandpointer
def _format_tree(self,rootkey:str,*,expandpointer:_ExpandPointer=_ExpandPointer([0]),expand_first=False):
root=self._get_root(rootkey)
if expand_first and root.children_containers:
if isinstance(root.children_containers[0],ChildNodeContainer):
expandpointer=_ExpandPointer([0,0,0])
elif isinstance(root.children_containers[0],ChildNodeMenuContainer):
expandpointer=_ExpandPointer([0,0,0,0])
diminish_pageination_by=0
blocks_to_return = self._format_tree_recursive(
parentnodes=[root],
expandpointer=expandpointer,
ancestral_pointer=_ExpandPointer([]),
rootkey=rootkey,
parents_pagination=1,
diminish_pageination_by=0
)
if len(blocks_to_return)>50:#this could probably be made more efficient, but for now, this should suffice
while len(blocks_to_return)>49:
diminish_pageination_by+=1
blocks_to_return = self._format_tree_recursive(
parentnodes=[root],
expandpointer=expandpointer,
ancestral_pointer=_ExpandPointer([]),
rootkey=rootkey,
parents_pagination=1,
diminish_pageination_by=diminish_pageination_by
)
blocks_to_return.append(ContextBlock(elements=[MarkdownTextObject(text="(blocks were repaginated to avoid exceeding slack limits)")]))
return blocks_to_return
def _format_tree_recursive(self,
parentnodes:list[TreeNode],
expandpointer:_ExpandPointer,#:list[int], #could just make last one the start_at pointer and only go deeper if theres more
ancestral_pointer:_ExpandPointer,
rootkey:str,
parents_pagination:int=10,
diminish_pageination_by=0 #in case we exceed block limits, we can retry with some number here to recursively diminish the pageination
)->list[Block]:
parents_pagination=parents_pagination if not diminish_pageination_by else max(1,parents_pagination-diminish_pageination_by)
child_insert,remaining_expandpointer=expandpointer[0],expandpointer[1:]
num_parents=len(parentnodes)
start_at=self._startat(child_insert,parents_pagination)
end_at=self._endat(start_at,parents_pagination,num_parents)
blocks:list[Block]=[]
before_blocks=[blocks for number,node in enumerate(parentnodes[start_at:child_insert],start_at) for blocks in self._formatblock(node,ancestral_pointer.append(number),rootkey)]
blocks.extend(before_blocks)
blocks_for_pointed_node=self._formatblock(parentnodes[child_insert],ancestral_pointer.append(child_insert),rootkey,remaining_expandpointer)
blocks.extend(blocks_for_pointed_node)
if remaining_expandpointer:#if there are more nodes to expand in the pointer list
new_parent=parentnodes[child_insert]
children_containers=new_parent.children_containers
if remaining_expandpointer[0] > len(new_parent.children_containers): #transitional
remaining_expandpointer=_ExpandPointer([0,0])
if len(remaining_expandpointer)==1: #transitional
remaining_expandpointer=_ExpandPointer([0]).extend(remaining_expandpointer)
container_opened_index=remaining_expandpointer[0]
selected_container=children_containers[container_opened_index]
if isinstance(selected_container,ChildNodeContainer):
selected_container_blocks=self._format_tree_recursive(
parentnodes=selected_container.child_nodes if selected_container.child_nodes else [TreeNode("_(this pane is empty)_")],
parents_pagination=selected_container.child_pageination,
expandpointer=remaining_expandpointer[1:],
ancestral_pointer=ancestral_pointer.append(child_insert).append(remaining_expandpointer[0]),
rootkey=rootkey,
diminish_pageination_by=diminish_pageination_by
)
else:
assert isinstance(selected_container,ChildNodeMenuContainer)
selected_container_blocks=self._format_tree_recursive(
parentnodes=selected_container.child_nodes[remaining_expandpointer[1]] if selected_container.child_nodes and selected_container.child_nodes[remaining_expandpointer[1]] else [TreeNode("_(this pane is empty)_")],
parents_pagination=selected_container.child_pageination,
expandpointer=remaining_expandpointer[2:],#since this contains multiple lists of nodes, we need two pointer indexes to find the next node to show
ancestral_pointer=ancestral_pointer.append(child_insert).append(remaining_expandpointer[0]).append(remaining_expandpointer[1]),
rootkey=rootkey,
diminish_pageination_by=diminish_pageination_by
)
blocks.extend(selected_container_blocks)
after_blocks= [blocks for number,node in enumerate(parentnodes[child_insert+1:end_at],child_insert+1) for blocks in self._formatblock(node,ancestral_pointer.append(number),rootkey)]
blocks.extend(after_blocks)
navig_blocks=self._make_prev_next_buttons(usePrev=start_at>0,useNext=end_at<num_parents,
prev_callback=(rootkey,ancestral_pointer.append(start_at-parents_pagination if start_at-parents_pagination>0 else 0)),
next_callback=(rootkey,ancestral_pointer.append(start_at+parents_pagination))) #if startat+parents_per_page<num_parents else 0
if navig_blocks:
blocks.append(navig_blocks)
return blocks
def _formatblock(self,
node:TreeNode,
pointer_to_block:_ExpandPointer,
rootkey:str,
remaining_expandpointer:_ExpandPointer=None):
blocks:list[Block]=[]
if isinstance(node.formatblocks,list):
blocks.extend(node.formatblocks)
elif isinstance(node.formatblocks,Block):
blocks.append(node.formatblocks)
elif isinstance(node.formatblocks,str) and node.formatblocks:
blocks.append(simple_slack_block(node.formatblocks))
def format_nth_container(n):
if not remaining_expandpointer or remaining_expandpointer[0]!=n:
return node.children_containers[n].format_container(rootkey,pointer_to_block.append(n),-1)
else:
child_selected=remaining_expandpointer[1] if isinstance(node.children_containers[n],ChildNodeMenuContainer) and len(remaining_expandpointer)>1 else 0
return node.children_containers[n].format_container(rootkey,pointer_to_block.append(n),child_selected)
if node.children_containers:
if node.first_child_container_on_side and blocks and 'accessory' in blocks[0].attributes: #and not blocks[0].accessory
blocks[0].accessory=format_nth_container(0) # type: ignore
start_at=1
else: start_at=0
after_blocks_container_elements=(format_nth_container(n) for n in range(start_at,len(node.children_containers)))
blocks.extend(ActionsBlock(elements=buttons_chunk) for buttons_chunk in chunked(after_blocks_container_elements,ActionsBlock.elements_max_length))
return blocks
def _make_prev_next_buttons(self,usePrev:bool,useNext:bool,prev_callback:Tuple[str,_ExpandPointer],next_callback:Tuple[str,_ExpandPointer])->Optional[ActionsBlock]:
if (not usePrev) and (not useNext): return None
buttons=[]
if(usePrev): buttons.append(self._button_to_replace_block(":arrow_left:",*prev_callback))
if(useNext): buttons.append(self._button_to_replace_block(":arrow_right:",*next_callback))
navig_buttons=ActionsBlock(
elements=buttons)
return navig_buttons
def _startat(self,value, pageinate):
mod=value%pageinate
return value-mod if value>=pageinate else 0
def _endat(self,startat,pageinate,length):
return startat+pageinate if startat+pageinate<length else length
__init__(self, app, kvstore)
special
¶
This is the managing class for the NodeUI, which handles posting nodes and then responding to InteractiveElements to expand/contract node children
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
app |
App |
A Slack Bolt App instance, for posting and registering actionhandlers |
required |
kvstore |
_type_ |
a KVStore instance, for storing and looking up Nodes |
required |
Source code in boltworks/gui/treenodeui.py
def __init__(self,app:App,kvstore:KVStore) -> None:
"""This is the managing class for the NodeUI, which handles posting nodes and then responding to InteractiveElements to expand/contract node children
Args:
app (App): A Slack Bolt App instance, for posting and registering actionhandlers
kvstore (_type_): a KVStore instance, for storing and looking up Nodes
"""
app.action(re.compile(f"{prefix_for_callback}.*"))(self._do_callback_action)
self.expiring_root_dict=ExpiringDict(max_age_seconds=120,max_len=20)
self.kvstore=kvstore #.namespaced(prefix_for_callback)
self._slack_chat_client=app.client
post_single_node(self, post_callable_or_channel, node, alt_text=None, expand_first=False)
¶
Posts a Single Node
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
post_callable_or_channel |
str | Say | Respond |
either an instance of Respond or Say, or a channelid to post to |
required |
node |
TreeNode |
the node to post |
required |
alt_text |
str |
for notifications that require a simple string |
None |
expand_first |
bool |
if set to True, the Node will post with its first child container expanded [to its first menu option] |
False |
Source code in boltworks/gui/treenodeui.py
def post_single_node(self,post_callable_or_channel:str|Say|Respond,node:TreeNode,alt_text:str=None,expand_first:bool=False):
"""Posts a Single Node
Args:
post_callable_or_channel: either an instance of Respond or Say, or a channelid to post to
node (TreeNode): the node to post
alt_text (str, optional): for notifications that require a simple string
expand_first (bool, optional): if set to True, the Node will post with its first child container expanded [to its first menu option]
"""
say=Say(self._slack_chat_client,post_callable_or_channel) if isinstance(post_callable_or_channel,str) else post_callable_or_channel
rootkey=self._rootkey_from_treenode(node)
return say(text=alt_text or node.text_formatting_as_str(),
blocks=self._format_tree(rootkey,expand_first=expand_first),unfurl_links=False)
post_treenodes(self, post_callable_or_channel, treenodes, post_all_together, global_header=None, *, message_if_none=None, expand_first_if_seperate=False, **other_global_tn_kwargs)
¶
Posts multiple Nodes together
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
post_callable_or_channel |
Union[PostMethod,str] |
either an instance of Respond or Say, or a channelid to post to |
required |
treenodes |
list[TreeNode] |
the nodes to post |
required |
post_all_together |
bool |
if True, will collect the nodes under a parent node so they can all be collapsed together to save space |
required |
global_header |
str |
if posting together, this will be the text of the parent node, otherwise just a header posted before the nodes |
None |
message_if_none |
str |
optionally, provide a string to post if there are no nodes |
None |
expand_first_if_seperate |
bool |
like expand_first for post_single_node, only effective if posting the blocks seperately |
False |
Source code in boltworks/gui/treenodeui.py
def post_treenodes(self,post_callable_or_channel:str|Say|Respond,treenodes:list[TreeNode],post_all_together:bool,global_header:str=None,*,message_if_none:str=None,expand_first_if_seperate=False,**other_global_tn_kwargs):
"""Posts multiple Nodes together
Args:
post_callable_or_channel (Union[PostMethod,str]): either an instance of Respond or Say, or a channelid to post to
treenodes (list[TreeNode]): the nodes to post
post_all_together (bool): if True, will collect the nodes under a parent node so they can all be collapsed together to save space
global_header (str, optional): if posting together, this will be the text of the parent node, otherwise just a header posted before the nodes
message_if_none (str, optional): optionally, provide a string to post if there are no nodes
expand_first_if_seperate (bool, optional): like expand_first for post_single_node, only effective if posting the blocks seperately
"""
say=Say(self._slack_chat_client,post_callable_or_channel) if isinstance(post_callable_or_channel,str) else post_callable_or_channel
if not treenodes:
if message_if_none: say(message_if_none)
return
if post_all_together:
num_treenodes=f" ({len(treenodes)}) " if len(treenodes)>1 else ""
alt_text=f"{global_header}: {num_treenodes} {treenodes[0].text_formatting_as_str()} "
say(text=alt_text,
blocks=self._format_tree(
self._rootkey_from_treenode(TreeNode.withSimpleSideButton(
formatblocks=global_header if global_header else [],
children=treenodes,
**other_global_tn_kwargs
)),expand_first=True
),unfurl_links=False)
else:
if global_header:
say(text=global_header)
for node in treenodes:
rootkey=self._rootkey_from_treenode(node)
alt_text=node.text_formatting_as_str()
say(text=alt_text,blocks=self._format_tree(rootkey,expand_first=expand_first_if_seperate),unfurl_links=False)
slack_block_optimize_treenode(children)
¶
This method will try to reduce the number of formatting blocks used by a list of Treenode, to fit more blocks without pageinating or going over the slack limit of 50
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
children |
list[TreeNode] |
the nodes to optimize |
required |
Returns:
| Type | Description |
|---|---|
list[TreeNode] |
a fewer number of nodes, with some adjacent ones combined |
Source code in boltworks/gui/treenodeui.py
def slack_block_optimize_treenode(children:list[TreeNode])->list[TreeNode]:
"""This method will try to reduce the number of formatting blocks used by a list of Treenode, to fit more blocks without pageinating or going over the slack limit of 50
Args:
children (list[TreeNode]): the nodes to optimize
Returns:
list[TreeNode]: a fewer number of nodes, with some adjacent ones combined
"""
if not children: return []
out_children = []
combined_node_buffer = None #will always be either None or a TreeNode with a str as formattingblocks
for n in children:
if not _is_simple_text_block(n) or n.children_containers: #if the nodes have complicated blocks or children
if combined_node_buffer:
out_children.append(combined_node_buffer) #then just flush the buffer
combined_node_buffer=None
out_children.append(n) #and append this node
else: #a simple terminal block that we can combine with other ones to optimize block count:
if isinstance(n.formatblocks,SectionBlock): n.formatblocks=n.formatblocks.text.text if n.formatblocks.text else '' #convert it to using strings
elif isinstance(n.formatblocks,list) and isinstance(n.formatblocks[0],SectionBlock): n.formatblocks=n.formatblocks[0].text.text if n.formatblocks[0].text else ''
if not combined_node_buffer: #if no buffer, just start one
combined_node_buffer=n
else: #otherwise add the string to existing combined_tn, assuming it fits in char limits
if len(combined_node_buffer.formatblocks)+len(n.formatblocks)+2 < 3000:#slack char limits for a single section block
combined_node_buffer.formatblocks+="\n\n"+n.formatblocks
else: #if would be over char limits, we have no choice but to flush the buffer and restart
if combined_node_buffer: out_children.append(combined_node_buffer)
combined_node_buffer=n #set this block as new buffer
if combined_node_buffer: out_children.append(combined_node_buffer) #flush the buffer one last time
return out_children
helper
special
¶
serializers
¶
SignedSerializer (Serializer)
¶
Source code in boltworks/helper/serializers.py
class SignedSerializer(Serializer):
def __init__(self,serializer:Serializer,symmetric_key,max_age:Union[int,None]=3600*24*90):
self._signer=itsdangerous.TimestampSigner(symmetric_key) if max_age else itsdangerous.Signer(symmetric_key)
self._max_age=max_age
self._serializer=serializer
def dumps(self,obj:Any):
serialized=self._serializer.dumps(obj)
signed=self._signer.sign(serialized)
return signed
def loads(self,signed_serialized:bytes):
unsigned=self._signer.unsign(signed_value=signed_serialized,max_age=self._max_age) if isinstance(self._signer,itsdangerous.TimestampSigner) else self._signer.unsign(signed_value=signed_serialized)
return self._serializer.loads(unsigned)
def __getstate__(self):
raise Exception("serializing this class is not allowed, for security reasons")
__init__(self, serializer, symmetric_key, max_age=7776000)
special
¶
Initialize self. See help(type(self)) for accurate signature.
Source code in boltworks/helper/serializers.py
def __init__(self,serializer:Serializer,symmetric_key,max_age:Union[int,None]=3600*24*90):
self._signer=itsdangerous.TimestampSigner(symmetric_key) if max_age else itsdangerous.Signer(symmetric_key)
self._max_age=max_age
self._serializer=serializer