OOP practice by ATM (feat. 파이썬, Airflow Dag)

idle-danie·2023년 7월 27일
OOP파이썬
2
post-thumbnail

Airflow Dag를 작성하다 보면 객체 지향 프로그래밍에 대한 중요성을 느낀다.
오랜만에 문법도 익힐 겸 ATM도 구현해보고 이전에 공부했던 전공 책을 꺼내어 OOP 개념과 python class 문법을 복습해보았다.

python class 구현 방법

파이썬에서 클래스를 구현하는 기본 방법을 간단한 계산기 구현 예제로 살펴보겠습니다.

 class Calculator:
     def __init__(self):
         self.result = 0
    
     def add(self, num):
         self.result += num
         return self.result

 cal1 = Calculator()
 cal2 = Calculator()

 class FourCal:
     def __init__(self, first, second):
         self.first = first
         self.second = second
     def setdata(self, first, second):
         self.first = first
         self.second = second
     def add(self):
         result = self.first + self.second
         return result
    
 class MoreFourCal(FourCal):
     pass

 a = MoreFourCal(4,2)
 print(a.add())

이 예제에서는 기본적인 클래스 구현과 상속에 대해 보여줍니다. Calculator 클래스는 기본적인 덧셈 기능을 제공하며, FourCal 클래스는 두 숫자를 더하는 메서드를 포함합니다. MoreFourCal 클래스는 FourCal 클래스를 상속받아 추가 기능 없이 그대로 사용합니다. 이제 이를 기반으로 ATM기를 구현하여 객체 지향 프로그래밍(OOP)의 원칙을 다시 익혀보겠습니다.

ATM기를 구현하며 OOP를 다시 익히다

OOP를 처음 접하는 분들에게는 어떠한 것도 참고하지 마시고 최소 기능만 구상하시고 ATM 구현해보시는 것을 추천합니다.

ATM, ATM controller Class 구현

# Import Bank class from the bank module

from bank import Bank

# Define the class ATM which represents the ATM machine

class ATM:

    # Constructor for ATM class. It takes a Bank object as argument
    
    def __init__(self, bank: Bank):
        self.bank = bank
        self.card_number = None
        self.account_number = None

    # Method for inserting card. It validates the card_number with the bank
    
    def insert_card(self, card_number: str) -> None:
        if self.bank.validate_card(card_number):
            self.card_number = card_number
        else:
            raise ValueError("Invalid card")
        
    # Method for entering PIN. It validates the PIN with the bank
    
    def enter_pin(self, pin: str) -> None:
        if not self.bank.validate_pin(self.card_number, pin):
            raise ValueError("Invalid PIN")
        
    # Method for getting a list of accounts linked to the card
    
    def get_accounts_list(self) -> list:
        return self.bank.get_accounts(self.card_number)
    
    # Method for selecting an account. It validates if the account_name exists in the bank
    
    def select_account(self, account_name: str) -> None:
        if account_name not in self.bank.get_accounts(self.card_number):
            raise ValueError("Account does not exist")
        self.account_number = account_name

    # Method for checking the balance of the selected account
    
    def check_balance(self) -> int:
        return self.bank.get_balance(self.card_number, self.account_number)
        
    # Method for depositing money into the selected account
    
    def deposit_money(self, amount: int) -> None:
        self.bank.deposit(self.card_number, self.account_number, amount)

    # Method for withdrawing money from the selected account
    
    def withdraw_money(self, amount: int) -> None:
        self.bank.withdraw(self.card_number, self.account_number, amount)

# Define the class ATMController which controls the ATM machine

class ATMController:
    def __init__(self, atm: ATM):
        self.atm = atm

    def insert_card(self, card_number: str) -> str:
        try:
            self.atm.insert_card(card_number)
            return "Your card has been successfully inserted"
        except ValueError as e:
            return str(e)

    def enter_pin(self, pin: str) -> str:
        try:
            self.atm.enter_pin(pin)
            return "PIN number entered successfully"
        except ValueError as e:
            return str(e)

    def get_accounts_list(self) -> list:
        try:
            return self.atm.get_accounts_list()
        except ValueError as e:
            return str(e)

    def select_account(self, account_name: str) -> str:
        try:
            self.atm.select_account(account_name)
            return "Account has been successfully selected"
        except ValueError as e:
            return str(e)

    def check_balance(self) -> int:
        try:
            return self.atm.check_balance()
        except ValueError as e:
            return str(e)

    def deposit_money(self, amount: int) -> str:
        try:
            self.atm.deposit_money(amount)
            return "Deposit has been processed"
        except ValueError as e:
            return str(e)

    def withdraw_money(self, amount: int) -> str:
        try:
            self.atm.withdraw_money(amount)
            return "Withdrawal has been processed"
        except ValueError as e:
            return str(e)

