feature : 외환보유액(합계) API 신규 연동

Author: WoongyuChoi

api/data_processor.py

@@ -9,7 +9,7 @@ class DataProcessor:
     """
 
     @staticmethod
-    def process_exchange_rate_data(response_data):
+    def process_statistic_data(response_data):
         if "StatisticSearch" not in response_data:
             logger.error(f"ValueError: {str(response_data)}")
 
@@ -27,8 +27,8 @@ def process_exchange_rate_data(response_data):
             processed_data.append(
                 {
                     "value": value,
-                    "item_code": row.get("ITEM_CODE1"),
-                    "item_name": row.get("ITEM_NAME1"),
+                    "itemCode": row.get("ITEM_CODE1"),
+                    "itemName": row.get("ITEM_NAME1"),
                     "time": row.get("TIME"),
                     "unit": (
                         row.get("UNIT_NAME").strip() if row.get("UNIT_NAME") else None

api/external.py

@@ -48,4 +48,25 @@ def fetch_exchange_rate(start_date, end_date, item_code=None):
         url = generate_statistic_url(params)
         response_data = fetch_data(url)
 
-        return DataProcessor.process_exchange_rate_data(response_data)
+        return DataProcessor.process_statistic_data(response_data)
+
+    @staticmethod
+    @default_params
+    def fetch_foreign_reserves(start_month, end_month):
+        """
+        외부 API를 호출하여 외환보유액 데이터를 조회합니다.
+        """
+
+        params = APIParams(
+            service_name="StatisticSearch",
+            table_code="732Y001",
+            period="M",
+            start_date=start_month,
+            end_date=end_month,
+            item_code="99",
+        )
+
+        url = generate_statistic_url(params)
+        response_data = fetch_data(url)
+
+        return DataProcessor.process_statistic_data(response_data)

api/server.py

@@ -21,7 +21,6 @@
 
 @app.route("/")
 def health_check():
-    logger.info("Health check called.")
     status = {
         "status": "UP",
         "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
@@ -47,7 +46,19 @@ def get_exchange_rate():
         end_date=params["end_date"],
         item_code=params["item_code"],
     )
-    logger.info("Exchange rate data fetched successfully.")
+    return data, 200
+
+
+@app.route("/api/foreign-reserves", methods=["GET"])
+@cache.cached(query_string=True)
+@json_utf8_response
+def get_foreign_reserves():
+    params = get_request_params("start_month", "end_month")
+
+    data = ExternalAPI.fetch_foreign_reserves(
+        start_month=params["start_month"],
+        end_month=params["end_month"],
+    )
     return data, 200
 
 

decorators/param_defaults.py

@@ -1,12 +1,35 @@
+import inspect
 from functools import wraps
-from utils import get_first_day_of_last_month, get_last_day_of_last_month
+
+from utils import (
+    get_first_day_of_last_month,
+    get_first_month_of_last_year,
+    get_last_day_of_last_month,
+    get_last_month_of_last_year,
+)
 
 
 def default_params(func):
     @wraps(func)
     def wrapper(*args, **kwargs):
-        kwargs["start_date"] = kwargs.get("start_date") or get_first_day_of_last_month()
-        kwargs["end_date"] = kwargs.get("end_date") or get_last_day_of_last_month()
+        func_signature = inspect.signature(func)
+        func_params = func_signature.parameters
+
+        if "start_date" in func_params:
+            kwargs["start_date"] = (
+                kwargs.get("start_date") or get_first_day_of_last_month()
+            )
+        if "end_date" in func_params:
+            kwargs["end_date"] = kwargs.get("end_date") or get_last_day_of_last_month()
+        if "start_month" in func_params:
+            kwargs["start_month"] = (
+                kwargs.get("start_month") or get_first_month_of_last_year()
+            )
+        if "end_month" in func_params:
+            kwargs["end_month"] = (
+                kwargs.get("end_month") or get_last_month_of_last_year()
+            )
+
         return func(*args, **kwargs)
 
     return wrapper

utils/__init__.py

@@ -1,10 +1,17 @@
-from .date_utils import get_first_day_of_last_month, get_last_day_of_last_month
+from .date_utils import (
+    get_first_day_of_last_month,
+    get_last_day_of_last_month,
+    get_first_month_of_last_year,
+    get_last_month_of_last_year,
+)
 from .fetch_utils import fetch_data, generate_statistic_url
 from .request_utils import get_request_params
 
 __all__ = [
     "get_first_day_of_last_month",
     "get_last_day_of_last_month",
+    "get_first_month_of_last_year",
+    "get_last_month_of_last_year",
     "fetch_data",
     "generate_statistic_url",
     "get_request_params",

utils/date_utils.py

@@ -4,8 +4,7 @@
 
 def get_first_day_of_last_month():
     """
-    지난달의 첫날을 반환합니다.
-    :return: 지난달의 첫날 (YYYYMMDD 형식)
+    지난달의 첫날(yyyyMMdd)을 반환합니다.
     """
     today = datetime.today()
     last_day_of_last_month = today.replace(day=1) - timedelta(days=1)
@@ -15,9 +14,26 @@ def get_first_day_of_last_month():
 
 def get_last_day_of_last_month():
     """
-    지난달의 마지막 날을 반환합니다.
-    :return: 지난달의 마지막 날 (YYYYMMDD 형식)
+    지난달의 마지막 날(yyyyMMdd)을 반환합니다.
     """
     today = datetime.today()
     last_day_of_last_month = today.replace(day=1) - timedelta(days=1)
     return last_day_of_last_month.strftime("%Y%m%d")
+
+
+def get_first_month_of_last_year():
+    """
+    지난 해의 첫 달(yyyyMM)을 반환합니다.
+    """
+    today = datetime.today()
+    first_month = today.replace(year=today.year - 1, month=1, day=1)
+    return first_month.strftime("%Y%m")
+
+
+def get_last_month_of_last_year():
+    """
+    지난 해의 마지막 달(yyyyMM)을 반환합니다.
+    """
+    today = datetime.today()
+    last_month = today.replace(year=today.year - 1, month=12, day=1)
+    return last_month.strftime("%Y%m")