@Marvin I want to use my own logger, for this I am using loguru, how to make this work ?

You can use your own logger with Prefect by overriding the logging configuration. This can be done by setting an environment variable or Prefect Profile setting using the syntax

PREFECT_LOGGING_[PATH]_[TO]_[KEY]

, where

[PATH]_[TO]_[KEY]

corresponds to the nested address of any setting. For example, to change the default logging levels for Prefect to

DEBUG

, you can set the environment variable

PREFECT_LOGGING_LEVEL="DEBUG"

. You can also configure the "root" Python logger, which receives logs from all loggers unless they explicitly opt out by disabling propagation. By default, the root logger is configured to output

WARNING

level logs to the console. As with other logging settings, you can override this from the environment or Prefect Profile. For more detailed information, you can refer to the Prefect logging guide.

@Marvin I am trying to make this work

import os
import sys

import httpx
from prefect import flow

from loguru import logger
@flow
def get_repo_info(repo_name: str = "PrefectHQ/prefect"):
    url = f"<https://api.github.com/repos/{repo_name}>"
    response = httpx.get(url)
    response.raise_for_status()
    repo = response.json()
    logger.info(f"{repo_name} repository statistics 🤓:")
    logger.info(f"Stars 🌠 : {repo['stargazers_count']}")
    logger.info(f"Forks 🍴 : {repo['forks_count']}")



if __name__ == "__main__":
    LOG_LEVEL = "INFO"
    WORKING_DIR = os.getenv(
        "PL_GF_WORKING_DIR",
        os.path.dirname(os.path.dirname(os.path.dirname(os.path.realpath(__file__)))),
    )
    logger.remove()
    logger.add(sys.stderr, level=LOG_LEVEL)
    logger.add(
        os.path.join(WORKING_DIR, "logs", "automation_cfn_{time}.log"),
        enqueue=True,
        level=LOG_LEVEL,
    )
    get_repo_info(repo_name="PrefectHQ/prefect")

Your code seems fine at first glance. However, there are a few things you might want to check: 1. Make sure you have the necessary permissions to write to the log file. The path you're using to store the log file might not exist or you might not have write permissions. 2. Ensure that the

loguru

and

httpx

packages are installed in your environment. If not, you can install them using pip:

bash
    pip install loguru httpx

3. The URL in your

get_repo_info

function seems to have

<

and

>

around it, which might cause an issue. The correct URL should be:

url = f"<https://api.github.com/repos/{repo_name}>"

4. The

get_repo_info

function is decorated with

@flow

, which is not the correct way to define a flow in Prefect. You should use the

Flow

context manager to define a flow and add tasks to it. Here's an example:

from prefect import Flow, task

    @task
    def get_repo_info(repo_name: str = "PrefectHQ/prefect"):
        # your code here

    with Flow("My Flow") as flow:
        get_repo_info()

5. Finally, you're not running the flow. After defining it, you should run it using the

run

method:

flow.run()

Please make these changes and let me know if you're still facing issues.

@Marvin still not working, this is my code now

import os
import sys
import time

import httpx
from loguru import logger
from prefect import flow, task
from prefect.task_runners import SequentialTaskRunner
@task
def get_repo_info(repo_name: str = "PrefectHQ/prefect"):
    url = f"<https://api.github.com/repos/{repo_name}>"
    response = httpx.get(url)
    response.raise_for_status()
    repo = response.json()
    <http://logger.info|logger.info>(f"{repo_name} repository statistics 🤓:")
    <http://logger.info|logger.info>(f"Stars 🌠 : {repo['stargazers_count']}")
    <http://logger.info|logger.info>(f"Forks 🍴 : {repo['forks_count']}")


@task
def print_values(values):
    for value in values:
        time.sleep(0.5)
        print(value, end="\r")


@flow(task_runner=SequentialTaskRunner())
def my_flow_1():
    print_values.submit(["AAAA"] * 15)
    print_values.submit(["BBBB"] * 10)
    get_repo_info()


if __name__ == "__main__":
    LOG_LEVEL = "INFO"
    WORKING_DIR = os.getenv(
        "PL_GF_WORKING_DIR",
        "C:/Users/andre/Predictive_Layer/prefect_test",
    )
    logger.remove()
    logger.add(sys.stderr, level=LOG_LEVEL)
    logger.add(
        os.path.join(WORKING_DIR, "logs", "prefect_{time}.log"),
        enqueue=True,
        level=LOG_LEVEL,
    )
    my_flow_1()

