How to check whether a flow is being triggered manually

Help
, ,
1

I’m creating a flow that has several sub-flows. This flow will be scheduled weekly. Some of the sub-flows might be triggered manually for checking, I want to create a separate logs file just for this particular sub-flows that are being triggered manually, how to check if a flow is being triggered manually inside the python flow code?

2

hi @Muhammad_Aziz_Rahman

If you’re using Prefect Cloud, you can check the Event associated with the scheduling of the current flow run.

So from within a flow run, you could grab the ID from get_run_context and then make an API call to fetch prefect.flow-run.Scheduled events with the ID you found in the context.

In the event, you will find a related_resource like "prefect.resource.role": "creator"

  • if a deployment scheduled this flow run, the deployment will be the creator
  • if an actor created an ad-hoc run of this flow, they will be listed as the creator

While not exactly what you might want, this example of using the Events API should get you started

import pendulum
from httpx import AsyncClient

from prefect.events import Event
from prefect.settings import PREFECT_API_KEY, PREFECT_API_URL


async def read_flow_runs_scheduled_in_last_day(
    limit: int = 50,
    event_name: str = None,
    filter: dict = None,
) -> list[Event]:
    if event_name is None:
        event_name = "prefect.flow-run.Scheduled"
    
    if filter is None:
        filter = {
            "occurred": {
                "since": pendulum.now().subtract(days=1).isoformat(),
                "until": pendulum.now().isoformat(),
            },
            "event": {
                "name": [event_name]
            }
        }
    
    headers = {
        "Authorization": "Bearer " + PREFECT_API_KEY.value(),
        "Content-Type": "application/json",
    }
    
    async with AsyncClient() as client:
        response = await client.post(
            f"{PREFECT_API_URL.value()}/events/filter",
            headers=headers,
            json={
                "limit": limit,
                "filter": filter,
            },
        )
        response.raise_for_status()
        return response.json()
if __name__ == "__main__":
    events = await read_flow_runs_scheduled_in_last_day()
        
    print(f"Number of flow runs scheduled in the last 24 hours: {len(events)}")

We’ll be adding more examples of using the Events API to the docs soon!

Let us know if you have further questions!

3

Hi @nate

Thanks for answering, but unfortunately I’m currently using Prefect Server on my EC2 instance.

My code roughly looks like this

def main_flow():
    #Add File Handler to both flow and task logger
    flow_logger = logging.getLogger('prefect.flow_runs')
    task_logger = logging.getLogger('prefect.task_runs')
    
    if flow_trigger_manually:
           handler = logging.FileHandler(filename="path of logs if flow_trigger_manually")
    else:
           handler = logging.FileHandler(filename="path of logs if flow_trigger_by_schedule)

    handler.setFormatter(logging.Formatter("%(asctime)s.%(msecs)03d | %(levelname)-7s | %(name)s - %(message)s"))

    flow_logger.addHandler(handler)
    task_logger.addHandler(handler)

I want to separate the logs for the flow that is being triggered manually. My problem is how to find the condition for flow_trigger_manually?