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

Code before:
    """
    Create a callable for Depends in FastAPI that returns a filter of the permitted pools for the user.

    :param method: whether filter readable or writable.
    """

    def depends_permitted_pools_filter(
        user: GetUserDep,
        auth_manager: AuthManagerDep,
    ) -> PermittedPoolFilter:
        authorized_pools: set[str] = auth_manager.get_authorized_pools(user=user, method=method)
        return PermittedPoolFilter(authorized_pools)

    return depends_permitted_pools_filter


ReadablePoolsFilterDep = Annotated[PermittedPoolFilter, Depends(permitted_pool_filter_factory("GET"))]


def requires_access_pool(method: ResourceMethod) -> Callable[[Request, BaseUser], Coroutine[Any, Any, None]]:
    async def inner(
        request: Request,
        user: GetUserDep,
    ) -> None:
        pool_name = request.path_params.get("pool_name")
        raw_team_name = Pool.get_team_name(pool_name) if pool_name else None
        if raw_team_name is None:
            with suppress(JSONDecodeError):
                raw_team_name = (await request.json()).get("team_name")
        team_name = Team.get_name_if_exists(raw_team_name) if raw_team_name else None

        _requires_access(
            is_authorized_callback=lambda: get_auth_manager().is_authorized_pool(
                method=method, details=PoolDetails(name=pool_name, team_name=team_name), user=user
            )
        )

    return inner


def requires_access_pool_bulk() -> Callable[[BulkBody[PoolBody], BaseUser], None]:
    def inner(
        request: BulkBody[PoolBody],
        user: GetUserDep,
    ) -> None:
        # Build the list of pool names provided as part of the request
        existing_pool_names = [
            cast("str", entity) if action.action == BulkAction.DELETE else cast("PoolBody", entity).pool
            for action in request.actions
            for entity in action.entities
            if action.action != BulkAction.CREATE

Review comment (suggestion):
Wouldn't it be cool if Pool would accept None parameters for pool name and handle the check internally so we don't have to do it in the code here?  This is just an observation and has nothing to do with the intent/fix of this PR but I'm just wondering as I would make the code read nicer.

Code after:
    """
    Create a callable for Depends in FastAPI that returns a filter of the permitted pools for the user.

    :param method: whether filter readable or writable.
    """

    def depends_permitted_pools_filter(
        user: GetUserDep,
        auth_manager: AuthManagerDep,
    ) -> PermittedPoolFilter:
        authorized_pools: set[str] = auth_manager.get_authorized_pools(user=user, method=method)
        return PermittedPoolFilter(authorized_pools)

    return depends_permitted_pools_filter


ReadablePoolsFilterDep = Annotated[PermittedPoolFilter, Depends(permitted_pool_filter_factory("GET"))]


def requires_access_pool(method: ResourceMethod) -> Callable[[Request, BaseUser], Coroutine[Any, Any, None]]:
    async def inner(
        request: Request,
        user: GetUserDep,
    ) -> None:
        pool_name = request.path_params.get("pool_name")
        for team_name in await _collect_teams_to_check(method, request, pool_name, Pool.get_team_name):

            def _callback(tn: str | None = team_name) -> bool:
                return get_auth_manager().is_authorized_pool(
                    method=method, details=PoolDetails(name=pool_name, team_name=tn), user=user
                )

            _requires_access(is_authorized_callback=_callback)

    return inner


def requires_access_pool_bulk() -> Callable[[BulkBody[PoolBody], BaseUser], None]:
    def inner(
        request: BulkBody[PoolBody],
        user: GetUserDep,
    ) -> None:
        # Build the list of pool names provided as part of the request
        existing_pool_names = [
            cast("str", entity) if action.action == BulkAction.DELETE else cast("PoolBody", entity).pool
            for action in request.actions
            for entity in action.entities
            if action.action != BulkAction.CREATE
        ]
        # For each pool, find its associated team (if it exists)
        pool_name_to_team = Pool.get_name_to_team_name_mapping(existing_pool_names)