And this is the output 155359.607 | INFO | prefect.engine - Created flow run 'quiet-albatross' for flow 'my-flow-1' 155359.679 | INFO | Flow run 'quiet-albatross' - Created task run 'print_values-0' for task 'print_values' 155359.680 | INFO | Flow run 'quiet-albatross' - Executing 'print_values-0' immediately... 155407.245 | INFO | Task run 'print_values-0' - Finished in state Completed() 155407.260 | INFO | Flow run 'quiet-albatross' - Created task run 'print_values-1' for task 'print_values' 155407.261 | INFO | Flow run 'quiet-albatross' - Executing 'print_values-1' immediately... 155412.316 | INFO | Task run 'print_values-1' - Finished in state Completed() 155412.329 | INFO | Flow run 'quiet-albatross' - Created task run 'get_repo_info-0' for task 'get_repo_info' 155412.330 | INFO | Flow run 'quiet-albatross' - Executing 'get_repo_info-0' immediately... 2023-11-30 155412.629 | INFO | __main__get repo info17 - PrefectHQ/prefect repository statistics 🤓: 2023-11-30 155412.630 | INFO | __main__get repo info18 - Stars 🌠 : 13421 2023-11-30 155412.630 | INFO | __main__get repo info19 - Forks 🍴 : 1386 155412.648 | INFO | Task run 'get_repo_info-0' - Finished in state Completed() 155412.668 | INFO | Flow run 'quiet-albatross' - Finished in state Completed('All states completed.') But I don't see the logs from the get_repo_info function in the flow_run interface

The Prefect UI doesn't automatically display logs from custom loggers like

loguru

. Prefect uses the standard Python

logging

module for its logs, and the UI is designed to display these logs. When you use a custom logger like

loguru

, the logs are not captured by the standard

logging

module, and therefore they don't appear in the Prefect UI. If you want your logs to appear in the Prefect UI, you'll need to use the standard Python

logging

module or configure

loguru

to work with it. Here's an example of how you can configure

loguru

to work with the standard

logging

module:

from loguru import logger
import logging

class PropagateHandler(logging.Handler):
    def emit(self, record):
        logging.getLogger(record.name).handle(record)

logger.add(PropagateHandler(), format="{message}")

This will add a handler to

loguru

that propagates logs to the standard

logging

module, which should make them appear in the Prefect UI. Please note that this might not be the perfect solution, as

loguru

and the standard

logging

module have some differences in how they handle logs. You might need to adjust this solution to fit your needs. Also, remember to replace

<http://logger.info|logger.info>

with

<http://logger.info|logger.info>

in your

get_repo_info

function. The

<http://logger.info|logger.info>

syntax is not valid Python code and will cause an error.

@Marvin I still don't see them in the ui all I see in the ui is Nov 30th, 2023 INFO Created task run 'print_values-0' for task 'print_values' 035942 PM prefect.flow_runs INFO Executing 'print_values-0' immediately... 035942 PM prefect.flow_runs INFO Finished in state Completed() 035949 PM print_values-0 prefect.task_runs INFO Created task run 'print_values-1' for task 'print_values' 035949 PM prefect.flow_runs INFO Executing 'print_values-1' immediately... 035949 PM prefect.flow_runs INFO Finished in state Completed() 035954 PM print_values-1 prefect.task_runs INFO Created task run 'get_repo_info-0' for task 'get_repo_info' 035954 PM prefect.flow_runs INFO Executing 'get_repo_info-0' immediately... 035954 PM prefect.flow_runs INFO Finished in state Completed() 035955 PM get_repo_info-0 prefect.task_runs INFO Finished in state Completed('All states completed.') 035955 PM prefect.flow_run

Based on the issues you're encountering and the information I found in the GitHub issues, it seems that Prefect's logging system doesn't currently support custom loggers in a way that would allow their logs to be displayed in the UI. This is a known limitation and there are open issues related to this topic. Here are some potential workarounds: 1. Use Prefect's logger: Instead of using

loguru

, use Prefect's built-in logger. You can get it by calling

