Repository: apache/airflow
File: airflow-core/src/airflow/api_fastapi/core_api/datamodels/pools.py (Python)

Code before:
def _call_function(function: Callable[[], int]) -> int:
    """
    Call the given function.

    Used for the BeforeValidator to get the actual values from the bound method.
    """
    return function()


PoolSlots = Annotated[
    int,
    BeforeValidator(lambda v: -1 if isinstance(v, str) and v.lower() in {"inf", "infinity", "∞"} else v),
    Field(ge=-1, description="Number of slots. Use -1 for unlimited."),
]


class BasePool(BaseModel):
    """Base serializer for Pool."""

    pool: str = Field(serialization_alias="name")
    slots: PoolSlots
    description: str | None = Field(default=None)
    include_deferred: bool


def _sanitize_open_slots(value: int | float) -> int | None:
    # JSON does not support Infinity
    if isinstance(value, float) and value == float("inf"):
        return None
    return value


class PoolResponse(BasePool):
    """Pool serializer for responses."""

    occupied_slots: Annotated[int, BeforeValidator(_call_function)]
    running_slots: Annotated[int, BeforeValidator(_call_function)]
    queued_slots: Annotated[int, BeforeValidator(_call_function)]
    scheduled_slots: Annotated[int, BeforeValidator(_call_function)]
    open_slots: Annotated[int | float, BeforeValidator(lambda v: _sanitize_open_slots(_call_function(v)))]
    deferred_slots: Annotated[int, BeforeValidator(_call_function)]
    team_name: str | None


class PoolCollectionResponse(BaseModel):
    """Pool Collection serializer for responses."""

    pools: Iterable[PoolResponse]
    total_entries: int



Review comment (suggestion):
Type is not good here I think for a before validator. It should receive Any I believe.

Code after:
def _call_function(function: Callable[[], int]) -> int:
    """
    Call the given function.

    Used for the BeforeValidator to get the actual values from the bound method.
    """
    return function()


PoolSlots = Annotated[
    int,
    Field(ge=-1, description="Number of slots. Use -1 for unlimited."),
]


class BasePool(BaseModel):
    """Base serializer for Pool."""

    pool: str = Field(serialization_alias="name")
    slots: PoolSlots
    description: str | None = Field(default=None)
    include_deferred: bool


def _sanitize_open_slots(value) -> int:
    if isinstance(value, float) and value == float("inf"):
        return -1
    return value


class PoolResponse(BasePool):
    """Pool serializer for responses."""

    occupied_slots: Annotated[int, BeforeValidator(_call_function)]
    running_slots: Annotated[int, BeforeValidator(_call_function)]
    queued_slots: Annotated[int, BeforeValidator(_call_function)]
    scheduled_slots: Annotated[int, BeforeValidator(_call_function)]
    open_slots: Annotated[int, BeforeValidator(lambda v: _sanitize_open_slots(_call_function(v)))]
    deferred_slots: Annotated[int, BeforeValidator(_call_function)]
    team_name: str | None


class PoolCollectionResponse(BaseModel):
    """Pool Collection serializer for responses."""

    pools: Iterable[PoolResponse]
    total_entries: int


class PoolPatchBody(StrictBaseModel):
    """Pool serializer for patch bodies."""