Function detectDynamicCrons

packages/python/src/crons.ts:186–228

* Call a Python function to dynamically detect cron entries.

(opts: {
  pythonBin: string;
  env: NodeJS.ProcessEnv;
  workPath: string;
  moduleName: string;
  attrName: string;
})

 * Call a Python function to dynamically detect cron entries.
 */
async function detectDynamicCrons(opts: {
  pythonBin: string;
  env: NodeJS.ProcessEnv;
  workPath: string;
  moduleName: string;
  attrName: string;
}): Promise<DynamicCronEntry[]> {
  const { pythonBin, env, workPath, moduleName, attrName } = opts;

  let stdout: string;
  try {
    const result = await execa(
      pythonBin,
      ['-c', script, moduleName, attrName],
      { env, cwd: workPath }
    );
    stdout = result.stdout;
  } catch (err: any) {
    // The Python script writes structured JSON errors to stdout before
    // exiting non-zero. Prefer that over execa's generic error.
    let detail = err?.stderr || err?.message || String(err);
    try {
      const parsed = JSON.parse(err?.stdout) as DynamicCronResult;
      if (parsed.error) detail = parsed.error;
    } catch {}
    throw new NowBuildError({
      code: 'PYTHON_DYNAMIC_CRON_DETECTION_FAILED',
      message: `Failed to detect dynamic cron entries: ${detail}`,
    });
  }

  let parsed: DynamicCronResult;
  try {
    parsed = JSON.parse(stdout);
  } catch {
    throw new NowBuildError({
      code: 'PYTHON_DYNAMIC_CRON_DETECTION_FAILED',
      message: `Dynamic cron detection returned invalid JSON: ${stdout}`,
    });
  }

  return parsed.entries || [];
}

/**
 * Convert a module:function string into a cron path.