ATM Class workflow

  • ATM 클래스는 Bank 클래스의 객체를 받아 초기화됩니다.
  • insert_card, enter_pin, get_accounts_list, select_account, check_balance, deposit_money, withdraw_money 메서드를 통해 ATM의 기본 기능을 수행합니다.
  • 각 메서드는 Bank 클래스의 메서드를 호출하여 필요한 작업을 수행합니다.

ATM Controller Class workflow

  • ATMController 클래스는 ATM 객체를 받아 초기화됩니다.
  • 각 메서드는 ATM 클래스의 메서드를 호출하여 결과를 반환하거나 예외를 처리합니다.
  • insert_card, enter_pin, get_accounts_list, select_account, check_balance, deposit_money, withdraw_money 메서드를 통해 ATM의 기능을 제어합니다.

Bank Class 구현

from abc import ABC, abstractmethod

# This class represents the blueprint for any bank

class Bank(ABC):

    # Method to check if a card is valid
    
    @abstractmethod
    def validate_card(self, card_number: str) -> bool:
        pass

    # Method to verify the entered PIN
    
    @abstractmethod
    def validate_pin(self, card_number: str, pin: str) -> bool:
        pass

    # Method to retrieve all accounts associated with a card
    
    @abstractmethod
    def get_accounts(self, card_number: str) -> list:
        pass

    # Method to check balance of an account
    
    @abstractmethod
    def get_balance(self, card_number: str, account_number: str) -> int:
        pass

    # Method to deposit money into an account
    
    @abstractmethod
    def deposit(self, card_number, account_number: str, amount: int) -> None:
        pass
        
    # Method to withdraw money from an account
    
    @abstractmethod
    def withdraw(self, card_number, account_number: str, amount: int) -> None:
        pass

# MockBank is a basic bank for testing purposes

class MockBank(Bank):

    # MockBank has some predefined accounts for testing
    
    def __init__(self):
        self.accounts = {
            "123456": {"pin": 1234, "accounts": {"account1": 5000, "account2": 10000}},
            "654321": {"pin": 4321, "accounts": {"account1": 3000, "account2": 20000}}
        }
    
    def validate_card(self, card_number: str) -> bool:
        return card_number in self.accounts
 
    def validate_pin(self, card_number: str, pin: int) -> bool:
        return self.accounts[card_number]["pin"] == pin
    
    def get_accounts(self, card_number: str) -> list:
        return list(self.accounts[card_number]["accounts"].keys())

    def get_balance(self, card_number: str, account_name: str) -> int:
        return self.accounts[card_number]["accounts"][account_name]
    
    def deposit(self, card_number: str, account_name: str, amount: int) -> None:
        self.accounts[card_number]["accounts"][account_name] += amount

    def withdraw(self, card_number: str, account_name: str, amount: int) -> None:
        if self.accounts[card_number]["accounts"][account_name] < amount:
            raise ValueError("balance is insufficient")
        self.accounts[card_number]["accounts"][account_name] -= amount

Bank Class workflow

  • Bank 클래스는 추상 클래스(ABC)로 정의되어 있으며, 여러 추상 메서드를 포함합니다.
  • MockBank 클래스는 Bank 클래스를 상속받아 실제 구현을 제공합니다.
  • validate_card, validate_pin, get_accounts, get_balance, deposit, withdraw 메서드를 통해 카드 유효성 검사, PIN 검사, 계좌 조회, 잔액 조회, 입금 및 출금 기능을 제공합니다.

컨트롤러 테스트 by unittest

from bank import MockBank
from atm import ATM, ATMController
import unittest

class ATMControllerTest(unittest.TestCase):
    def setUp(self):
        self.bank = MockBank()
        self.atm = ATM(self.bank)
        self.atm_controller = ATMController(self.atm)

    def test_insert_card(self):
        result = self.atm_controller.insert_card("123456")
        self.assertEqual(result, "Your card has been successfully inserted")

    def test_enter_pin(self):
        self.atm_controller.insert_card("123456")
        result = self.atm_controller.enter_pin(1234)
        self.assertEqual(result, "PIN number entered successfully")

    def test_get_accounts_list(self):
        self.atm_controller.insert_card("123456")
        result = self.atm_controller.get_accounts_list()
        self.assertEqual(result, ["account1", "account2"])

    def test_select_account(self):
        self.atm_controller.insert_card("123456")
        self.atm_controller.get_accounts_list()
        result = self.atm_controller.select_account("account1")
        self.assertEqual(result, "Account has been successfully selected")

    def test_check_balance(self):
        self.atm_controller.insert_card("123456")
        self.atm_controller.get_accounts_list()
        self.atm_controller.select_account("account1")
        result = self.atm_controller.check_balance()
        self.assertEqual(result, 5000)

    def test_deposit_money(self):
        self.atm_controller.insert_card("123456")
        self.atm_controller.get_accounts_list()
        self.atm_controller.select_account("account1")
        result = self.atm_controller.deposit_money(1000)
        self.assertEqual(result, "Deposit has been processed")

    def test_withdraw_money(self):
        self.atm_controller.insert_card("123456")
        self.atm_controller.get_accounts_list()
        self.atm_controller.select_account("account1")
        result = self.atm_controller.withdraw_money(3000)
        self.assertEqual(result, "Withdrawal has been processed")

