ancient-wolf-19325
06/25/2024, 6:09 PMFileSensor to monitor it within a test workflow. Here's a concise example:
1. Create a temporary file: Use Python's tempfile module to create a temporary file.
2. Define the sensor and tasks: Define a FileSensor and a simple task that the sensor will trigger.
3. Test the sensor: Write a test function to validate the sensor's behavior.
Example:
import tempfile
from flytekit import task, workflow
from flytekit.sensor.file_sensor import FileSensor
def test_sensor_task():
sensor = FileSensor(name="test_sensor")
tmp_file = tempfile.NamedTemporaryFile()
@task()
def t1():
print("flyte")
@workflow
def wf():
sensor(tmp_file.name) >> t1()
if __name__ == "__main__":
wf()
This example demonstrates how to set up and test a FileSensor by creating a temporary file and defining a simple task that gets triggered by the sensor.
Would you like more details on any specific part of this process?
Sources:
- Flytekit File Sensor Test Example
Slack Link
https://flyte-org.slack.com/archives/CP2HDHKE1/p1719338974.415999 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.