43 lines
1.5 KiB
Python
43 lines
1.5 KiB
Python
import logging
|
|
from typing import Dict, Callable
|
|
|
|
from core import Request, Response
|
|
from core.exceptions import BankNodeError
|
|
|
|
from bank_protocol.commands import (account_balance,
|
|
account_create,
|
|
account_deposit,
|
|
account_remove,
|
|
account_withdrawal,
|
|
bank_code,
|
|
bank_number_of_clients,
|
|
bank_total_amount)
|
|
|
|
|
|
class CommandHandler:
|
|
def __init__(self, config: Dict):
|
|
self.logger = logging.getLogger(__name__)
|
|
self.config = config
|
|
|
|
self.registered_commands: Dict[str, Callable] = {
|
|
"BC": bank_code,
|
|
"AC": account_create,
|
|
"AD": account_deposit,
|
|
"AW": account_withdrawal,
|
|
"AB": account_balance,
|
|
"AR": account_remove,
|
|
"BA": bank_total_amount,
|
|
"BN": bank_number_of_clients
|
|
}
|
|
|
|
def execute(self, request: Request) -> Response:
|
|
if request.command_code not in self.registered_commands:
|
|
self.logger.warning("Unknown command %s", request.command_code)
|
|
raise BankNodeError(f"Unknown command {request.command_code}")
|
|
|
|
command = self.registered_commands[request.command_code]
|
|
|
|
response = command(request, self.config)
|
|
|
|
return f"{request.command_code} {response}"
|