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

Code before:


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


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

    name: str | None = Field(default=None, alias="pool")
    slots: PoolSlots | None = None
    description: str | None = None
    include_deferred: bool | None = None
    team_name: str | None = Field(max_length=50, default=None)


class PoolBody(BasePool, StrictBaseModel):
    """Pool serializer for post bodies."""

    pool: str = Field(alias="name", max_length=256)

Review comment (security):
Why 'or float' ? the beforevalidator will return None or int ? 
```suggestion
    open_slots: Annotated[int, BeforeValidator(lambda v: _sanitize_open_slots(_call_function(v)))]
```

Code after:

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."""

    name: str | None = Field(default=None, alias="pool")
    slots: PoolSlots | None = None
    description: str | None = None
    include_deferred: bool | None = None
    team_name: str | None = Field(max_length=50, default=None)


class PoolBody(BasePool, StrictBaseModel):
    """Pool serializer for post bodies."""

    pool: str = Field(alias="name", max_length=256)
    description: str | None = None
    include_deferred: bool = False