Python 코드가 전부 한 파일에 존재한다!? (코드 분리하기)

소프트웨어 개발에서 "작동하는 코드"와 "좋은 코드"는 큰 차이가 있습니다.

특히 서버 애플리케이션처럼 장기간 운영되고 지속적인 유지보수가 필요한 코드는 처음부터 확장성과 유지보수성을 고려해야 합니다.

이 글에서는 TCP 소켓 기반의 오디오 처리 서버 코드를 리팩토링하면서 적용한 예시를 살펴보겠습니다.

 

1. 모듈화와 단일 책임 원칙

1.1 파일 분리

하나의 큰 파일을 여러 개의 작은 모듈로 분리하는 것은 코드의 가독성과 유지보수성을 크게 향상시킵니다.

각 모듈은 명확한 책임을 가지고 있어야 합니다.

project/
├── main.py              # 애플리케이션 진입점
├── constants.py         # 상수 정의
├── logger.py           # 로깅 설정
├── connection.py       # TCP 서버 연결 관리
├── client_handler.py   # 클라이언트 연결 처리
├── data_receiver.py    # 데이터 수신 및 처리
├── packet_parser.py    # 바이너리 패킷 파싱
├── audio_processor.py  # 오디오 데이터 처리
└── README.md          # 프로젝트 문서

 

 

1.2 단일 책임 원칙 적용

각 클래스는 하나의 책임만 가져야 합니다. 예를 들어 PacketParser 클래스:

class PacketParser:
    """바이너리 패킷을 파싱하는 책임만 가진 클래스"""
    
    @staticmethod
    def parse_header(raw_bytes):
        """패킷 헤더 파싱"""
        if not raw_bytes or len(raw_bytes) < HEADER_SIZE:
            return None, None
        
        try:
            # 헤더 필드 추출
            language = raw_bytes[:2]
            length = raw_bytes[2:4]
            message = raw_bytes[4:8]
            # ... 나머지 필드 추출
            
            # 기타 변환 작업
            message_id = int.from_bytes(message_id, byteorder='little')
            data_length = int.from_bytes(length, byteorder='little')
            
            return message_id, data_length
        except Exception as e:
            logger.error(f"Error parsing header: {e}")
            return None, None

 

 

2. 설정 관리

2.1 상수 분리

모든 상수를 한 곳에서 관리하면 설정 변경이 용이해집니다:

# constants.py

# 네트워크 설정
DEFAULT_HOST = "0.0.0.0"
DEFAULT_PORT = 1234
DEFAULT_THREAD_COUNT = 1

# 프로토콜 상수
HEADER_SIZE = 16
SAMPLING_RATE = 44000

# 패킷 식별자
MEDIA_PACKET_ID = 0x0000xxxx
END_STREAM_PACKET_ID = 0xxxxxxxxx

# 오디오 처리 파라미터
BUFFER_PROCESS_THRESHOLD = 20
PACKET_RECEIVE_TIMEOUT = 30
WAIT_TIME = 0.1

 

 

2.2 명령행 인자 지원

사용자가 실행 시점에 설정을 변경할 수 있도록 합니다:

def parse_arguments():
    """명령행 인자 파싱"""
    parser = argparse.ArgumentParser(
        description='TCP Binary Server for Audio Processing'
    )
    parser.add_argument(
        '--host', 
        type=str, 
        default=DEFAULT_HOST,
        help=f'Host IP to listen on (default: {DEFAULT_HOST})'
    )
    parser.add_argument(
        '--port', 
        type=int, 
        default=DEFAULT_PORT,
        help=f'Port to listen on (default: {DEFAULT_PORT})'
    )
    parser.add_argument(
        '--threads', 
        type=int, 
        default=DEFAULT_THREAD_COUNT,
        help=f'Number of worker threads (default: {DEFAULT_THREAD_COUNT})'
    )
    return parser.parse_args()

 

 

3. 로깅 체계 구축

3.1 로깅 설정 모듈화

로깅은 디버깅과 모니터링에 필수적입니다:

# logger.py
def setup_logging():
    """로깅 설정"""
    logger = logging.getLogger()
    logger.setLevel(logging.DEBUG)
    
    # 기존 핸들러 제거 (중복 방지)
    if logger.hasHandlers():
        logger.handlers.clear()
    
    # 콘솔 출력 설정
    console_handler = logging.StreamHandler()
    console_formatter = logging.Formatter(
        '%(asctime)s.%(msecs)03d [%(levelname)s] %(message)s',
        datefmt='%Y/%m/%d %I:%M:%S'
    )
    console_handler.setFormatter(console_formatter)
    logger.addHandler(console_handler)
    
    # 파일 출력 설정
    log_filename = f"{datetime.now().strftime('%Y%m%d')}.log"
    file_handler = logging.FileHandler(log_filename)
    file_formatter = logging.Formatter(
        '%(asctime)s.%(msecs)03d [%(levelname)s] %(message)s', 
        datefmt='%Y/%m/%d %I:%M:%S'
    )
    file_handler.setFormatter(file_formatter)
    logger.addHandler(file_handler)
    
    return logger

 

 

3.2 상세한 로그 메시지

각 작업의 진행 상황과 오류를 명확하게 기록합니다:

def process_audio_buffer(audio_buffer):
    """오디오 버퍼 처리"""
    if not audio_buffer or len(audio_buffer) == 0:
        logger.warning("Empty audio buffer received")
        return False
        
    try:
        audio_data = np.concatenate(audio_buffer)
        is_valid = audio_data.size > 0
        
        if is_valid:
            logger.info(f"Successfully processed audio buffer: {audio_data.size} samples")
        else:
            logger.warning("Invalid audio data in buffer")
            
        return is_valid
    except Exception as e:
        logger.error(f"Error processing audio buffer: {e}")
        return False

 

