Repository: apache/airflow
File: airflow-core/src/airflow/models/pool.py (Python)

Code before:
from sqlalchemy.orm import Mapped

from airflow.exceptions import AirflowException, PoolNotFound
from airflow.models.base import Base
from airflow.ti_deps.dependencies_states import EXECUTION_STATES
from airflow.utils.db import exists_query
from airflow.utils.session import NEW_SESSION, provide_session
from airflow.utils.sqlalchemy import mapped_column, with_row_locks
from airflow.utils.state import TaskInstanceState

if TYPE_CHECKING:
    from sqlalchemy.orm.session import Session
    from sqlalchemy.sql import Select


def validate_pool_name(name: str) -> None:
    """
    Validate that pool name only contains valid characters for stats reporting.

    Pool names must only contain ASCII alphabets, numbers, underscores, dots, and dashes
    to ensure compatibility with stats naming requirements.

    :param name: The pool name to validate
    :raises ValueError: If the pool name contains invalid characters
    """
    if not re.match(r"^[a-zA-Z0-9_.-]+$", name):
        raise ValueError(
            f"Pool name '{name}' is invalid. Pool names must only contain "
            "ASCII alphabets (a-z, A-Z), numbers (0-9), underscores (_), dots (.), and dashes (-)."
        )


class PoolStats(TypedDict):
    """Dictionary containing Pool Stats."""

    total: int | float  # Note: float("inf") is used to mark infinite slots
    running: int
    deferred: int
    queued: int
    open: int | float  # Note: float("inf") is used to mark infinite slots
    scheduled: int


class Pool(Base):
    """the class to get Pool info."""

    __tablename__ = "slot_pool"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    pool: Mapped[str] = mapped_column(String(256), unique=True)
    # -1 for infinite

Review comment (refactor):
Just a thought : Do we need this additional validation method ? Is it possible to refactor a code a bit and use the same validation logic used while reporting stats ? 
Else, any new validation rule would have to be updated here

Code after:
from sqlalchemy import Boolean, ForeignKey, Integer, String, Text, func, select
from sqlalchemy.orm import Mapped

from airflow.exceptions import AirflowException, PoolNotFound
from airflow.models.base import Base
from airflow.ti_deps.dependencies_states import EXECUTION_STATES
from airflow.utils.db import exists_query
from airflow.utils.session import NEW_SESSION, provide_session
from airflow.utils.sqlalchemy import mapped_column, with_row_locks
from airflow.utils.state import TaskInstanceState

if TYPE_CHECKING:
    from sqlalchemy.orm.session import Session
    from sqlalchemy.sql import Select

logger = logging.getLogger(__name__)


def normalize_pool_name_for_stats(name: str) -> str:
    """
    Normalize pool name for stats reporting by replacing invalid characters.

    Stats names must only contain ASCII alphabets, numbers, underscores, dots, and dashes.
    Invalid characters are replaced with underscores.

    :param name: The pool name to normalize
    :return: Normalized pool name safe for stats reporting
    """
    # Check if normalization is needed
    if re.match(r"^[a-zA-Z0-9_.-]+$", name):
        return name

    # Replace invalid characters with underscores
    normalized = re.sub(r"[^a-zA-Z0-9_.-]", "_", name)

    # Log warning
    logger.warning(
        "Pool name '%s' contains invalid characters for stats reporting. "
        "Reporting stats with normalized name '%s'. "
        "Consider renaming the pool to avoid this warning.",
        name,
        normalized,
    )

    return normalized


class PoolStats(TypedDict):
    """Dictionary containing Pool Stats."""

    total: int | float  # Note: float("inf") is used to mark infinite slots