AttributeError排查求助：ChangeDetectorService无process属性

运行时触发错误：

AttributeError: 'ChangeDetectorService' object has no attribute 'process'

核心问题 #

1. 方法未归属类：ChangeDetectorService类定义后，所有业务方法（包括process）都写在了类外部的try块中，未缩进至类内部，导致这些方法不属于类成员，实例无法调用。
2. 实例方法缺失self参数：所有需通过实例调用的方法都未声明第一个参数self，即使缩进正确也会触发调用错误。
3. 语法与拼写错误：存在多余代码、缩进错误、变量拼写错误等问题。

修正后的ChangeDetectorService.py代码 #

# ChangeDetectorService.py
from datetime import datetime
from datetime import timedelta
from common.database.mysql_manager import MysqlManager
from common.utils.config import Config, log_start, log_end
import mysql.connector

class ChangeDetectorService:
    def __init__(self, ini_file, app_name):
        self.config_object = Config(ini_file, app_name)
        self.logger = self.config_object.get_logger()
        self.mysql_manager = MysqlManager(self.config_object)

    def get_connection(self):
        connection = mysql.connector.connect(user='user', password='zzzz',
                                             host='host',
                                             database='dbname'
                                             )
        return connection

    def init_change_detector(self):
        print("Change Detector started at " + datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
        # 使用__init__中已实例化的mysql_manager
        conn = self.mysql_manager.get_db_connection()
        cursor = conn.cursor()
        query = ('SELECT table_name, column_name, key_name '
                 'FROM cmooptimizationsvc.csj_chngdtct_tab_col '
                 'WHERE is_active=1 '
                 'ORDER BY table_name, column_name')
        cursor.execute(query)
        records = cursor.fetchall()
        for record in records:
            self.process_col_tab_chg(record[0], record[1], record[2])
        if conn.is_connected():
            conn.close()
            cursor.close()

    def insert_change_log(self, table_name, key_name, attr_name, old_attr_value, new_attr_value):
        insert_query = """INSERT INTO csj_shipment_changelog(table_name, key_name,
                                                             attr_name, old_attr_value,
                                                             new_attr_value)
                           VALUES (%s, %s, %s, %s, %s)"""
        conn = self.get_connection()
        cursor = conn.cursor()
        tuple1 = (table_name, key_name, attr_name, old_attr_value, new_attr_value)
        cursor.execute(insert_query, tuple1)
        conn.commit()
        cursor.close()
        conn.close()

    def find_past_shipment(self, table_name, key_name, column_name, before_date, curr_key):
        saved_col_name = column_name
        saved_key_name = key_name
        conn = self.get_connection()
        cursor = conn.cursor()
        query = f'SELECT {saved_key_name}, {saved_col_name} FROM {table_name} ' \
                f'WHERE rec_cre_dt_utc < "{before_date.strftime("%Y-%m-%d 00:00:00")}" ' \
                f'AND shipment_num = "{curr_key}" ORDER BY rec_cre_dt_utc DESC LIMIT 1'
        cursor.execute(query)
        records = cursor.fetchone()
        cursor.close()
        conn.close()
        return records[1] if records is not None else 0

    def process_col_tab_chg(self, table_name, column_name, key_name):
        saved_key_name = key_name
        saved_col_name = column_name
        old_val = 0
        ini_time_for_now = datetime.now()
        date_before_1day = ini_time_for_now - timedelta(days=1)
        query = f'SELECT {key_name}, {saved_col_name}, rec_cre_dt_utc FROM {table_name} ' \
                f'WHERE rec_cre_dt_utc >= "{date_before_1day.strftime("%Y-%m-%d 00:00:00")}"'
        conn = self.get_connection()
        cursor = conn.cursor()
        cursor.execute(query)
        for (key_name_val, column_name_val, rec_cre_dt_utc) in cursor:
            curr_attr_val = column_name_val
            curr_key_val = key_name_val
            old_val = self.find_past_shipment(table_name,
                                              saved_key_name,
                                              saved_col_name,
                                              rec_cre_dt_utc,
                                              curr_key_val
                                              )
            if curr_attr_val != old_val and old_val != 0:
                self.insert_change_log(table_name, key_name_val, saved_col_name, old_val, curr_attr_val )
        cursor.close()
        conn.close()

    def cleanup(self):
        print("Change Detector stopped " + datetime.now().strftime("%Y-%m-%d %H:%M:%S"))

    def process(self):
        start = datetime.now()
        self.init_change_detector()
        end = datetime.now()
        time_diff = (end - start)
        execution_time = time_diff.total_seconds()
        print("Elapsed time(secs): " + str(execution_time))
        return True  # 返回值匹配job中的success变量

if __name__ == "__main__":
    try:
        # 测试用例，可根据实际情况调整参数
        service = ChangeDetectorService("../config/cmo_backend_local_config.ini", "test_app")
        service.process()
    except Exception as e:
        print(f"Exception: {e}")
    finally:
        # 实例化后调用cleanup
        if 'service' in locals():
            service.cleanup()

关键修正说明 #

1. 将所有业务方法缩进至ChangeDetectorService类内部，确保方法属于类成员。
2. 为所有实例方法添加第一个参数self，保证实例可正确调用方法。
3. 删除get_connection方法中多余的enter code here代码。
4. 修正process_col_tab_chg方法中cursor = conn.cursor()的缩进错误，以及cursor.close缺少括号的问题。
5. 修正init_change_detector中self.mysqlManager的拼写错误，改为与__init__中一致的self.mysql_manager。
6. 将字符串拼接改为f-string，提升代码可读性与安全性。
7. 在process方法中添加返回值True，匹配job代码中success = service.process()的变量接收需求。
8. 调整finally块位置，确保在实例化后调用cleanup方法。

内容的提问来源于stack exchange，提问作者Auro