4. 예외 처리 강화

4.1 세분화된 예외 처리

각 작업 단계별로 적절한 예외 처리를 추가합니다:

def receive_all(self, data_length):
    """정확한 길이의 데이터 수신"""
    rx_length = 0
    total_data = b''
    
    while rx_length < data_length:
        try:
            data = self.conn.recv(data_length - rx_length)
            if not data:
                self.logger.debug("Connection closed while receiving data")
                return None
                
            rx_length += len(data)
            total_data += data
        except socket.timeout:
            self.logger.warning("Socket timeout while receiving data")
            return None
        except ConnectionResetError:
            self.logger.error("Connection reset by peer")
            return None
        except Exception as e:
            self.logger.error(f"Unexpected error receiving data: {e}")
            return None
            
    return total_data

 

 

4.2 자원 정리 보장

리소스 해제를 보장하기 위해 context manager와 finally 블록을 활용합니다:

def _accept_connections(self):
    """클라이언트 연결 수락 및 처리"""
    with concurrent.futures.ThreadPoolExecutor(max_workers=self.thread_count) as executor:
        try:
            while True:
                try:
                    conn, addr = self.server_socket.accept()
                    self.logger.info(f"New connection from {addr}")
                    executor.submit(self._handle_client, conn, addr)
                except Exception as e:
                    self.logger.error(f"Error accepting client: {e}")
                    continue
        except KeyboardInterrupt:
            self.logger.info("Server shutdown requested")
        finally:
            self.logger.info("Shutting down connection acceptor")

 

 

5. 코드 문서화

5.1 독스트링(Docstring) 활용

모든 모듈, 클래스, 메서드에 명확한 문서화를 추가합니다:

class AudioProcessor:
    """오디오 데이터 처리를 담당하는 클래스
    
    이 클래스는 바이너리 형태의 오디오 데이터를 받아서
    처리 가능한 형태로 변환하고 필요한 전처리를 수행합니다.
    
    Attributes:
        logger: 로깅을 위한 Logger 인스턴스
    """
    
    @staticmethod
    def process_audio_packet(media_body):
        """단일 오디오 패킷을 처리합니다.
        
        Args:
            media_body (bytes): 처리할 오디오 데이터
            
        Returns:
            numpy.ndarray: 처리된 오디오 데이터
            None: 처리 실패 시
        
        Raises:
            Exception: 오디오 처리 중 발생한 예외
        """

 

 

5.2 README 작성

프로젝트의 목적, 구조, 설치 방법, 사용법 등을 명확하게 문서화합니다:

# TCP Binary Audio Processing Server

TCP 소켓 통신을 통해 오디오 데이터를 수신하고 처리하는 서버입니다.

## 주요 기능

- TCP 소켓을 통한 바이너리 데이터 수신
- 멀티스레드 방식으로 여러 클라이언트 동시 처리
- 오디오 패킷 수신 및 처리
- 처리 결과를 클라이언트에게 응답

## 설치 및 실행

### 필수 라이브러리
```bash
pip install soundfile librosa numpy
```

### 실행 방법
```bash
python main.py --host 127.0.0.1 --port 8000 --threads 4
```
```

 

 

 

6. 코드 구조화

6.1 클래스 계층 구조

관련된 기능들을 논리적으로 그룹화합니다:

class Connection:
    """서버 연결 관리"""
    def __init__(self, host, port, thread_count):
        self.host = host
        self.port = port
        self.thread_count = thread_count
        self.server_socket = None
        self.logger = logging.getLogger()

class ClientHandler:
    """클라이언트 연결 처리"""
    def __init__(self, conn, thread_name):
        self.conn = conn
        self.thread_name = thread_name
        self.data_receiver = DataReceiver(conn, thread_name)

class DataReceiver:
    """데이터 수신 및 처리"""
    def __init__(self, conn, thread_name):
        self.conn = conn
        self.current_buffer = []

 

 

6.2 메서드 분리

복잡한 로직을 작은 단위의 메서드로 분리합니다:

class DataReceiver:
    def receive_media_body(self):
        """미디어 데이터 수신 메인 로직"""
        self._initialize_reception()
        
        while not self._is_reception_complete():
            data = self._receive_next_packet()
            if not data:
                continue
                
            if self._is_media_packet(data):
                self._process_media_packet(data)
            elif self._is_end_packet(data):
                break
                
        return self._finalize_reception()
    
    def _initialize_reception(self):
        """수신 초기화"""
        self.current_buffer = []
        self.receive_packet = 0
    
    def _is_reception_complete(self):
        """수신 완료 여부 확인"""
        return len(self.current_buffer) >= BUFFER_PROCESS_THRESHOLD

 

 

코드 리팩토링은 단순히 "작동하는" 코드를 "유지보수 가능한" 코드로 변환하는 과정입니다.

1. 모듈화를 통한 코드 구조 개선
2. 설정의 중앙화 및 유연성 확보
3. 강력한 로깅 시스템 구축
4. 철저한 예외 처리
5. 명확한 문서화
6. 논리적인 코드 구조화

이러한 개선은 코드의 가독성을 높이고, 버그 수정을 용이하게 하며, 새로운 기능 추가를 더 쉽게 만듭니다.

코드의 품질을 지속적으로 개선하는 것은 장기적으로 프로젝트의 핵심 요소입니다.