if __name__ == '__main__':
    unittest.main()

# if hard to understand unittest, use this
# def test_ATMController():
    
#     bank = MockBank()
#     atm = ATM(bank)
#     atm_controller = ATMController(atm)

#     result = atm_controller.insert_card("654321")
#     assert result == "Your card has been successfully inserted"

#     result = atm_controller.enter_pin(4321)
#     assert result == "PIN number entered successfully"
    
#     result = atm_controller.get_accounts_list()
#     assert result == ["account1", "account2"]
   
#     result = atm_controller.select_account("account1")
#     assert result == "Account has been successfully selected"
  
#     result = atm_controller.check_balance()
#     assert result == 3000
   
#     result = atm_controller.deposit_money(1000)
#     assert result == "Deposit has been processed"
  
#     result = atm_controller.withdraw_money(3000)
#     assert result == "Withdrawal has been processed"

# test_ATMController()

Test Class workflow

  • ATMControllerTest 클래스는 unittest.TestCase를 상속받아 ATM의 다양한 기능을 테스트합니다.
  • setUp 메서드를 통해 테스트 환경을 초기화합니다.
  • 각 테스트 메서드는 ATMController 클래스의 메서드를 호출하여 기능을 검증합니다.

해당 예제를 통해 기본적인 클래스 구현, 상속, 추상 클래스 사용 및 유닛 테스트 작성 방법을 알아보았습니다.

Airflow DAG 작성에서 객체 지향 프로그래밍(OOP)의 중요성

개인적인 생각으로, Airflow DAG(DAG: Directed Acyclic Graph)을 작성할 때 객체 지향 프로그래밍(OOP)이 중요한 이유는 특정 작업(task)을 클래스로 정의하고, 이를 여러 DAG에서 재사용하는 것이 Airflow 아키텍처의 지향점에 부합하다고 생각해서이다.

코드 예시: OOP를 활용한 Airflow DAG 작성

Step 1: 공통 작업 클래스를 정의

from airflow.models import BaseOperator
from airflow.utils.decorators import apply_defaults

class MyTaskOperator(BaseOperator):
    @apply_defaults
    def __init__(self, param1, param2, *args, **kwargs):
        super(MyTaskOperator, self).__init__(*args, **kwargs)
        self.param1 = param1
        self.param2 = param2

    def execute(self, context):
        # 작업 실행 로직
        print(f"Executing task with {self.param1} and {self.param2}")

Step 2: DAG 작성 시 작업 클래스를 활용

from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from datetime import datetime

# DAG 기본 설정
default_args = {
    'owner': 'airflow',
    'start_date': datetime(2023, 1, 1),
    'retries': 1,
}

# DAG 정의
with DAG(dag_id='my_dag', default_args=default_args, schedule_interval='@daily') as dag:
    start = DummyOperator(task_id='start')

    # MyTaskOperator를 사용하여 작업 생성
    task1 = MyTaskOperator(task_id='task1', param1='value1', param2='value2')
    task2 = MyTaskOperator(task_id='task2', param1='value3', param2='value4')

    end = DummyOperator(task_id='end')

    # 작업 의존성 설정
    start >> task1 >> task2 >> end

코드 설명

  1. MyTaskOperator 클래스 정의:

    • MyTaskOperator는 BaseOperator를 상속받아 구현한 클래스입니다.
    • __init__ 메서드에서 필요한 매개변수를 초기화합니다.
    • execute 메서드에서 실제 작업을 수행하는 로직을 작성합니다.

  2. DAG 정의:

    • DAG를 정의할 때 MyTaskOperator를 사용하여 작업을 생성합니다.
    • start, task1, task2, end 작업을 정의하고, 작업 간의 의존성을 설정합니다.

결론

Airflow DAG를 작성할 때 객체 지향 프로그래밍(OOP)을 활용하면 코드의 재사용성, 유지보수성, 확장성, 가독성을 크게 향상시킬 수 있기에, 예시처럼 작업을 클래스로 정의하고 이를 여러 DAG에서 재사용하면, 복잡한 워크플로우를 보다 효율적으로 관리할 수 있습니다 :)

ATM 코드에 대한 실행방법과 추가로 현금함을 구현하기 위한 가이드라인은 아래 github 링크의 README.md에 포함되어 있습니다 :)
소스코드: https://github.com/idle-danie/OOP_atm

profile
idle-danie
wanna be idéal DE
이전 포스트

ML/DL 용어정리

다음 포스트

OpenAI를 활용한 챗봇 설계 및 실험 (feat. fine-tuning & parameter 정밀 분석)

0개의 댓글

관련 채용 정보