class_docstr: Save Rendered Template Fields.
docstr: Keep only Last X (num_to_keep) number of records for a task by deleting others.
  
  In the case of data for a mapped task either all of the rows or none of the rows will be deleted, so
  we don't end up with partial data for a set of mapped Task Instances left in the database.
  
  :param task_id: Task ID
  :param dag_id: Dag ID
  :param num_to_keep: Number of Records to keep
  :param session: SqlAlchemy Session
docstr_tok: ["Keep","only","Last","X","(","num_to_keep",")","number","of","records","for","a","task","by","deleting","others",".","In","the","case","of","data","for","a","mapped","task","either","all","of","the","rows","or","none","of","the","rows","will","be","deleted",",","so","we","do","n't","end","up","with","partial","data","for","a","set","of","mapped","Task","Instances","left","in","the","database",".",":","param","task_id",":","Task","ID",":","param","dag_id",":","Dag","ID",":","param","num_to_keep",":","Number","of","Records","to","keep",":","param","session",":","SqlAlchemy","Session"]
code: def delete_old_records(
      cls,
      task_id: str,
      dag_id: str,
      num_to_keep: int = conf.getint(
          "core", "max_num_rendered_ti_fields_per_task", fallback=0
      ),
      session: Session = NEW_SESSION,
  ) -> None:
      """
      Keep only Last X (num_to_keep) number of records for a task by deleting others.
  
      In the case of data for a mapped task either all of the rows or none of the rows will be deleted, so
      we don't end up with partial data for a set of mapped Task Instances left in the database.
  
      :param task_id: Task ID
      :param dag_id: Dag ID
      :param num_to_keep: Number of Records to keep
      :param session: SqlAlchemy Session
      """
      if num_to_keep <= 0:
          return
  
      from airflow.models.dagrun import DagRun
  
      tis_to_keep_query = (
          select(
              cls.dag_id, cls.task_id, cls.run_id, DagRun.execution_date
          )
          .where(cls.dag_id == dag_id, cls.task_id == task_id)
          .join(cls.dag_run)
          .distinct()
          .order_by(DagRun.execution_date.desc())
          .limit(num_to_keep)
      )
  
      cls._do_delete_old_records(
          dag_id=dag_id,
          task_id=task_id,
          ti_clause=tis_to_keep_query.subquery(),
          session=session,
      )
      session.flush()
  
imports: from __future__ import annotations
  import os
  from typing import TYPE_CHECKING
  import sqlalchemy_jsonfield
  from sqlalchemy import Column, ForeignKeyConstraint, Integer, PrimaryKeyConstraint, delete, exists, select, text
  from sqlalchemy.ext.associationproxy import association_proxy
  from sqlalchemy.orm import relationship
  from airflow.configuration import conf
  from airflow.models.base import StringID, TaskInstanceDependencies
  from airflow.serialization.helpers import serialize_template_field
  from airflow.settings import json
  from airflow.utils.retries import retry_db_transaction
  from airflow.utils.session import NEW_SESSION, provide_session
label_desc: Use image node_id 7 for calling the RenderedTaskInstanceFields obj's underlying member method code with example usage: obj.delete_old_records(cls, task_id, dag_id, num_to_keep, session) without return types