diff --git a/solutions/object_oriented_design/call_center/__init__.py b/solutions/object_oriented_design/call_center/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/solutions/object_oriented_design/call_center/call_center.ipynb b/solutions/object_oriented_design/call_center/call_center.ipynb new file mode 100644 index 0000000..deac342 --- /dev/null +++ b/solutions/object_oriented_design/call_center/call_center.ipynb @@ -0,0 +1,206 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "This notebook was prepared by [Donne Martin](https://github.com/donnemartin). Source and license info is on [GitHub](https://github.com/donnemartin/system-design-primer-primer)." + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "# Design a call center" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Constraints and assumptions\n", + "\n", + "* What levels of employees are in the call center?\n", + " * Operator, supervisor, director\n", + "* Can we assume operators always get the initial calls?\n", + " * Yes\n", + "* If there is no free operators or the operator can't handle the call, does the call go to the supervisors?\n", + " * Yes\n", + "* If there is no free supervisors or the supervisor can't handle the call, does the call go to the directors?\n", + " * Yes\n", + "* Can we assume the directors can handle all calls?\n", + " * Yes\n", + "* What happens if nobody can answer the call?\n", + " * It gets queued\n", + "* Do we need to handle 'VIP' calls where we put someone to the front of the line?\n", + " * No\n", + "* Can we assume inputs are valid or do we have to validate them?\n", + " * Assume they're valid" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Solution" + ] + }, + { + "cell_type": "code", + "execution_count": 1, + "metadata": { + "collapsed": false + }, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "Overwriting call_center.py\n" + ] + } + ], + "source": [ + "%%writefile call_center.py\n", + "from abc import ABCMeta, abstractmethod\n", + "from collections import deque\n", + "from enum import Enum\n", + "\n", + "\n", + "class Rank(Enum):\n", + "\n", + " OPERATOR = 0\n", + " SUPERVISOR = 1\n", + " DIRECTOR = 2\n", + "\n", + "\n", + "class Employee(metaclass=ABCMeta):\n", + "\n", + " def __init__(self, employee_id, name, rank, call_center):\n", + " self.employee_id = employee_id\n", + " self.name = name\n", + " self.rank = rank\n", + " self.call = None\n", + " self.call_center = call_center\n", + "\n", + " def take_call(self, call):\n", + " \"\"\"Assume the employee will always successfully take the call.\"\"\"\n", + " self.call = call\n", + " self.call.employee = self\n", + " self.call.state = CallState.IN_PROGRESS\n", + "\n", + " def complete_call(self):\n", + " self.call.state = CallState.COMPLETE\n", + " self.call_center.notify_call_completed(self.call)\n", + "\n", + " @abstractmethod\n", + " def escalate_call(self):\n", + " pass\n", + "\n", + " def _escalate_call(self):\n", + " self.call.state = CallState.READY\n", + " call = self.call\n", + " self.call = None\n", + " self.call_center.notify_call_escalated(call)\n", + "\n", + "\n", + "class Operator(Employee):\n", + "\n", + " def __init__(self, employee_id, name):\n", + " super(Operator, self).__init__(employee_id, name, Rank.OPERATOR)\n", + "\n", + " def escalate_call(self):\n", + " self.call.level = Rank.SUPERVISOR\n", + " self._escalate_call()\n", + "\n", + "\n", + "class Supervisor(Employee):\n", + "\n", + " def __init__(self, employee_id, name):\n", + " super(Operator, self).__init__(employee_id, name, Rank.SUPERVISOR)\n", + "\n", + " def escalate_call(self):\n", + " self.call.level = Rank.DIRECTOR\n", + " self._escalate_call()\n", + "\n", + "\n", + "class Director(Employee):\n", + "\n", + " def __init__(self, employee_id, name):\n", + " super(Operator, self).__init__(employee_id, name, Rank.DIRECTOR)\n", + "\n", + " def escalate_call(self):\n", + " raise NotImplemented('Directors must be able to handle any call')\n", + "\n", + "\n", + "class CallState(Enum):\n", + "\n", + " READY = 0\n", + " IN_PROGRESS = 1\n", + " COMPLETE = 2\n", + "\n", + "\n", + "class Call(object):\n", + "\n", + " def __init__(self, rank):\n", + " self.state = CallState.READY\n", + " self.rank = rank\n", + " self.employee = None\n", + "\n", + "\n", + "class CallCenter(object):\n", + "\n", + " def __init__(self, operators, supervisors, directors):\n", + " self.operators = operators\n", + " self.supervisors = supervisors\n", + " self.directors = directors\n", + " self.queued_calls = deque()\n", + "\n", + " def dispatch_call(self, call):\n", + " if call.rank not in (Rank.OPERATOR, Rank.SUPERVISOR, Rank.DIRECTOR):\n", + " raise ValueError('Invalid call rank: {}'.format(call.rank))\n", + " employee = None\n", + " if call.rank == Rank.OPERATOR:\n", + " employee = self._dispatch_call(call, self.operators)\n", + " if call.rank == Rank.SUPERVISOR or employee is None:\n", + " employee = self._dispatch_call(call, self.supervisors)\n", + " if call.rank == Rank.DIRECTOR or employee is None:\n", + " employee = self._dispatch_call(call, self.directors)\n", + " if employee is None:\n", + " self.queued_calls.append(call)\n", + "\n", + " def _dispatch_call(self, call, employees):\n", + " for employee in employees:\n", + " if employee.call is None:\n", + " employee.take_call(call)\n", + " return employee\n", + " return None\n", + "\n", + " def notify_call_escalated(self, call): # ...\n", + " def notify_call_completed(self, call): # ...\n", + " def dispatch_queued_call_to_newly_freed_employee(self, call, employee): # ..." + ] + } + ], + "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.4.3" + } + }, + "nbformat": 4, + "nbformat_minor": 0 +} diff --git a/solutions/object_oriented_design/call_center/call_center.py b/solutions/object_oriented_design/call_center/call_center.py new file mode 100644 index 0000000..7bdd992 --- /dev/null +++ b/solutions/object_oriented_design/call_center/call_center.py @@ -0,0 +1,117 @@ +from abc import ABCMeta, abstractmethod +from collections import deque +from enum import Enum + + +class Rank(Enum): + + OPERATOR = 0 + SUPERVISOR = 1 + DIRECTOR = 2 + + +class Employee(metaclass=ABCMeta): + + def __init__(self, employee_id, name, rank, call_center): + self.employee_id = employee_id + self.name = name + self.rank = rank + self.call = None + self.call_center = call_center + + def take_call(self, call): + """Assume the employee will always successfully take the call.""" + self.call = call + self.call.employee = self + self.call.state = CallState.IN_PROGRESS + + def complete_call(self): + self.call.state = CallState.COMPLETE + self.call_center.notify_call_completed(self.call) + + @abstractmethod + def escalate_call(self): + pass + + def _escalate_call(self): + self.call.state = CallState.READY + call = self.call + self.call = None + self.call_center.notify_call_escalated(call) + + +class Operator(Employee): + + def __init__(self, employee_id, name): + super(Operator, self).__init__(employee_id, name, Rank.OPERATOR) + + def escalate_call(self): + self.call.level = Rank.SUPERVISOR + self._escalate_call() + + +class Supervisor(Employee): + + def __init__(self, employee_id, name): + super(Operator, self).__init__(employee_id, name, Rank.SUPERVISOR) + + def escalate_call(self): + self.call.level = Rank.DIRECTOR + self._escalate_call() + + +class Director(Employee): + + def __init__(self, employee_id, name): + super(Operator, self).__init__(employee_id, name, Rank.DIRECTOR) + + def escalate_call(self): + raise NotImplemented('Directors must be able to handle any call') + + +class CallState(Enum): + + READY = 0 + IN_PROGRESS = 1 + COMPLETE = 2 + + +class Call(object): + + def __init__(self, rank): + self.state = CallState.READY + self.rank = rank + self.employee = None + + +class CallCenter(object): + + def __init__(self, operators, supervisors, directors): + self.operators = operators + self.supervisors = supervisors + self.directors = directors + self.queued_calls = deque() + + def dispatch_call(self, call): + if call.rank not in (Rank.OPERATOR, Rank.SUPERVISOR, Rank.DIRECTOR): + raise ValueError('Invalid call rank: {}'.format(call.rank)) + employee = None + if call.rank == Rank.OPERATOR: + employee = self._dispatch_call(call, self.operators) + if call.rank == Rank.SUPERVISOR or employee is None: + employee = self._dispatch_call(call, self.supervisors) + if call.rank == Rank.DIRECTOR or employee is None: + employee = self._dispatch_call(call, self.directors) + if employee is None: + self.queued_calls.append(call) + + def _dispatch_call(self, call, employees): + for employee in employees: + if employee.call is None: + employee.take_call(call) + return employee + return None + + def notify_call_escalated(self, call): # ... + def notify_call_completed(self, call): # ... + def dispatch_queued_call_to_newly_freed_employee(self, call, employee): # ... \ No newline at end of file