Daniel Burkhardt Cerigo · 933319c0 · 03899935 · f3eddd33 · d36ad9b8
--- a/python/projects/phiphi/phiphi/api/projects/job_runs/flow_runner_flow.py
+++ b/python/projects/phiphi/phiphi/api/projects/job_runs/flow_runner_flow.py
-    return job_params_dict
+        with platform_db.get_session() as session:
+            gather = gathers.crud.get_gather(
+                session=session, project_id=project_id, gather_id=job_source_id
+            )
+        if gather is None:
+            raise ValueError(f"Gather with ID {job_source_id=} not found.")
+        elif type(gather) == gathers.apify_facebook_posts.schemas.ApifyFacebookPostGatherResponse:
+            job_params = apify_input_schemas.ApifyFacebookPostsInput(
+                only_posts_older_than=gather.only_posts_older_than,
+                only_posts_newer_than=gather.only_posts_newer_than,
+                account_urls=gather.account_url_list,
+                results_per_url_limit=gather.limit_posts_per_account,
+            )
+        else:
+            raise NotImplementedError(f"Run for gather type {type(gather)=} not implemented yet.")
--- a/python/projects/phiphi/phiphi/api/projects/job_runs/flow_runner_flow.py
+++ b/python/projects/phiphi/phiphi/api/projects/job_runs/flow_runner_flow.py
+    if type(job_params) == apify_input_schemas.ApifyFacebookPostsInput:
+        deployment_name = "gather_apify_facebook_posts_flow/main_deployment"
+    else:
+        raise NotImplementedError(
+            f"Run for job_params type {type(job_params)=} not implemented yet."
+        )
+    job_run_flow: objects.FlowRun = asyncio.run(
+        deployments.run_deployment(
+            name=deployment_name,
+            parameters=job_params.dict(by_alias=True),
+            as_subflow=True,
+            timeout=0,  # this means it returns immediately with the metadata
+            tags=[
+                f"project_id:{project_id}",
+                f"job_type:{job_type}",
+                f"job_source_id:{job_source_id}",
--- a/python/projects/phiphi/phiphi/api/projects/job_runs/flow_runner_flow.py
+++ b/python/projects/phiphi/phiphi/api/projects/job_runs/flow_runner_flow.py
        job_run_id: ID of the row in the job_runs table.
-        job_run_flow: Flow run object for the job.
    """
-    _ = {
-        "prefect_inner_flow_run_id": job_run_flow.id,
-        "prefect_inner_flow_run_name": job_run_flow.name,
-        "prefect_inner_flow_run_status": "started",
-        "prefect_inner_flow_run_started_at": datetime.now(),
-    }
-    # Update row in job_runs table for `job_run_id`
+    job_run_update_started = job_runs.schemas.JobRunUpdateStarted(
+        id=job_run_id,
+        flow_run_id=str(runtime.flow_run.id),
+        flow_run_name=runtime.flow_run.name,
+    )
+    with platform_db.get_session() as session: