Repository: apache/airflow
File: airflow-core/src/airflow/dag_processing/manager.py (Python)

Code before:
    parsing_result: DagFileParsingResult | None,
    session: Session,
    *,
    is_callback_only: bool = False,
    relative_fileloc: str | None = None,
) -> DagFileStat:
    """Take the parsing result and stats about the parser process and convert it into a DagFileStat."""
    if is_callback_only:
        # Callback-only processing - don't update timestamps to avoid stale DAG detection issues
        stat = DagFileStat(
            last_duration=run_duration,
            run_count=run_count,  # Don't increment for callback-only processing
        )
        Stats.incr("dag_processing.callback_only_count")
    else:
        # Actual DAG parsing or import error
        stat = DagFileStat(
            last_finish_time=finish_time,
            last_duration=run_duration,
            run_count=run_count + 1,
        )

    # Note: relative_fileloc has a None default. In practice it is always provided but code defensively here in case
    if relative_fileloc is not None and stat.last_duration is not None:
        file_name = Path(relative_fileloc).stem
        Stats.timing(f"dag_processing.last_duration.{file_name}", stat.last_duration)
        Stats.timing("dag_processing.last_duration", stat.last_duration, tags={"file_name": file_name})

    if parsing_result is None:
        # No DAGs were parsed - this happens for callback-only processing
        # Don't treat this as an import error when it's callback-only
        if not is_callback_only:
            stat.import_errors = 1
    else:
        # record DAGs and import errors to database
        import_errors = {}
        if parsing_result.import_errors:
            import_errors = {
                (bundle_name, rel_path): error for rel_path, error in parsing_result.import_errors.items()
            }

        # Build the set of files that were parsed. This includes the file that was parsed,
        # even if it no longer contains DAGs, so we can clear old import errors.
        files_parsed: set[tuple[str, str]] | None = None
        if relative_fileloc is not None:
            files_parsed = {(bundle_name, relative_fileloc)}
            files_parsed.update(import_errors.keys())

        if (warnings := parsing_result.warnings) and isinstance(warnings[0], dict):
            warnings = [DagWarning(**warn) for warn in warnings]


Review comment (security):
I also wonder if we need to sanitize the filename potentially, similar to `normalize_pool_name_for_stats`?

Code after:
    parsing_result: DagFileParsingResult | None,
    session: Session,
    *,
    is_callback_only: bool = False,
    relative_fileloc: str | None = None,
) -> DagFileStat:
    """Take the parsing result and stats about the parser process and convert it into a DagFileStat."""
    if is_callback_only:
        # Callback-only processing - don't update timestamps to avoid stale DAG detection issues
        stat = DagFileStat(
            last_duration=run_duration,
            run_count=run_count,  # Don't increment for callback-only processing
        )
        Stats.incr("dag_processing.callback_only_count")
    else:
        # Actual DAG parsing or import error
        stat = DagFileStat(
            last_finish_time=finish_time,
            last_duration=run_duration,
            run_count=run_count + 1,
        )

    # Note: relative_fileloc has a None default. In practice it is always provided but code defensively here in case
    if relative_fileloc is not None and stat.last_duration is not None:
        # Normalize names to ensure they only contain valid characters for stats (alphanumeric, underscore, dot, dash)
        file_name = normalize_name_for_stats(Path(relative_fileloc).stem)
        # bundle_name is included to distinguish files with the same name across different bundles
        normalized_bundle = normalize_name_for_stats(bundle_name)
        Stats.timing(f"dag_processing.last_duration.{normalized_bundle}.{file_name}", stat.last_duration)
        Stats.timing(
            "dag_processing.last_duration",
            stat.last_duration,
            tags={"file_name": file_name, "bundle_name": normalized_bundle},
        )

    if parsing_result is None:
        # No DAGs were parsed - this happens for callback-only processing
        # Don't treat this as an import error when it's callback-only
        if not is_callback_only:
            stat.import_errors = 1
    else:
        # record DAGs and import errors to database
        import_errors = {}
        if parsing_result.import_errors:
            import_errors = {
                (bundle_name, rel_path): error for rel_path, error in parsing_result.import_errors.items()
            }

        # Build the set of files that were parsed. This includes the file that was parsed,
        # even if it no longer contains DAGs, so we can clear old import errors.
        files_parsed: set[tuple[str, str]] | None = None