How do I run python code on each task in Airflow

I have a sequence of Tasks on my DAG

eg.

task1 >> task2 >> task3

What I need is to put across the tasks functionality without repeating code.
For example logging, would like to execute log.info(task_id) per task
Or some external operation, like http call per task

I am new to airflow and probably missing all the concepts, but how I would do something like this?
a custom operator?
An interceptor for each task execution.

I would like to avoid something like this

[task1,logging] >> [task2,logging] >> [task3, logging]

Another example
If I try to process the xcom params stored per task with a common way how would I do it? Instead of putting the process logic on every task I would like to have the xcom per task on this seperate operator/task/whatever will be to process, same goes for logging or other functionalities

Triggering my operator/task/whatever per task i guess would be one way to do this, but again do I need to add it every time like this on each task?

[task1,process_xcom] >> [task2,process_xcom] >> [task3, process_xcom]

> A similar concept in java could be the usage of Aspects on Services

Few option that I think can help in this case :

1. you can set in the dag default_args "on_execute_callback" that each task before it starts will call this function or "on_success_callback"/"on_failure_callback" when it ends.

the advantage here is that all tasks will call this function and you need to defined it once per dag.

    def log(context):
        print(context["ti"].task_id)

    with DAG(
       dag_id="test_dag",
       schedule_interval=None,
       default_args={
          "start_date": datetime(2022, 1, 1),
          "on_execute_callback": log,
          "catchup": False,
    },
  ) as dag:

2.another way is, on each task you can call pre_execute and/or post_execute, for example if you want to do something with the xcom.

in this example I push all json keys to xcom as separate key

def push_json_xcom(context, result):
    ti = context["ti"]
    data = result if isinstance(result, dict) else json.loads(result)
    for key, value in data.items():
        ti.xcom_push(key, value)

PythonOperator(task_id="test_id", post_execute=push_json_xcom, python_callable=python_func)