prefect.utilities.logging.get_logger()

. This logger's messages should appear in the UI. 2. Log to a file and read it in another task: If you need to use

loguru

, you could log to a file and then read the file in another task. You could then log the contents of the file using Prefect's logger, which would make them appear in the UI. 3. Use a custom logging handler: You could create a custom logging handler that sends logs to Prefect's logger. This would involve creating a subclass of

logging.Handler

and implementing the

emit

method to send logs to Prefect's logger. Please note that these are workarounds and might not work in all situations. The best solution would be for Prefect to add support for custom loggers, but this is not currently available.

@Marvin in a similar question you suggested something like this

if not any(isinstance(handler, APILogHandler) for handler in logger.handlers):
    api_log_handler = APILogHandler()
    logger.addHandler(api_log_handler)

can I use that now ?

The code snippet you provided is a way to add a custom log handler to the standard Python logger. The

APILogHandler

is a custom handler that presumably sends logs to an API. In your case, you want to send logs from

loguru

to Prefect's logger so they appear in the Prefect UI. To do this, you would need to create a custom handler that sends logs to Prefect's logger. Here's an example of how you can do this:

from loguru import logger
import logging
import prefect

class PrefectLogHandler(logging.Handler):
    def emit(self, record):
        prefect_logger = prefect.utilities.logging.get_logger(record.name)
        prefect_logger.handle(record)

logger.add(PrefectLogHandler(), format="{message}")

This will add a handler to

loguru

that sends logs to Prefect's logger. However, please note that this might not work as expected. Prefect's logging system is complex and might not handle logs from

loguru

correctly. This is a workaround and might not be a perfect solution.

@Marvin AttributeError: module 'prefect.utilities' has no attribute 'logging'

I apologize for the confusion. The correct way to get the Prefect logger is

prefect.utilities.logging.get_logger()

. Here's the corrected code:

from loguru import logger
import logging
import prefect.utilities.logging

class PrefectLogHandler(logging.Handler):
    def emit(self, record):
        prefect_logger = prefect.utilities.logging.get_logger(record.name)
        prefect_logger.handle(record)

logger.add(PrefectLogHandler(), format="{message}")

This code adds a handler to

loguru

that sends logs to Prefect's logger. However, as I mentioned before, this might not work as expected due to the complexities of Prefect's logging system. This is a workaround and might not be a perfect solution.

@Marvin this is working if I do

import os
import sys

import httpx
from prefect import flow, task
from loguru import logger
from prefect.logging.handlers import APILogHandler
@task(retries=2)
def get_repo_info(repo_owner: str, repo_name: str):
    """Get info about a repo - will retry twice after failing"""
    url = f"<https://api.github.com/repos/{repo_owner}/{repo_name}>"
    api_response = httpx.get(url)
    api_response.raise_for_status()
    repo_info = api_response.json()
    return repo_info


@task
def get_contributors(repo_info: dict):
    contributors_url = repo_info["contributors_url"]
    response = httpx.get(contributors_url)
    response.raise_for_status()
    contributors = response.json()
    return contributors


@flow(name="Repo Info")
def repo_info(repo_owner: str = "PrefectHQ", repo_name: str = "prefect"):
    """
    Given a GitHub repository, logs the number of stargazers
    and contributors for that repo.
    """
    repo_info = get_repo_info(repo_owner, repo_name)
    <http://logger.info|logger.info>(f"Stars 🌠 : {repo_info['stargazers_count']}")

    contributors = get_contributors(repo_info)
    <http://logger.info|logger.info>(f"Number of contributors 👷: {len(contributors)}")


if __name__ == "__main__":
    LOG_LEVEL = "INFO"
    WORKING_DIR = os.getenv(
        "PL_GF_WORKING_DIR",
        "C:/Users/andre/Predictive_Layer/prefect_test",
    )
    logger.remove()
    logger.add(sys.stderr, level=LOG_LEVEL)
    logger.add(
        os.path.join(WORKING_DIR, "logs", "first_deployment_{time}.log"),
        enqueue=True,
        level=LOG_LEVEL,
    )
    api_log_handler = APILogHandler()
    logger.add(api_log_handler)
    repo_info()

but if I try to do a deployment with this

# create your first deployment
repo_info.serve(name="my-first-deployment_new")

