Logging

Hatchet comes with a built-in logging view where you can push logs from your workflows. This is useful for debugging and monitoring your workflows.

You can use either Python’s built-in logging package, or the context.log method for more control over the logs that are sent.

Using the built-in `logging` package

You can pass a custom logger to the Hatchet class when initializing it. For example:

examples/logger/client.py



import logging

from hatchet_sdk import ClientConfig, Hatchet

logging.basicConfig(level=logging.INFO)

root_logger = logging.getLogger()

hatchet = Hatchet(
    debug=True,
    config=ClientConfig(
        logger=root_logger,
    ),
)

It’s recommended that you pass the root logger to the Hatchet class, as this will ensure that all logs are captured by the Hatchet logger. If you have workflows defined in multiple files, they should be children of the root logger. For example, with the following file structure:

client.py
worker.py
- workflow.py

You should pass the root logger to the Hatchet class in client.py:

examples/logger/client.py



import logging

from hatchet_sdk import ClientConfig, Hatchet

logging.basicConfig(level=logging.INFO)

root_logger = logging.getLogger()

hatchet = Hatchet(
    debug=True,
    config=ClientConfig(
        logger=root_logger,
    ),
)

And then in workflows/workflow.py, you should create a child logger:

examples/logger/workflow.py


import logging
import time

from examples.logger.client import hatchet
from hatchet_sdk import Context, EmptyModel

logger = logging.getLogger(__name__)

logging_workflow = hatchet.workflow(
    name="LoggingWorkflow",
)


@logging_workflow.task()
def root_logger(input: EmptyModel, ctx: Context) -> dict[str, str]:
    for i in range(12):
        logger.info("executed step1 - {}".format(i))
        logger.info({"step1": "step1"})

        time.sleep(0.1)

    return {"status": "success"}

Using the `context.log` method

You can also use the context.log method to log messages from your workflows. This method is available on the Context object that is passed to each task in your workflow. For example:

examples/logger/workflow.py



@logging_workflow.task()
def context_logger(input: EmptyModel, ctx: Context) -> dict[str, str]:
    for i in range(12):
        ctx.log("executed step1 - {}".format(i))
        ctx.log({"step1": "step1"})

        time.sleep(0.1)

    return {"status": "success"}

Each task is currently limited to 1000 log lines.

In TypeScript, there are two options for logging from your tasks. The first is to use the ctx.log() method (from the Context) to send logs:

src/examples/logger.ts


const workflow: Workflow = {
  id: 'logger-example',
  description: 'test',
  on: {
    event: 'user:create',
  },
  steps: [
    {
      name: 'logger-step1',
      run: async (ctx) => {
        // log in a for loop
        // eslint-disable-next-line no-plusplus
        for (let i = 0; i < 10; i++) {
          ctx.log(`log message ${i}`);
          await sleep(200);
        }

        return { step1: 'completed step run' };
      },
    },
  ],
};

This has the benefit of being easy to use out of the box (no setup required!), but it’s limited in its flexibiliy and how pluggable it is with your existing logging setup.

Hatchet also allows you to “bring your own” logger when you define a workflow:

src/examples/byo-logger.ts

const logger = pino();

class PinoLogger implements Logger {
  logLevel: LogLevel;
  context: string;

  constructor(context: string, logLevel: LogLevel = 'DEBUG') {
    this.logLevel = logLevel;
    this.context = context;
  }

  debug(message: string): void {
    logger.debug(message);
  }

  info(message: string): void {
    logger.info(message);
  }

  green(message: string): void {
    logger.info(`%c${message}`);
  }

  warn(message: string, error?: Error): void {
    logger.warn(`${message} ${error}`);
  }

  error(message: string, error?: Error): void {
    logger.error(`${message} ${error}`);
  }
}

const hatchet = Hatchet.init({
  log_level: 'DEBUG',
  logger: (ctx, level) => new PinoLogger(ctx, level),
});

In this example, we create Pino logger that implement’s Hatchet’s Logger interface and pass it to the Hatchet client constructor. We can then use that logger in our steps:

src/examples/byo-logger.ts


const sleep = (ms: number) =>
  new Promise((resolve) => {
    setTimeout(resolve, ms);
  });

const workflow: Workflow = {
  id: 'byo-logger-example',
  description: 'An example showing how to pass a custom logger to Hatchet',
  on: {
    event: 'byo-logger:spawn',
  },
  steps: [
    {
      name: 'logger-step1',
      run: async (ctx) => {
        // eslint-disable-next-line no-plusplus
        for (let i = 0; i < 5; i++) {
          logger.info(`log message ${i}`);
          await sleep(500);
        }

        return { step1: 'completed step run' };
      },
    },
  ],
};

Manual Slot Release OpenTelemetry

We use cookies

Logging

Using the built-in logging package

Using the context.log method

Using the built-in `logging` package

Using the `context.log` method