Add get_events_subscriber example

README.md

@@ -1,14 +1,14 @@
-# prefect-demos
+# Prefect Demos
 Welcome to our repository dedicated to showcasing a variety of Prefect demos. 
 
 These demos are meant to be end-to-end examples showcasing how multiple features within prefect can be utilized to accommodate different use cases demonstrating the versatility and power of Prefect as a workflow application tool. These demos focus specifically on the code and any prefect features necessary to create these examples, any external configurations necessary are assumed to have been completed separately. Dive into our demos to explore how Prefect seamlessly orchestrates complex data processes, ensuring efficient and reliable execution of your data tasks. 
 
 Get inspired, learn best practices, and discover innovative ways to leverage Prefect in your data projects!
 
-# Flows
+## Flows
 Demos separated by project.
 
-#### Datalake Usage
+### Datalake usage
 - [datalake_listener.py](flows/aws/datalake/datalake_listener.py)
 
     *Processes Near Earth Object (NEO) data from an AWS S3 bucket, filters for potentially hazardous objects, flattens the data structure, and then uploads the results back to S3 in CSV format*
@@ -21,9 +21,9 @@ Demos separated by project.
 
     *Automates the deployment of two Prefect workflows for data processing: `datalake_listener`, which triggers on AWS S3 object creation, and `fetch_neo_by_date`, which fetches Near Earth Object data daily, using a Docker image from an ECR repository for execution on an ECS push work pool*
 
+### Get events subscriber
 
+Subscribe to Prefect events and print out the events as they occur.
 
-
-
-
+- [get_events_subscriber.py](flows/get_events_subscriber.py)
 

flows/get_events_subscriber.py

@@ -0,0 +1,24 @@
+# get_events_subscriber takes an EventFilter as input.
+# prints Prefect events as they occur
+# creates an infinite loop 
+# so you'll need to add your own logic to exit it 
+# once you've seen the events you're looking for
+
+from prefect.events import get_events_subscriber
+from prefect import flow
+import asyncio
+
+
+@flow(log_prints=True)
+async def subs():
+    seen_events = set()
+    async with get_events_subscriber() as subscriber:
+        async for event in subscriber:
+            print(f"📬 Received event {event.event!r}")
+            seen_events.add(event.event)
+            if len(seen_events) > 5:
+                break
+
+
+if __name__ == "__main__":
+    asyncio.run(subs())