Skip to content

Instantly share code, notes, and snippets.

@JnBrymn-EB

JnBrymn-EB/BotSketch.ipynb

Last active October 24, 2019 17:58
Show Gist options
  • Save JnBrymn-EB/1b1db26c5116349d85f39ce400e5a8a4 to your computer and use it in GitHub Desktop.
Save JnBrymn-EB/1b1db26c5116349d85f39ce400e5a8a4 to your computer and use it in GitHub Desktop.
Download ZIP
Display the source blob
Display the rendered blob
Raw
BotSketch.ipynb
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# Goal 1 : Processor Decorators\n",
"Make nice decorators for functions and class methods that filter certain types of messages. Start with \"message contains string\" as example but then generalize.\n",
"\n",
"\n",
"## Decorators primer\n",
"\n",
"If you've never worked with Python decorators before, the whole idea is this:\n",
"\n",
"```python\n",
"@decorator_func\n",
"def thing(x):\n",
" print(x*x)\n",
"```\n",
"\n",
"The `decorator_func` is simply a function that takes a function and returns a function. The above is exactly equivalent to.\n",
"\n",
"```python\n",
"def thing(x):\n",
" print(x*x)\n",
" \n",
"thing = decorator_func(thing)\n",
"```\n",
"\n",
"Based on this simply idea you can create really complex and useful functionality. But it is often _confuuuusing_ to implement."
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Subgoal A: make a message filter decorator that works on methods or functions\n",
"Our bots main purpose is to appropriately route and process messages. Lets make a couple of message processors. A message processor is a callable that takes a single argument called `message`.\n",
"\n",
"Let's start with a simple `Message` class."
]
},
{
"cell_type": "code",
"execution_count": 11,
"metadata": {},
"outputs": [],
"source": [
"class Message():\n",
" \"\"\"I assume a simple model for a message for now.\"\"\"\n",
" def __init__(self, text, user, room, event):\n",
" self.text = text\n",
" self.user = user\n",
" self.room = room\n",
" self.event = event"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"Some message processors will be bare functions, others will be methods in classes so that they have access to state."
]
},
{
"cell_type": "code",
"execution_count": 12,
"metadata": {},
"outputs": [],
"source": [
"def message_processor(message):\n",
" print(f'function: {message.text}')\n",
" \n",
"class MessageProcessor():\n",
" def process(self, message):\n",
" print(f'class method: {message.text}')"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"A real message processor would do something a little more interesting, like respond in slack.\n",
"\n",
"Here is an example of how we could process some messages:"
]
},
{
"cell_type": "code",
"execution_count": 13,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"function: message about kittens\n",
"function: message about something else\n",
"class method: message about kittens\n",
"class method: message about something else\n"
]
}
],
"source": [
"message_processor(\n",
" Message(text='message about kittens', user='bob', room='general', event='message_sent')\n",
")\n",
"message_processor(\n",
" Message(text='message about something else', user='bob', room='general', event='message_sent')\n",
")\n",
"\n",
"MessageProcessor().process(\n",
" Message(text='message about kittens', user='bob', room='general', event='message_sent')\n",
")\n",
"MessageProcessor().process(\n",
" Message(text='message about something else', user='bob', room='general', event='message_sent')\n",
")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"Now lets create a rather stupid filter decorator that only allows messages about \"kittens\" to be processed and see how it works"
]
},
{
"cell_type": "code",
"execution_count": 14,
"metadata": {},
"outputs": [],
"source": [
"def contains_kittens(processor):\n",
" def wrapped_processor(message):\n",
" if 'kittens' in message.text:\n",
" return processor(message=message)\n",
" return wrapped_processor\n",
"\n",
"@contains_kittens\n",
"def message_processor(message):\n",
" print('function: ' + message.text)\n",
" \n",
"class MessageProcessor():\n",
" @contains_kittens\n",
" def process(self, message):\n",
" print('class method: ' + message.text)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"Let's process messages with the filtered functions"
]
},
{
"cell_type": "code",
"execution_count": 15,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"function: message about kittens\n"
]
}
],
"source": [
"message_processor(Message('message about kittens','bob','general','message_sent'))\n",
"message_processor(Message('message about something else','bob','general','message_sent'))"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"☝️Works!\n",
"\n",
"And now for the filtered methods"
]
},
{
"cell_type": "code",
"execution_count": 16,
"metadata": {},
"outputs": [
{
"ename": "TypeError",
"evalue": "wrapped_processor() takes 1 positional argument but 2 were given",
"output_type": "error",
"traceback": [
"\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
"\u001b[0;31mTypeError\u001b[0m Traceback (most recent call last)",
"\u001b[0;32m<ipython-input-16-3a1bcb8fd94b>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[0;32m----> 1\u001b[0;31m \u001b[0mMessageProcessor\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mprocess\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mMessage\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m'message about kittens'\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m'bob'\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m'general'\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m'message_sent'\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m 2\u001b[0m \u001b[0mMessageProcessor\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mprocess\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mMessage\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m'message about something else'\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m'bob'\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m'general'\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m'message_sent'\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
"\u001b[0;31mTypeError\u001b[0m: wrapped_processor() takes 1 positional argument but 2 were given"
]
}
],
"source": [
"MessageProcessor().process(Message('message about kittens','bob','general','message_sent'))\n",
"MessageProcessor().process(Message('message about something else','bob','general','message_sent'))"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"Whoops! The decorator only works with functions b/c methods first arg is always `self` - quick fix:"
]
},
{
"cell_type": "code",
"execution_count": 17,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"function: message about kittens\n",
"class method: message about kittens\n"
]
}
],
"source": [
"def contains_kittens(processor):\n",
" def wrapped_processor(*args):\n",
" if isinstance(args[0], Message):\n",
" message = args[0]\n",
" else:\n",
" message = args[1]\n",
" if 'kittens' in message.text:\n",
" return processor(*args)\n",
" return wrapped_processor\n",
"\n",
"@contains_kittens\n",
"def message_processor(message):\n",
" print('function: ' + message.text)\n",
" \n",
"class MessageProcessor():\n",
" @contains_kittens\n",
" def process(self, message):\n",
" print('class method: ' + message.text)\n",
" \n",
"message_processor(Message('message about kittens','bob','general','message_sent'))\n",
"message_processor(Message('message about something else','bob','general','message_sent'))\n",
"MessageProcessor().process(Message('message about kittens','bob','general','message_sent'))\n",
"MessageProcessor().process(Message('message about something else','bob','general','message_sent'))"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"That works and it correctly filters out non-kittens messages.\n",
"\n",
"\n",
"## Subgoal B: Make a generic message text filtering decorator \n",
"Writing this boilerplate for every possible string filter seems like overkill, so let's make a filter that takes any type of string. \n",
"\n",
"We'll use it like this:\n",
"```python\n",
"@contains_string('kittens')\n",
"def message_processor(message):\n",
" print('function: ' + message.text)\n",
"```\n",
"\n",
"Making this is confusing because we have to go one level deeper.\n",
"\n",
"* `contains_kittens` was a decorator - it took a processor and returned a wrapped pcoressor\n",
"* `contains_string` is a decorator **factory**\n",
" * it takes a string and returns **decorator**\n",
" * and _that_ decorator takes a processor and returns a wrapped processor"
]
},
{
"cell_type": "code",
"execution_count": 8,
"metadata": {},
"outputs": [],
"source": [
"def contains_string(string):\n",
" def decorator(processor):\n",
" def wrapped_processor(*args):\n",
" if isinstance(args[0], Message):\n",
" message = args[0]\n",
" else:\n",
" message = args[1]\n",
" if string in message.text:\n",
" return processor(*args)\n",
" return wrapped_processor\n",
" return decorator"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"Now let's use it."
]
},
{
"cell_type": "code",
"execution_count": 18,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"function: message about kittens\n",
"class method: message about kittens\n"
]
}
],
"source": [
"@contains_string('kittens')\n",
"def message_processor(message):\n",
" print('function: ' + message.text)\n",
" \n",
"class MessageProcessor():\n",
" @contains_string('kittens')\n",
" def process(self, message):\n",
" print('class method: ' + message.text)\n",
" \n",
"message_processor(Message('message about kittens','bob','general','message_sent'))\n",
"message_processor(Message('message about something else','bob','general','message_sent'))\n",
"MessageProcessor().process(Message('message about kittens','bob','general','message_sent'))\n",
"MessageProcessor().process(Message('message about something else','bob','general','message_sent'))"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### Subgoal C: Make generalized decorator _maker_\n",
"That was better. There was much less confusing boiler plate. But it only filters messages based upon the text they contain. It would be better to have some simple way of creating processor decorator based upon any filter we can think of.\n",
"\n",
"The idea is to _use_ a decorator that takes a filter and turns it into a processor decorator. Like this\n",
"\n",
"```python\n",
"@processor_filter\n",
"def contains_kittens(message):\n",
" return 'kittens' in message.user\n",
"```\n",
"\n",
"Then we could use it like this:\n",
"\n",
"```python\n",
"@contains_kittens\n",
"def message_processor(message):\n",
" do_special_kitten_things()\n",
"```\n",
"\n",
"Let's try to implement `process_decorator`:"
]
},
{
"cell_type": "code",
"execution_count": 34,
"metadata": {},
"outputs": [],
"source": [
"# this decorator is defined in the library and people never see it, \n",
"# they just use it to decorate functions\n",
"def processor_filter(filter_func):\n",
" def decorator(processor):\n",
" def wrapped_processor(*args):\n",
" if isinstance(args[0], Message):\n",
" message = args[0]\n",
" else:\n",
" message = args[1]\n",
" if filter_func(message=message):\n",
" return processor(*args)\n",
" return wrapped_processor\n",
" return decorator"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"Now in application code:\n",
"\n",
"1) Create the new decorator"
]
},
{
"cell_type": "code",
"execution_count": 35,
"metadata": {},
"outputs": [],
"source": [
"@processor_filter\n",
"def contains_kittens(message):\n",
" return 'kittens' in message.text"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"2) Attach it to some processors"
]
},
{
"cell_type": "code",
"execution_count": 36,
"metadata": {},
"outputs": [],
"source": [
"@contains_kittens\n",
"def message_processor(message):\n",
" print('function: ' + message.text)\n",
" \n",
"class MessageProcessor():\n",
" @contains_kittens\n",
" def process(self, message):\n",
" print('class method: ' + message.text)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"3) And at runtime, use it:"
]
},
{
"cell_type": "code",
"execution_count": 39,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"function: message about kittens\n",
"class method: message about kittens\n"
]
}
],
"source": [
"message_processor(Message('message about kittens','bob','general','message_sent'))\n",
"message_processor(Message('message about something else','bob','general','message_sent'))\n",
"MessageProcessor().process(Message('message about kittens','bob','general','message_sent'))\n",
"MessageProcessor().process(Message('message about something else','bob','general','message_sent'))"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"Here ☝️☝️☝️`contains_kittens` works the same as the one defined at the beginning, but compare the complexity of the code used to generate it now:\n",
"\n",
"**now**\n",
"```python\n",
"@processor_filter\n",
"def contains_kittens(message):\n",
" return 'kittens' in message.user\n",
"```\n",
"\n",
"VS. at the beginning when we created it from scratch.\n",
"\n",
"**before**\n",
"```python\n",
"def contains_kittens(func):\n",
" def wrapper(*args): # boilerplate\n",
" if isinstance(args[0], Message): # boilerplate\n",
" message = args[0] # boilerplate\n",
" else: # boilerplate\n",
" message = args[1] # boilerplate\n",
" if 'kittens' in message.text: # APPLICATION CODE!\n",
" return func(*args) # boilerplate\n",
" return wrapper\n",
"```\n",
"\n",
"This new code is also flexible. You can create any filter you want. Not just text filters:\n",
"\n",
"```python\n",
"@processor_filter\n",
"def from_bob(message):\n",
" return message.user == 'bob'\n",
"```\n",
"\n",
"Now I have a `from_bob` decorator that filters to include only messages from Bob."
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Subgoal D: Make generalized decorator maker _maker_\n",
"Whoa geez. I told you that things can get confusing to think about with decorators. But watch what I'm doing and I think you'll understand.\n",
"\n",
"The problem I'm dealing with here is that the above filters are not quite general enough. We might want to filter messages with `kittens` but we might also want to filter messages about something else. When using a `processor_filter` it now only takes 3 lines to define one of these filters, but wouldn't it be nice to use code like this to define a whole family of parameterized processor filters:\n",
"\n",
"```python\n",
"@processor_filter\n",
"def contains_string(string, message):\n",
" return string in message.user\n",
"```\n",
"\n",
"And then you could use it like \n",
"\n",
"```python\n",
"@contains_string('kittens')\n",
"def kitten_processor(message):\n",
" print(f'awe kittens! \"{message.text}\"')\n",
"```\n",
"\n",
"How do we modify the earlier definition of `processor_filter` so that the decorator it creates can take an argument? But also, how do we retain the previous behavior where it _doesn't_ take arguments?\n",
"\n",
"Here is the previous definition:\n",
"```python\n",
"def processor_filter(filter_func):\n",
" def decorator(processor):\n",
" def wrapped_processor(*args):\n",
" if isinstance(args[0], Message):\n",
" message = args[0]\n",
" else:\n",
" message = args[1]\n",
" if filter_func(message=message):\n",
" return processor(*args)\n",
" return wrapped_processor\n",
" return decorator\n",
"```\n",
"\n",
"But we know that the inner `decorator` function can now _sometimes_ take arguments `@message_about('kittens')` and sometimes not `@contains_kittens`. Let's deal with the case with the existing case of no arguments:\n",
"\n",
"```python\n",
"def processor_filter(filter_func):\n",
" def decorator(*dec_args, **dec_kwargs):\n",
" if dec_args and callable(dec_args[0]):\n",
" processor = dec_args[0]\n",
" def wrapped_processor(*args):\n",
" if isinstance(args[0], Message):\n",
" message = args[0]\n",
" else:\n",
" message = args[1]\n",
" if filter_func(message=message):\n",
" return processor(message)\n",
" return wrapped_processor\n",
" else:\n",
" # Do something clever here!!!\n",
"\n",
" return decorator\n",
"```\n",
"\n",
"Works just like before. But what to do when arguments are specified? Let's try this:"
]
},
{
"cell_type": "code",
"execution_count": 40,
"metadata": {},
"outputs": [],
"source": [
"# this decorator is defined in the library and people never see it, \n",
"# they just use it to decorate functions\n",
"from functools import partial\n",
"\n",
"def processor_filter(filter_func):\n",
" def decorator(*dec_args, **dec_kwargs):\n",
" if dec_args and callable(dec_args[0]):\n",
" processor = dec_args[0]\n",
" def wrapped_processor(*args):\n",
" if isinstance(args[0], Message):\n",
" message = args[0]\n",
" else:\n",
" message = args[1]\n",
" if filter_func(message=message):\n",
" return processor(*args)\n",
" return wrapped_processor\n",
" else:\n",
" new_filter_func = partial(filter_func, *dec_args, **dec_kwargs)\n",
" return processor_filter(new_filter_func)\n",
"\n",
" return decorator"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"And then in the application code:\n",
"\n",
"1) Create filters of both types"
]
},
{
"cell_type": "code",
"execution_count": 41,
"metadata": {},
"outputs": [],
"source": [
"@processor_filter\n",
"def from_user(user, message):\n",
" return message.user == user\n",
"\n",
"@processor_filter\n",
"def contains_kittens(message):\n",
" return 'kittens' in message.text"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"2) Attach filters. (Note: we can stack them!)"
]
},
{
"cell_type": "code",
"execution_count": 42,
"metadata": {},
"outputs": [],
"source": [
"@contains_kittens\n",
"@from_user('bob')\n",
"def message_processor(message):\n",
" print(f'function: \"{message.text}\" from {message.user}')\n",
" \n",
"class MessageProcessor():\n",
" @contains_kittens\n",
" @from_user('bob')\n",
" def process(self, message):\n",
" print(f'class method: \"{message.text}\" from {message.user}')\n",
" "
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"3) Use them. In this case we see messages from bob about kittens, messages from will about kittens, and messages from bob about something else. Because of these filters we should only process messages from bob about kittens."
]
},
{
"cell_type": "code",
"execution_count": 43,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"function: \"message about kittens\" from bob\n",
"class method: \"message about kittens\" from bob\n"
]
}
],
"source": [
"message_processor(Message('message about kittens','bob','general','message_sent'))\n",
"message_processor(Message('message about something else','bob','general','message_sent'))\n",
"message_processor(Message('message about kittens','will','general','message_sent'))\n",
"MessageProcessor().process(Message('message about kittens','bob','general','message_sent'))\n",
"MessageProcessor().process(Message('message about something else','bob','general','message_sent'))\n",
"MessageProcessor().process(Message('message about kittens','will','general','message_sent'))"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"Very nice and clean. Easy to make very different types of filters. *No boilerplate code.* You can also combine the filters.\n",
"\n",
"## Surprise Feature that Spontaneously Appeared!\n",
"\n",
"And look at this neat trick. You can make filters from other filters. Like so:"
]
},
{
"cell_type": "code",
"execution_count": 45,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"\"message about kittens\" from will\n"
]
}
],
"source": [
"from_will = from_user('will') # MIND BLOWN! 🤯\n",
"\n",
"@from_will\n",
"def message_processor(message):\n",
" print(f'\"{message.text}\" from {message.user}')\n",
" \n",
"message_processor(Message('message about kittens','bob','general','message_sent'))\n",
"message_processor(Message('message about kittens','will','general','message_sent'))"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"\"Holy cow that works?!\" \n",
"\n",
"Yes. 😏 It does."
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Final touches with `functools.wraps`\n",
"\n",
"In Python, functions have attributes - `__name__`, `__module__`, `__doc__`, and `__dict__`"
]
},
{
"cell_type": "code",
"execution_count": 19,
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"'a_rose_by_any_other_name'"
]
},
"execution_count": 19,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"def a_rose_by_any_other_name():\n",
" pass\n",
"\n",
"a_rose_by_any_other_name.__name__"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"but when you wrap them, the resulting functions metadata is that of the wrapper function"
]
},
{
"cell_type": "code",
"execution_count": 20,
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"'a_hamburer'"
]
},
"execution_count": 20,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"def transmogrify(func):\n",
" def a_hamburer():\n",
" return func()\n",
" return a_hamburer\n",
"\n",
"@transmogrify\n",
"def a_rose_by_any_other_name():\n",
" pass\n",
"\n",
"a_rose_by_any_other_name.__name__"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"you can fix this with `functools.wraps`"
]
},
{
"cell_type": "code",
"execution_count": 21,
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"'a_rose_by_any_other_name'"
]
},
"execution_count": 21,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"from functools import wraps\n",
"\n",
"def transmogrify(func):\n",
" @wraps(func) # <--- this line\n",
" def a_hamburer():\n",
" return func()\n",
" return a_hamburer\n",
"\n",
"@transmogrify\n",
"def a_rose_by_any_other_name():\n",
" pass\n",
"\n",
"a_rose_by_any_other_name.__name__"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## A note on _cleverness_\n",
"\n",
"> _Why, that's clever!_\n",
"\n",
"As a developer, this might be the biggest insult I have in my arsenal. When I say \"clever\" I mean \n",
"* witty, but in a way that attempts to publically demonstrate your intelligence\n",
"* complex for the sake of complexity\n",
"* \"magic\", in that the end result functionality is almost unexplainable\n",
"* dismissive of the people you work with and the company you work for because debugging this crap will be impossible\n",
"\n",
"Congratulations... you are _soooo_ smart.\n",
"\n",
"The code I've demonstrated here might very well be \"clever\". In the span of 18 lines I have\n",
"* decorators\n",
"* triply nested functions\n",
"* recursion\n",
"* function currying\n",
"* closures\n",
"* \\*args and \\*\\*kwargs (twice!)\n",
"\n",
"But... occasionally there comes a time when clever code can and should be used. Here is my criteria:\n",
"\n",
"* The cleverness accomplish something really helpful.\n",
"* There is no _less_ clever way of doing the same thing.\n",
"* The cleverness is well contained and won't spawn more cleverness or counter-cleverness."
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# Goal 2 - Bot Framework\n",
"It's easy to make a bot with one long horrendous imperative `if/else` method but this leads to unmaintainable, heavily coupled code.\n",
"\n",
"```python\n",
"def message_hook(message):\n",
" # greeting flow\n",
" if message.event = \"enter_room\" and message.room = \"general\":\n",
" print(f'Welcome to Penny University. Click <button> to tell use about yourself {message.user}')\n",
" else if message.event = \"welcome_button_click\":\n",
" print(f'{message.user}, here is a form to fill out: <interests>')\n",
" else if message.event = \"form_completed\":\n",
" print(f'{message.user} just completed their form! Let us rejoice.')\n",
"```\n",
"\n",
"This is already complicated, there are going to be more flows, bot commands, etc. And this completely ignores database access etc. This is going to be a nightmare.\n",
"\n",
"Instead, let's build a **composable** bot framework with a **declarative** construction that is easy to build, reason about, and extend.\n",
"\n",
"`main.py`\n",
"```python\n",
"bot = Bot(GreetingModule(), SearchModule(), CommandModule())\n",
"\n",
"def message_hook(message):\n",
" bot(message)\n",
"```\n",
"-------------\n",
"`greeting.py`\n",
"```python\n",
"class GreetingModule():\n",
" @event_type('entered_room')\n",
" @in_room('general')\n",
" def greet_newcomer(self, message):\n",
" print(f'Welcome to Penny University. Click <button> to tell use about yourself {message.user}')\n",
" \n",
" @event_type('welcome_button_click')\n",
" def send_out_form(self, message):\n",
" print(f'{message.user}, here is a form to fill out: <interests>')\n",
" \n",
" @event_type('form_completed')\n",
" def rejoice(self, message):\n",
" print(f'{message.user} just completed their form! Let us rejoice.')\n",
"```"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Bot framework"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"Let's create the bot framework ... _in its entirety_ 🤯 "
]
},
{
"cell_type": "code",
"execution_count": 46,
"metadata": {},
"outputs": [],
"source": [
"from functools import partial\n",
"\n",
"class Bot():\n",
" def __init__(self, *message_processors):\n",
" self.message_processors = message_processors\n",
" \n",
" def __call__(self, message):\n",
" for message_processor in self.message_processors:\n",
" message_processor(message)\n",
" \n",
"class BotModule():\n",
" def __call__(self, message):\n",
" member_names = [member_name for member_name in dir(self) if member_name[:1] != '_']\n",
" members = [getattr(self, member_name) for member_name in member_names]\n",
" methods = [member for member in members if callable(member)]\n",
" for method in methods:\n",
" method(message) \n",
" \n",
"\n",
"def processor_filter(filter_func):\n",
" def decorator(*dec_args, **dec_kwargs):\n",
" if dec_args and callable(dec_args[0]):\n",
" processor = dec_args[0]\n",
" def wrapped_processor(*args):\n",
" if isinstance(args[0], Message):\n",
" message = args[0]\n",
" else:\n",
" message = args[1]\n",
" if filter_func(message=message):\n",
" return processor(*args)\n",
" return wrapped_processor\n",
" else:\n",
" new_filter_func = partial(filter_func, *dec_args, **dec_kwargs)\n",
" return processor_filter(new_filter_func)\n",
" return decorator "
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Build a Bot"
]
},
{
"cell_type": "code",
"execution_count": 47,
"metadata": {},
"outputs": [],
"source": [
"@processor_filter\n",
"def event_type(event_name, message):\n",
" return message.event == event_name\n",
"\n",
"@processor_filter\n",
"def in_room(room, message):\n",
" return message.room == room\n",
"\n",
"class GreetingModule(BotModule):\n",
" @event_type('entered_room')\n",
" @in_room('general')\n",
" def greet_newcomer(self, message):\n",
" print(f'Welcome to Penny University. Click <button> to tell use about yourself {message.user}')\n",
" \n",
" @event_type('welcome_button_click')\n",
" def send_out_form(self, message):\n",
" print(f'{message.user}, here is a form to fill out: <interests>')\n",
" \n",
" @event_type('form_completed')\n",
" def rejoice(self, message):\n",
" print(f'{message.user} just completed their form! Their interests include: {message.text}')"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Create and then user the bot"
]
},
{
"cell_type": "code",
"execution_count": 48,
"metadata": {},
"outputs": [],
"source": [
"bot = Bot(GreetingModule())"
]
},
{
"cell_type": "code",
"execution_count": 49,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"bob: I have arrived (in general; event_type=entered_room)\n",
"Welcome to Penny University. Click <button> to tell use about yourself bob\n",
"\n",
"bob: - (in general; event_type=welcome_button_click)\n",
"bob, here is a form to fill out: <interests>\n",
"\n",
"bob: data science, neurobiology, long walks alng the beach (in general; event_type=form_completed)\n",
"bob just completed their form! Their interests include: data science, neurobiology, long walks alng the beach\n",
"\n"
]
}
],
"source": [
"messages = [\n",
" Message('I have arrived','bob','general','entered_room'),\n",
" Message('-','bob','general','welcome_button_click'),\n",
" Message('data science, neurobiology, long walks alng the beach','bob','general','form_completed'),\n",
"]\n",
" \n",
"for message in messages:\n",
" print(f'{message.user}: {message.text} (in {message.room}; event_type={message.event})')\n",
" bot(message)\n",
" print()\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"Works as advertised! ✅\n",
"\n",
"\n",
"[<sub>This gist can be found here.</sub>](https://gist.github.com/JnBrymn-EB/1b1db26c5116349d85f39ce400e5a8a4)"
]
}
],
"metadata": {
"kernelspec": {
"display_name": "Python 3",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.7.4"
}
},
"nbformat": 4,
"nbformat_minor": 2
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment