Function getServiceCronsDynamic

packages/python/src/crons.ts:122–181

(opts: {
  serviceName: string;
  cronEntrypoint: string;
  handlerFunction?: string;
  pythonBin: string;
  env: NodeJS.ProcessEnv;
  workPath: string;
})

}

async function getServiceCronsDynamic(opts: {
  serviceName: string;
  cronEntrypoint: string;
  handlerFunction?: string;
  pythonBin: string;
  env: NodeJS.ProcessEnv;
  workPath: string;
}): Promise<ServiceCronEntry[]> {
  const {
    serviceName,
    cronEntrypoint,
    handlerFunction,
    pythonBin,
    env,
    workPath,
  } = opts;

  if (!handlerFunction) {
    throw new NowBuildError({
      code: 'PYTHON_DYNAMIC_CRON_NO_HANDLER',
      message:
        'Dynamic cron detection requires a "module:object" entrypoint where the object has a get_crons() method.',
    });
  }

  const moduleName = entrypointToModule(cronEntrypoint);

  const entries = await detectDynamicCrons({
    pythonBin,
    env,
    workPath,
    moduleName,
    attrName: handlerFunction,
  });

  console.log(
    `Detected ${entries.length} cron entry(s): ${entries.map(e => `${e.module_function} (${e.schedule})`).join(', ')}`
  );

  if (entries.length === 0) {
    throw new NowBuildError({
      code: 'PYTHON_DYNAMIC_CRON_EMPTY',
      message:
        `Dynamic cron detection returned no entries. ` +
        `"${moduleName}.${handlerFunction}.get_crons()" returned an empty iterable.`,
    });
  }

  return entries.map(entry => {
    const cronPath = moduleColonFuncToCronPath(
      serviceName,
      entry.module_function
    );
    return {
      path: cronPath,
      schedule: entry.schedule,
      resolvedHandler: entry.module_function,
    };