Python connection.py를 SOLID 원칙으로 리팩토링하기

실무에서 사용하던 DB/서비스 연결 모듈을 SOLID 원칙에 맞게 단계적으로 개선한 과정을 정리합니다.

---

개요

Airflow 기반 데이터 파이프라인에서 사용하던 connection.py는 하나의 Connection 클래스가 Hive, PostgreSQL, Oracle, Redis, SFTP 등 모든 서비스 연결을 담당하고 있었습니다.
기능은 동작했지만, 확장하거나 유지보수할수록 코드가 점점 복잡해지는 문제가 있었습니다.

이를 SOLID 원칙 기준으로 5단계에 걸쳐 개선했습니다.

---

1. _resolve_config_path 분리 (SRP — 단일 책임 원칙)

기존 코드

def _load_config(self, filename):
    config = configparser.ConfigParser()
    config_path = os.path.join(CONF_DIR, filename)
    if not os.path.exists(config_path):
        local_path = os.path.join(
            os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'cfg', filename
        )
        if os.path.exists(local_path):
            config_path = local_path
        else:
            raise FileNotFoundError(f"Configuration file not found: {config_path}")
    config.read(config_path)
    return config

개선된 코드

def _resolve_config_path(self, filename: str) -> str:
    """
    설정 파일의 실제 경로를 탐색하여 반환합니다.
    1순위: 운영 경로 (CONF_DIR)
    2순위: 로컬 경로 (테스트 환경 폴백)
    """
    primary = os.path.join(CONF_DIR, filename)
    if os.path.exists(primary):
        return primary

    fallback = os.path.join(
        os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'cfg', filename
    )
    if os.path.exists(fallback):
        return fallback

    raise FileNotFoundError(f"Configuration file not found: {primary}")

def _load_config(self, filename: str):
    """지정된 파일명으로 설정 파일을 파싱하여 반환합니다."""
    config = configparser.ConfigParser()
    config.read(self._resolve_config_path(filename))
    return config

왜 이렇게 개선해야 하는가?

_load_config는 원래 하나의 메서드에서 "경로 탐색" 과 "파일 파싱" 두 가지를 동시에 처리하고 있었습니다.
SRP는 클래스/함수가 하나의 이유만으로 변경되어야 한다고 말합니다.

• 운영 경로 전략이 바뀌면 → _resolve_config_path만 수정
• 파싱 방식이 바뀌면 → _load_config만 수정

두 책임을 분리함으로써 각 메서드의 변경 이유가 명확해집니다.

---

2. ConfigLoader.get_config() 통합 (OCP — 개방/폐쇄 원칙)

기존 코드

def get_sftp_config(self):
    config = self._load_config('sftp.conf')
    return config[self.env]

def get_postgresql_config(self):
    config = self._load_config('postgresql.conf')
    return config[self.env]

def get_oracle_config(self):
    config = self._load_config('oracle.conf')
    return config[self.env]

def get_redis_cluster_config(self):
    config = self._load_config('redis_cluster.conf')
    return config[self.env]

def get_hive_config(self):
    config = self._load_config('hive.conf')
    return config[self.env]

개선된 코드

def get_config(self, service_name: str):
    """
    서비스명에 해당하는 환경 설정 섹션을 반환합니다.
    새 서비스 추가 시 {service_name}.conf 파일만 추가하면 됩니다.
    """
    return self._load_config(f'{service_name}.conf')[self.env]

사용 측 코드도 일관된 방식으로 통일됩니다.

# 변경 전
cfg = self.loader.get_sftp_config()
cfg = self.loader.get_postgresql_config()

# 변경 후
cfg = self.loader.get_config('sftp')
cfg = self.loader.get_config('postgresql')

왜 이렇게 개선해야 하는가?

OCP는 "확장에는 열려 있고, 수정에는 닫혀 있어야 한다" 는 원칙입니다.

기존 코드는 새 서비스(예: HDFS, ETT)를 추가할 때마다 ConfigLoader에 get_xxx_config() 메서드를 직접 추가(수정)해야 했습니다.
개선 후에는 {service_name}.conf 파일만 추가하면 get_config('hdfs')로 바로 사용할 수 있습니다. ConfigLoader를 수정할 필요가 없습니다.

---

3. BaseConnection 추상 인터페이스 도입 (LSP — 리스코프 치환 원칙)

기존 코드

class Connection:
    def __init__(self):
        self.loader = ConfigLoader()  # 구현체 직접 생성

    def close(self, connections):
        for name in connections:       # 문자열로 attribute 이름 전달
            conn = getattr(self, name, None)
            try:
                if conn is not None:
                    conn.close()
            except Exception:
                pass

개선된 코드

from abc import ABC, abstractmethod

class BaseConnection(ABC):
    """모든 연결 클래스의 추상 기반 클래스."""

    @abstractmethod
    def connect(self):
        """연결 객체를 생성하여 반환합니다."""
        ...

# 연결 해제는 모듈 수준 독립 함수로 분리
def close_connections(*conns) -> None:
    """서비스와 무관한 범용 연결 해제 함수."""
    for conn in conns:
        if conn is None:
            continue
        try:
            conn.close()
        except Exception:
            pass

