system-design-primer/solutions/object_oriented_design/call_center/call_center.py

123 lines
3.2 KiB
Python
Raw Normal View History

2017-03-04 09:50:44 +07:00
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 NotImplementedError('Directors must be able to handle any call')
2017-03-04 09:50:44 +07:00
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):
pass
def notify_call_completed(self, call):
pass
def dispatch_queued_call_to_newly_freed_employee(self, call, employee):
pass