vincbeck commented on code in PR #46391:
URL: https://github.com/apache/airflow/pull/46391#discussion_r1952931812
##########
providers/standard/src/airflow/providers/standard/triggers/file.py:
##########
@@ -75,3 +84,47 @@ async def run(self) -> typing.AsyncIterator[TriggerEvent]:
yield TriggerEvent(True)
return
await asyncio.sleep(self.poke_interval)
+
+
+class FileDeleteTrigger(BaseEventTrigger):
+ """
+ A trigger that fires exactly once after it finds the requested file and
then delete the file.
+
+ The difference between ``FileTrigger`` and ``FileDeleteTrigger`` is
``FileDeleteTrigger`` can only find a
+ specific file.
+
+ :param filepath: File (relative to the base path set within the
connection).
+ :param poke_interval: Time that the job should wait in between each try
+ """
+
+ def __init__(
+ self,
+ filepath: str,
+ poke_interval: float = 5.0,
+ **kwargs,
+ ):
+ super().__init__()
+ self.filepath = filepath
+ self.poke_interval = poke_interval
+
+ def serialize(self) -> tuple[str, dict[str, Any]]:
+ """Serialize FileDeleteTrigger arguments and classpath."""
+ return (
+ "airflow.providers.standard.triggers.file.FileDeleteTrigger",
+ {
+ "filepath": self.filepath,
+ "poke_interval": self.poke_interval,
+ },
+ )
+
+ async def run(self) -> AsyncIterator[TriggerEvent]:
+ """Loop until the relevant file is found."""
+ while True:
+ if os.path.isfile(self.filepath):
+ mod_time_f = os.path.getmtime(self.filepath)
+ mod_time =
datetime.datetime.fromtimestamp(mod_time_f).strftime("%Y%m%d%H%M%S")
+ self.log.info("Found file %s last modified: %s",
self.filepath, mod_time)
+ os.remove(self.filepath)
Review Comment:
Sounds good to me
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