사용 측 코드 변화:

# 변경 전 — 문자열로 attribute 이름 전달 (동작 보장 불가)
connection.Connection().close(["conn_oracle", "rc11"])

# 변경 후 — 연결 객체를 직접 전달 (타입 안전)
connection.close_connections(self.conn_oracle, self.rc11)

왜 이렇게 개선해야 하는가?

LSP는 "서브타입은 언제나 기반 타입으로 교체할 수 있어야 한다" 는 원칙입니다.

BaseConnection을 도입하면:

• 모든 연결 클래스가 connect() 구현을 강제받아 인터페이스 일관성을 보장합니다.
• OracleConnection, HiveConnection 등이 BaseConnection 타입으로 다형적으로 사용될 수 있습니다.

또한 기존 close(connections)는 문자열 attribute 이름 리스트를 받는 위험한 방식이었습니다.
close_connections(*conns)는 연결 객체를 직접 받으므로 런타임 오류 없이 안전하게 동작합니다.

---

4. Constructor Injection 적용 (DIP — 의존성 역전 원칙)

기존 코드

class Connection:
    def __init__(self):
        self.loader = ConfigLoader()  # 구현체에 강하게 결합

개선된 코드

from typing import Optional

class _BaseServiceConnection(BaseConnection):
    def __init__(self, loader: Optional[ConfigLoader] = None):
        self.loader = loader or ConfigLoader()  # 외부 주입 or 기본값

왜 이렇게 개선해야 하는가?

DIP는 "고수준 모듈이 저수준 모듈에 의존해서는 안 된다" 는 원칙입니다.

기존 코드는 Connection 내부에서 ConfigLoader()를 직접 생성해 구현체에 강하게 결합되어 있었습니다.

Constructor Injection을 적용하면:

• 테스트 시 Mock ConfigLoader를 주입해 실제 파일 시스템 없이 단위 테스트 가능
• 확장 시 다른 설정 소스(환경변수, AWS Parameter Store 등)로 교체 가능

# 테스트 코드 예시
mock_loader = MockConfigLoader()
conn = OracleConnection(loader=mock_loader)  # 실제 설정 파일 불필요

---

5. 서비스별 클래스 분리 (SRP — 단일 책임 원칙)

기존 코드

class Connection:
    """5가지 서비스 연결을 모두 담당하는 단일 클래스"""

    def sftp_connection(self): ...
    def redis_cluster_connection(self, cluster_section): ...
    def postgresql_connection(self): ...
    def oracle_connection(self): ...
    def hive_connection(self): ...

개선된 코드

class _BaseServiceConnection(BaseConnection):
    """공통 기반: loader, _get_aes 공유"""
    ...

class SftpConnection(_BaseServiceConnection):
    def connect(self): ...   # SFTP만 책임

class RedisClusterConnection(_BaseServiceConnection):
    def connect(self, cluster_section: str = 'rc00'): ...   # Redis만 책임

class PostgresqlConnection(_BaseServiceConnection):
    def connect(self): ...   # PostgreSQL만 책임

class OracleConnection(_BaseServiceConnection):
    def connect(self): ...   # Oracle만 책임

class HiveConnection(_BaseServiceConnection):
    def connect(self): ...   # Hive만 책임

사용 측 코드:

# 직접 서비스 클래스를 사용
self.conn_oracle = connection.OracleConnection().connect()
self.rc11        = connection.RedisClusterConnection().connect('rc11')

# 해제
connection.close_connections(self.conn_oracle, self.rc11)

왜 이렇게 개선해야 하는가?

기존 Connection 클래스는 5가지 서비스의 연결 로직을 모두 보유하고 있었습니다.
Oracle 연결 방식이 바뀌면 Connection 전체를 수정해야 하고, 이는 SFTP나 Hive 연결에도 영향을 줄 위험이 있습니다.

분리 후에는:

• 변경 영향 범위 최소화: OracleConnection만 수정하면 됩니다.
• 가독성 향상: 클래스 이름만 봐도 어떤 서비스인지 명확합니다.
• 테스트 용이성: 서비스별로 독립적인 단위 테스트 작성 가능합니다.

---

최종 클래스 구조

BaseConnection (ABC)                  ← 순수 인터페이스 계약
└── _BaseServiceConnection            ← 프로젝트 공통 기반 (loader, AES)
    ├── SftpConnection
    ├── RedisClusterConnection
    ├── PostgresqlConnection
    ├── OracleConnection
    └── HiveConnection

close_connections(*conns)             ← 모듈 수준 범용 연결 해제 함수
ConfigLoader                          ← 환경별 설정 파일 로드

---

개선 전/후 비교 요약

개선 항목	SOLID 원칙	핵심 효과
_resolve_config_path 분리	SRP	경로 탐색과 파일 파싱 책임 분리
get_config() 메서드 통합	OCP	새 서비스 추가 시 코드 수정 불필요
BaseConnection ABC 도입	LSP	connect() 구현 강제, 다형성 보장
Constructor Injection	DIP	구현체 교체/테스트 가능 구조
서비스별 클래스 분리	SRP	변경 영향 범위 최소화

---