I don't see the logs anymore

When you deploy a flow using Prefect Cloud or Prefect Server, the flow runs in a different environment, which might not have access to the same logging configuration as your local environment. This is why you're not seeing the logs from

loguru

in the Prefect UI after deploying the flow. The

serve

method you're using to deploy the flow is not a standard Prefect method and seems to be specific to your environment or a custom wrapper you're using. Therefore, it's hard for me to provide specific advice on how to configure logging for deployed flows in your case. However, here are some general suggestions: 1. Configure logging in the flow: Instead of configuring logging in the

if __name__ == "__main__":

block, you could try configuring it inside the flow or in a separate task that runs at the start of the flow. This might ensure that the logging configuration is applied when the flow runs, not just when you run the script locally. 2. Use Prefect's logger: As I mentioned before, Prefect's logging system is designed to work with the standard Python

logging

module. If you use Prefect's logger, the logs should appear in the Prefect UI, even for deployed flows. You can get Prefect's logger by calling

prefect.utilities.logging.get_logger()

. 3. Check the deployment environment: Check the environment where the flow runs after deployment. This environment might have different logging settings or might not have

loguru

installed. You might need to adjust the logging configuration or install

loguru

in this environment. Please note that these are general suggestions and might not work in your specific case. The best solution would depend on the details of your deployment environment and how you're using Prefect.

@Marvin ok I am trying different things now, but sitll no luck

import logging
import sys
import os
from prefect.logging.handlers import APILogHandler
log_file_path = os.path.join("C:/Users/andre/Predictive_Layer/prefect_test/logs", "prefect.log")
os.makedirs(os.path.dirname(log_file_path), exist_ok=True)
log_format = "%(asctime)s - %(levelname)s - %(message)s"
logging.basicConfig(
    level=<http://logging.INFO|logging.INFO>,
    format=log_format,
    handlers=[
        logging.StreamHandler(sys.stdout),
        logging.FileHandler(log_file_path),
        APILogHandler()
    ],
)

logger = logging.getLogger(__name__)

import httpx
from prefect import flow, task
from utils.logger import logger

@task(retries=2)
def get_repo_info(repo_owner: str, repo_name: str):
    """Get info about a repo - will retry twice after failing"""
    url = f"<https://api.github.com/repos/{repo_owner}/{repo_name}>"
    api_response = httpx.get(url)
    api_response.raise_for_status()
    repo_info = api_response.json()
    return repo_info


@task
def get_contributors(repo_info: dict):
    contributors_url = repo_info["contributors_url"]
    response = httpx.get(contributors_url)
    response.raise_for_status()
    contributors = response.json()
    return contributors


@flow(name="Repo Info")
def repo_info(repo_owner: str = "PrefectHQ", repo_name: str = "prefect"):
    """
    Given a GitHub repository, logs the number of stargazers
    and contributors for that repo.
    """
    repo_info = get_repo_info(repo_owner, repo_name)
    <http://logger.info|logger.info>(f"Stars 🌠 : {repo_info['stargazers_count']}")

    contributors = get_contributors(repo_info)
    <http://logger.info|logger.info>(f"Number of contributors 👷: {len(contributors)}")


if __name__ == "__main__":
    <http://logger.info|logger.info>("Starting flow run!!!!!")
    repo_info()

Nothing is being logged either lcoally or remotely in the interface

Based on the Prefect documentation, it seems that Prefect uses its own logging configuration and provides a logger that you can use in your flows and tasks. Here's how you can use it:

from prefect import get_logger

logger = get_logger()

@task
def my_task():
    <http://logger.info|logger.info>("This is an info message")

This logger should work with the Prefect UI and show logs there. However, it might not have all the features of

loguru

that you're using. If you want to use a custom logger with Prefect, you can customize the logging configuration by creating your own version of

logging.yml

with custom settings. You can specify the path to this file with the

PREFECT_LOGGING_SETTINGS_PATH

environment variable. If the file does not exist at the specified location, Prefect ignores the setting and uses the default configuration. However, please note that this might not make logs from

loguru

appear in the Prefect UI. The Prefect UI is designed to work with the standard Python

logging

module and might not support custom loggers like

loguru

. For more information, you can refer to the Prefect logging guide.