如何捕获从 MYSQL 触发器（嵌入在另一个 py 脚本中）内部调用的 python 脚本的输出？

Posted 2023-04-13

【中文标题】如何捕获从 MYSQL 触发器（嵌入在另一个 py 脚本中）内部调用的 python 脚本的输出？

【英文标题】：How to capture output from a python script called from inside a MYSQL trigger (which is embedded in another py script)?

【发布时间】：2021-09-18 12:29:02

【问题描述】：

我需要做以下事情：

使用第三方 api 抓取推文并将推文内容和其他详细信息存储在 mysql 数据库中。

在 db 表中添加新行之前，我需要获取推文来源位置的 temp 并在表中添加 temp 值。

为了完成第二个任务，我编写了 2 个 python 脚本。第一个 python 脚本将 mysql 触发器添加到数据库中。在 mysql 触发器内部，我在阅读 http://crazytechthoughts.blogspot.sg/2011/12/call-external-program-from-mysql.html 后使用 sys.eval() 调用第二个使用 api 获取温度数据的 python 脚本。

两个脚本单独工作都非常好，但触发器无法从第二个脚本中获取温度数据。我该如何解决？

Script to add trigger

        def create_Triggers(self):
        
        try:
     
            mysql_trig = """
            
            CREATE TRIGGER mysql_Trigger 
            BEFORE INSERT ON tweets FOR EACH ROW BEGIN 
            DECLARE loc CHAR(255); DECLARE result Double; DECLARE city CHAR(255); 
            DECLARE station CHAR(255); DECLARE dtime CHAR(255); SET city = NEW.city; 
            SET station =  "OYSN:9:YE"; SET dtime = NEW.dtime; 
            SET loc = CONCAT('python D:test case scripts/Weather_enrichment_triggers/weather_enrichment.py',city,station,dtime); 
            SET result = sys_eval(loc); 
            SET NEW.Temperature = result; 
            END;
            """
           
            self.curr.execute(mysql_trig)
            print("trigger executed")
            self.connection.commit()
            
            #self.curr.execute(postgre_trig)
            #self.connection.commit()
        
        except Exception as e:
            print(e)

    def test_triggers(self, query):

            self.curr.execute(query)
            self.connection.commit()

Script to fetch temp data(called from inside the trigger)

city  = sys.argv[1]
station = sys.argv[2]
dtime = sys.argv[3]

def weather_info(city,station,dtime):
        
#Get weather information for a given city and date
   template_request = "https://api.weather.com/v1/location/station/observations/historical.json?apiKey=apikey&units=m&startDate=start_date&endDate=end_date"
   df_header = ["City", "Year", "Month", "Day", "Hour", "Temperature(C)", "Condition"]

   def get_weather_data(city, year, month, day, station):
       start_date = "%d%02d%02d" % (year, month, day)
       end_date = "%d%02d%02d" % (year, month, day)
       request = template_request.format(station=station, start_date=start_date, end_date=end_date)
       request_data = json.loads(requests.get(request).content)
       weather_data = []
       last_dt = None
       for observation in request_data["observations"]:
           dt = datetime.fromtimestamp(observation["valid_time_gmt"]+3600)
           if last_dt and dt.hour > (last_dt.hour + 1):
               last_row = deepcopy(weather_data[-1])
               last_row[4] = last_row[4]+1
               weather_data.append(last_row)
           weather_data.append([city, year, month, dt.day, dt.hour, observation["temp"], observation["wx_phrase"]])
           last_dt = dt
       return weather_data
    
   dtime = datetime.strptime(dtime, '%Y-%m-%d %H:%M:%S%z')
   data = get_weather_data(city, dtime.year, dtime.month, dtime.day, station)

   weather_df = pd.DataFrame(data, columns=df_header).drop_duplicates(subset=["City", "Year", "Month", "Day", "Hour"])
   avg = (weather_df["Temperature(C)"].values).mean()
   weather_df = pd.DataFrame()
   return float(avg)



temp = weather_info(city, station, dtime)
temp = str(temp)
sys.stdout.write(temp)

【问题讨论】：

不要为此使用mysql触发器，在没有任何触发器的情况下使用python。

是的，我同意你的观点，但我手头的任务是将 mysql 的性能与另一个数据库进行比较，因此我不得不使用 mysql。

您可以使用 mysql 作为数据存储，但不要使用触发器。

【参考方案1】：

sys_eval 函数只返回它调用的脚本的exit code。

它似乎没有提供一种干净的方式将数据从外部脚本传回 MySql 服务器。

另外，它非常不安全。只是说说而已。

编辑

一种普遍接受的方法是轮询。

拥有您的第一个脚本，将数据插入到您的表中，将 tweets.Temperaure 设置为 NULL。

摆脱触发器和 sys_eval 扩展。

将此索引放在该列上。

ALTER TABLE tweets ADD INDEX (Temperature, city)

让你的第二个 python 脚本循环运行，每隔几秒就运行一次

SELECT id, city 
  FROM tweets
 WHERE temperature IS NULL

然后，对于检索到的每一行，让您的脚本获取城市的温度并执行

UPDATE tweets 
   SET Temperature = ###your_value###
 WHERE id = ###the_id_you_SELECTed###

索引使那些频繁的 SELECT 语句变得非常快，尤其是当它们不返回任何行时。权衡：您最新的行还没有温度。

另一种选择是最简单的：让您的第一个 python 程序执行插入以查找每条推文的温度并将其与行的其余数据一起直接插入。

不管怎样，很少有应用程序使用这些 MySQL 扩展，如 sys_eval，如果您将应用程序放在生产 MySQL 服务器上，它不太可能有扩展。

【讨论】：

是的，我同意你的看法，但我手头的任务要求我使用 python 使用 mysql db，如果没有，我是否可以通过其他方式将数据从外部脚本传递到 mysql 服务器sys_eval(

请看我的编辑。