dev4py.utils.async_joptional

The AsyncJOptional class is like the JOptional one but simplify the use of Awaitable values

  1"""The `AsyncJOptional` class is like the `JOptional` one but simplify the use of Awaitable values"""
  2
  3# Copyright 2022 the original author or authors (i.e.: St4rG00se for Dev4py).
  4#
  5# Licensed under the Apache License, Version 2.0 (the "License");
  6# you may not use this file except in compliance with the License.
  7# You may obtain a copy of the License at
  8#
  9#      https://www.apache.org/licenses/LICENSE-2.0
 10#
 11# Unless required by applicable law or agreed to in writing, software
 12# distributed under the License is distributed on an "AS IS" BASIS,
 13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 14# See the License for the specific language governing permissions and
 15# limitations under the License.
 16
 17from __future__ import annotations
 18
 19from functools import partial
 20from typing import Generic, Final, Optional, TypeAlias, cast, Any, Awaitable
 21
 22from dev4py.utils.joptional import JOptional
 23from dev4py.utils.awaitables import is_awaitable
 24from dev4py.utils.objects import async_require_non_none, non_none, is_none, require_non_none, to_self, to_none
 25from dev4py.utils.types import T, V, SyncOrAsync, Supplier, Function, R, Consumer, Runnable, Predicate
 26
 27_VType: TypeAlias = SyncOrAsync[Optional[V]]  # pragma: no mutate
 28
 29
 30class AsyncJOptional(Generic[T]):
 31    """A JOptional class for async value mapping"""
 32
 33    __CREATE_KEY: Final[object] = object()
 34    __NO_VALUE_ERROR_MSG: Final[str] = "No value present"
 35    __EMPTY: AsyncJOptional[Any]
 36
 37    @classmethod
 38    # pylint: disable=E1136
 39    def of(cls, value: _VType[T]) -> AsyncJOptional[T]:
 40        """
 41        Returns a JOptional describing the given non-None value or Awaitable value
 42
 43        Args:
 44            value: The value (or Awaitable value) to describe, which must be non-None and not an Awaitable of None
 45
 46        Returns:
 47            JOptional[T]: A JOptional of value T or Awaitable[T] type with the value present
 48
 49        Raises:
 50            TypeError: Raises a TypeError on terminal operation if value is None or is an Awaitable of None
 51        """
 52        return AsyncJOptional(async_require_non_none(value, cls.__NO_VALUE_ERROR_MSG), cls.__CREATE_KEY)
 53
 54    @classmethod
 55    # pylint: disable=E1136
 56    def of_noneable(cls, value: _VType[T]) -> AsyncJOptional[T]:
 57        """
 58        Returns a JOptional describing the given value or Awaitable value, if non-None, otherwise returns an empty
 59        AsyncJOptional
 60
 61        Args:
 62            value: The possibly-None or Awaitable of None value to describe
 63
 64        Returns:
 65            JOptional[T]: A JOptional of value T  or Awaitable[T] type with the value present if the specified value is
 66            non-None, otherwise an empty AsyncJOptional
 67        """
 68        return AsyncJOptional(value, cls.__CREATE_KEY)
 69
 70    @classmethod
 71    def empty(cls) -> AsyncJOptional[T]:
 72        """Return an empty instance of AsyncJOptional"""
 73        return cls.__EMPTY
 74
 75    # pylint: disable=E1136
 76    def __init__(self, value: _VType[T], create_key: object):
 77        """AsyncJOptional private constructor: Constructs an instance with the described value"""
 78        assert create_key == self.__CREATE_KEY, \
 79            "AsyncJOptional private constructor! Please use AsyncJOptional.of or AsyncJOptional.of_noneable"
 80        self._value: _VType[T] = value
 81
 82    async def _get_value(self) -> Optional[T]:
 83        """
 84        Private method in order to manage Sync or Awaitable value
 85
 86        Returns:
 87            Optional[T]: the current value
 88        """
 89        if is_awaitable(self._value):
 90            self._value = await cast(Awaitable[Optional[T]], self._value)
 91        return cast(Optional[T], self._value)
 92
 93    async def is_present(self) -> bool:
 94        """
 95        If a value is present, returns true, otherwise false
 96
 97        Note: A value is present when not None and not an Awaitable of None
 98
 99        Returns:
100            bool: true if a value is not None and not an Awaitable of None, otherwise false
101        """
102        return non_none(await self._get_value())
103
104    async def is_empty(self) -> bool:
105        """
106        If a value is not present, returns true, otherwise false.
107
108        Note: A value is empty when None or an Awaitable of None
109
110        Returns:
111            bool: true if a value is None or an Awaitable of None, otherwise false
112        """
113        return is_none(await self._get_value())
114
115    def get(self) -> Awaitable[T]:
116        """
117        If a value is present, returns the value, otherwise raises a ValueError
118
119        Returns:
120            value: The non-None value described by this AsyncJOptional
121
122        Raises:
123            ValueError: Raises a ValueError if the value is empty
124        """
125        return self.or_else_raise()
126
127    def or_else(self, other: Optional[T] = None) -> Awaitable[Optional[T]]:
128        """
129        If a value is present, returns the value, otherwise returns other
130
131        Args:
132            other: The value to be returned, if no value is present. May be None.
133
134        Returns:
135            value: The value, if present, otherwise other
136        """
137        return self.or_else_get(cast(Supplier[Optional[T]], partial(to_self, obj=other)))
138
139    async def or_else_get(self, supplier: Supplier[Optional[T]] = to_none) -> Optional[T]:
140        """
141        If a value is present, returns the value, otherwise returns the result produced by the supplying function
142
143        Args:
144            supplier: The supplying function that produces a value to be returned
145
146        Returns:
147            value: The value, if present, otherwise the result produced by the supplying function
148
149        Raises:
150            TypeError: if the supplying function is None
151        """
152        require_non_none(supplier)
153        return cast(Optional[T], self._value if await self.is_present() else supplier())
154
155    @staticmethod  # pragma: no mutate
156    def __or_else_raise_supplier_lambda() -> Exception:
157        """
158        private static method to replace or_else_raise lambda supplier by a function
159        Note: lambda are not used in order to be compatible with multiprocessing (lambda are not serializable)
160        """
161        # lambda: cast(Exception, ValueError(AsyncJOptional.__NO_VALUE_ERROR_MSG))
162        return cast(Exception, ValueError(AsyncJOptional.__NO_VALUE_ERROR_MSG))
163
164    async def or_else_raise(
165            self,
166            supplier: Supplier[Exception] = __or_else_raise_supplier_lambda
167    ) -> T:
168        """
169        If a value is present, returns the value, otherwise raises an exception produced by the exception supplying
170        function
171
172        Args:
173            supplier: The supplying function that produces an exception to be raised (default: ValueError)
174
175        Returns:
176            value: The value, if present
177
178        Raises:
179            Exception: If value is empty
180            TypeError: if the exception supplying function is None
181        """
182        require_non_none(supplier)
183        if await self.is_empty():
184            raise supplier()
185        return cast(T, self._value)
186
187    def or_get(self, supplier: Supplier[AsyncJOptional[T]]) -> AsyncJOptional[T]:
188        """
189        This is the `or` equivalent in java Optional (Reminder: `or` is a python keyword)
190
191        If a value is present, returns an AsyncJOptional describing the value, otherwise returns an AsyncJOptional
192        produced by the supplying function
193
194        Args:
195            supplier: The supplying function that produces an AsyncJOptional to be returned
196
197        Returns:
198            AsyncJOptional[T]: Returns an AsyncJOptional describing the value of this AsyncJOptional, if a value is
199            present, otherwise an AsyncJOptional produced by the supplying function
200
201        Raises:
202            TypeError: if the supplying function is None
203        """
204        require_non_none(supplier)
205
206        async def _async_or_get(current: AsyncJOptional[T], suppl: Supplier[AsyncJOptional[T]]) -> Optional[T]:
207            return await current.get() if await current.is_present() else await require_non_none(suppl()).or_else_get()
208
209        return AsyncJOptional.of_noneable(_async_or_get(self, supplier))
210
211    @staticmethod  # pragma: no mutate
212    def __map_lambda(v: T, mapper: Function[T, _VType[R]]) -> AsyncJOptional[R]:
213        """
214        private static method to replace inner map lambda supplier by a function
215        Note: lambda are not used in order to be compatible with multiprocessing (lambda are not serializable)
216        """
217        # lambda v: AsyncJOptional.of_noneable(mapper(v))
218        return AsyncJOptional.of_noneable(mapper(v))
219
220    def map(self, mapper: Function[T, _VType[R]]) -> AsyncJOptional[R]:
221        """
222        If a value is present, returns an AsyncJOptional describing (as if by of_noneable) the result of applying the
223        given mapping function to the value, otherwise returns an empty AsyncJOptional
224
225        If the mapping function returns a None result then this method returns an empty AsyncJOptional.
226
227        Note: mapping function can be sync or Awaitable (like async coroutine)
228
229        Args:
230            mapper: The mapping function to apply to a value, if present
231
232        Returns:
233            AsyncJOptional[R]: An AsyncJOptional describing the result of applying a mapping function to the value of
234            this AsyncJOptional, if a value is present, otherwise an empty AsyncJOptional
235
236        Raises:
237            TypeError: If the mapping function is None
238        """
239        require_non_none(mapper)
240        return self.flat_map(partial(AsyncJOptional.__map_lambda, mapper=mapper))
241
242    def flat_map(self, mapper: Function[T, AsyncJOptional[R]]) -> AsyncJOptional[R]:
243        """
244        If a value is present, returns the result of applying the given AsyncJOptional-bearing mapping function to the
245        value, otherwise returns an empty AsyncJOptional
246
247        This method is similar to map(Function), but the mapping function is one whose result is already an
248        AsyncJOptional, and if invoked, flatMap does not wrap it within an additional AsyncJOptional
249
250        Args:
251            mapper: The mapping function to apply to a value, if present
252
253        Returns:
254            AsyncJOptional[R]: The result of applying a AsyncJOptional-bearing mapping function to the value of this
255            AsyncJOptional, if a value is present, otherwise an empty AsyncJOptional
256
257        Raises:
258            TypeError: if the mapping function is None
259        """
260        require_non_none(mapper)
261
262        async def _async_flat_map(
263                current: AsyncJOptional[T], flat_mapper: Function[T, AsyncJOptional[R]]
264        ) -> Optional[R]:
265            if await current.is_present():
266                new_async_joptional: AsyncJOptional[R] = require_non_none(flat_mapper(await current.get()))
267                return await new_async_joptional.or_else_get()
268            return None
269
270        return AsyncJOptional.of_noneable(_async_flat_map(self, mapper))
271
272    async def if_present(self, consumer: Consumer[T]) -> None:
273        """
274        If a value is present, performs the given consumer with the value, otherwise does nothing
275
276        Args:
277            consumer: The consumer to be performed, if a value is present
278
279        Returns:
280            Nothing
281
282        Raises:
283            TypeError: if the consumer is None
284        """
285        require_non_none(consumer)
286        # pylint: disable=W0106
287        (await self.is_present()) and consumer(cast(T, self._value))
288
289    async def if_empty(self, empty_action: Runnable) -> None:
290        """
291        If a value is not present, performs the given runnable, otherwise does nothing
292
293        Args:
294            empty_action: The runnable to be performed, if a value is not present
295
296        Returns:
297            Nothing
298
299        Raises:
300            TypeError: if the runnable is None
301        """
302        require_non_none(empty_action)
303        # pylint: disable=W0106
304        (await self.is_empty()) and empty_action()
305
306    async def if_present_or_else(self, consumer: Consumer[T], empty_action: Runnable) -> None:
307        """
308        If a value is present, performs the given consumer with the value, otherwise performs the given empty_action
309
310        Args:
311            consumer: The consumer to be performed, if a value is present
312            empty_action: The runnable to be performed, if a value is not
313
314        Returns:
315            Nothing
316
317        Raises:
318            TypeError: if the consumer is None or if the runnable is None
319        """
320        require_non_none(consumer)
321        require_non_none(empty_action)
322        # pylint: disable=W0106
323        consumer(cast(T, self._value)) if await self.is_present() else empty_action()
324
325    def filter(self, predicate: Predicate[T]) -> AsyncJOptional[T]:
326        """
327        If a value is present, and the value matches the given predicate, returns an AsyncJOptional describing the
328        value, otherwise returns an empty AsyncJOptional
329
330        Args:
331            predicate: The predicate to apply to a value, if present
332
333        Returns:
334            AsyncJOptional[T]: An AsyncJOptional describing the value of this AsyncJOptional, if a value is present and
335            the value matches the given predicate, otherwise an empty AsyncJOptional
336
337        Raises:
338            TypeError: if the predicate is None
339        """
340        require_non_none(predicate)
341
342        async def _async_filter(current: AsyncJOptional[T]) -> Optional[T]:
343            return await current.or_else_get() if await current.is_empty() or predicate(await current.get()) else None
344
345        return AsyncJOptional.of_noneable(_async_filter(self))
346
347    def peek(self, consumer: Consumer[T]) -> AsyncJOptional[T]:
348        """
349        If a value is present, performs the given consumer with the value, otherwise does nothing and returns the
350        current equivalent of AsyncJOptional
351
352        Args:
353            consumer: The consumer to be performed, if a value is present
354
355        Returns:
356             AsyncJOptional[T]: The current equivalent of AsyncJOptional
357
358        Raises:
359            TypeError: if the consumer is None
360        """
361        require_non_none(consumer)
362
363        async def _async_peek(current: AsyncJOptional[T], cons: Consumer[T]) -> Optional[T]:
364            await current.if_present(cons)
365            return await current.or_else_get()
366
367        return AsyncJOptional.of_noneable(_async_peek(self, consumer))
368
369    def to_joptional(self) -> JOptional[Awaitable[Optional[T]]]:
370        """
371        Convert the current AsyncJOptional to a JOptional
372
373        Note: the returned JOptional will contain an Awaitable[Optional[T]] value
374
375        Returns:
376            JOptional[Awaitable[Optional[T]]]: The corresponding JOptional
377        """
378        return JOptional.of_noneable(cast(Awaitable[Optional[T]], self._get_value()))
379
380
381# INIT STATIC VARIABLES
382# noinspection PyProtectedMember
383# noinspection PyUnresolvedReferences
384# pylint: disable=W0212
385AsyncJOptional._AsyncJOptional__EMPTY = AsyncJOptional(None, AsyncJOptional._AsyncJOptional__CREATE_KEY)  # type: ignore
class AsyncJOptional(typing.Generic[~T]): 
 31class AsyncJOptional(Generic[T]):
 32    """A JOptional class for async value mapping"""
 33
 34    __CREATE_KEY: Final[object] = object()
 35    __NO_VALUE_ERROR_MSG: Final[str] = "No value present"
 36    __EMPTY: AsyncJOptional[Any]
 37
 38    @classmethod
 39    # pylint: disable=E1136
 40    def of(cls, value: _VType[T]) -> AsyncJOptional[T]:
 41        """
 42        Returns a JOptional describing the given non-None value or Awaitable value
 43
 44        Args:
 45            value: The value (or Awaitable value) to describe, which must be non-None and not an Awaitable of None
 46
 47        Returns:
 48            JOptional[T]: A JOptional of value T or Awaitable[T] type with the value present
 49
 50        Raises:
 51            TypeError: Raises a TypeError on terminal operation if value is None or is an Awaitable of None
 52        """
 53        return AsyncJOptional(async_require_non_none(value, cls.__NO_VALUE_ERROR_MSG), cls.__CREATE_KEY)
 54
 55    @classmethod
 56    # pylint: disable=E1136
 57    def of_noneable(cls, value: _VType[T]) -> AsyncJOptional[T]:
 58        """
 59        Returns a JOptional describing the given value or Awaitable value, if non-None, otherwise returns an empty
 60        AsyncJOptional
 61
 62        Args:
 63            value: The possibly-None or Awaitable of None value to describe
 64
 65        Returns:
 66            JOptional[T]: A JOptional of value T  or Awaitable[T] type with the value present if the specified value is
 67            non-None, otherwise an empty AsyncJOptional
 68        """
 69        return AsyncJOptional(value, cls.__CREATE_KEY)
 70
 71    @classmethod
 72    def empty(cls) -> AsyncJOptional[T]:
 73        """Return an empty instance of AsyncJOptional"""
 74        return cls.__EMPTY
 75
 76    # pylint: disable=E1136
 77    def __init__(self, value: _VType[T], create_key: object):
 78        """AsyncJOptional private constructor: Constructs an instance with the described value"""
 79        assert create_key == self.__CREATE_KEY, \
 80            "AsyncJOptional private constructor! Please use AsyncJOptional.of or AsyncJOptional.of_noneable"
 81        self._value: _VType[T] = value
 82
 83    async def _get_value(self) -> Optional[T]:
 84        """
 85        Private method in order to manage Sync or Awaitable value
 86
 87        Returns:
 88            Optional[T]: the current value
 89        """
 90        if is_awaitable(self._value):
 91            self._value = await cast(Awaitable[Optional[T]], self._value)
 92        return cast(Optional[T], self._value)
 93
 94    async def is_present(self) -> bool:
 95        """
 96        If a value is present, returns true, otherwise false
 97
 98        Note: A value is present when not None and not an Awaitable of None
 99
100        Returns:
101            bool: true if a value is not None and not an Awaitable of None, otherwise false
102        """
103        return non_none(await self._get_value())
104
105    async def is_empty(self) -> bool:
106        """
107        If a value is not present, returns true, otherwise false.
108
109        Note: A value is empty when None or an Awaitable of None
110
111        Returns:
112            bool: true if a value is None or an Awaitable of None, otherwise false
113        """
114        return is_none(await self._get_value())
115
116    def get(self) -> Awaitable[T]:
117        """
118        If a value is present, returns the value, otherwise raises a ValueError
119
120        Returns:
121            value: The non-None value described by this AsyncJOptional
122
123        Raises:
124            ValueError: Raises a ValueError if the value is empty
125        """
126        return self.or_else_raise()
127
128    def or_else(self, other: Optional[T] = None) -> Awaitable[Optional[T]]:
129        """
130        If a value is present, returns the value, otherwise returns other
131
132        Args:
133            other: The value to be returned, if no value is present. May be None.
134
135        Returns:
136            value: The value, if present, otherwise other
137        """
138        return self.or_else_get(cast(Supplier[Optional[T]], partial(to_self, obj=other)))
139
140    async def or_else_get(self, supplier: Supplier[Optional[T]] = to_none) -> Optional[T]:
141        """
142        If a value is present, returns the value, otherwise returns the result produced by the supplying function
143
144        Args:
145            supplier: The supplying function that produces a value to be returned
146
147        Returns:
148            value: The value, if present, otherwise the result produced by the supplying function
149
150        Raises:
151            TypeError: if the supplying function is None
152        """
153        require_non_none(supplier)
154        return cast(Optional[T], self._value if await self.is_present() else supplier())
155
156    @staticmethod  # pragma: no mutate
157    def __or_else_raise_supplier_lambda() -> Exception:
158        """
159        private static method to replace or_else_raise lambda supplier by a function
160        Note: lambda are not used in order to be compatible with multiprocessing (lambda are not serializable)
161        """
162        # lambda: cast(Exception, ValueError(AsyncJOptional.__NO_VALUE_ERROR_MSG))
163        return cast(Exception, ValueError(AsyncJOptional.__NO_VALUE_ERROR_MSG))
164
165    async def or_else_raise(
166            self,
167            supplier: Supplier[Exception] = __or_else_raise_supplier_lambda
168    ) -> T:
169        """
170        If a value is present, returns the value, otherwise raises an exception produced by the exception supplying
171        function
172
173        Args:
174            supplier: The supplying function that produces an exception to be raised (default: ValueError)
175
176        Returns:
177            value: The value, if present
178
179        Raises:
180            Exception: If value is empty
181            TypeError: if the exception supplying function is None
182        """
183        require_non_none(supplier)
184        if await self.is_empty():
185            raise supplier()
186        return cast(T, self._value)
187
188    def or_get(self, supplier: Supplier[AsyncJOptional[T]]) -> AsyncJOptional[T]:
189        """
190        This is the `or` equivalent in java Optional (Reminder: `or` is a python keyword)
191
192        If a value is present, returns an AsyncJOptional describing the value, otherwise returns an AsyncJOptional
193        produced by the supplying function
194
195        Args:
196            supplier: The supplying function that produces an AsyncJOptional to be returned
197
198        Returns:
199            AsyncJOptional[T]: Returns an AsyncJOptional describing the value of this AsyncJOptional, if a value is
200            present, otherwise an AsyncJOptional produced by the supplying function
201
202        Raises:
203            TypeError: if the supplying function is None
204        """
205        require_non_none(supplier)
206
207        async def _async_or_get(current: AsyncJOptional[T], suppl: Supplier[AsyncJOptional[T]]) -> Optional[T]:
208            return await current.get() if await current.is_present() else await require_non_none(suppl()).or_else_get()
209
210        return AsyncJOptional.of_noneable(_async_or_get(self, supplier))
211
212    @staticmethod  # pragma: no mutate
213    def __map_lambda(v: T, mapper: Function[T, _VType[R]]) -> AsyncJOptional[R]:
214        """
215        private static method to replace inner map lambda supplier by a function
216        Note: lambda are not used in order to be compatible with multiprocessing (lambda are not serializable)
217        """
218        # lambda v: AsyncJOptional.of_noneable(mapper(v))
219        return AsyncJOptional.of_noneable(mapper(v))
220
221    def map(self, mapper: Function[T, _VType[R]]) -> AsyncJOptional[R]:
222        """
223        If a value is present, returns an AsyncJOptional describing (as if by of_noneable) the result of applying the
224        given mapping function to the value, otherwise returns an empty AsyncJOptional
225
226        If the mapping function returns a None result then this method returns an empty AsyncJOptional.
227
228        Note: mapping function can be sync or Awaitable (like async coroutine)
229
230        Args:
231            mapper: The mapping function to apply to a value, if present
232
233        Returns:
234            AsyncJOptional[R]: An AsyncJOptional describing the result of applying a mapping function to the value of
235            this AsyncJOptional, if a value is present, otherwise an empty AsyncJOptional
236
237        Raises:
238            TypeError: If the mapping function is None
239        """
240        require_non_none(mapper)
241        return self.flat_map(partial(AsyncJOptional.__map_lambda, mapper=mapper))
242
243    def flat_map(self, mapper: Function[T, AsyncJOptional[R]]) -> AsyncJOptional[R]:
244        """
245        If a value is present, returns the result of applying the given AsyncJOptional-bearing mapping function to the
246        value, otherwise returns an empty AsyncJOptional
247
248        This method is similar to map(Function), but the mapping function is one whose result is already an
249        AsyncJOptional, and if invoked, flatMap does not wrap it within an additional AsyncJOptional
250
251        Args:
252            mapper: The mapping function to apply to a value, if present
253
254        Returns:
255            AsyncJOptional[R]: The result of applying a AsyncJOptional-bearing mapping function to the value of this
256            AsyncJOptional, if a value is present, otherwise an empty AsyncJOptional
257
258        Raises:
259            TypeError: if the mapping function is None
260        """
261        require_non_none(mapper)
262
263        async def _async_flat_map(
264                current: AsyncJOptional[T], flat_mapper: Function[T, AsyncJOptional[R]]
265        ) -> Optional[R]:
266            if await current.is_present():
267                new_async_joptional: AsyncJOptional[R] = require_non_none(flat_mapper(await current.get()))
268                return await new_async_joptional.or_else_get()
269            return None
270
271        return AsyncJOptional.of_noneable(_async_flat_map(self, mapper))
272
273    async def if_present(self, consumer: Consumer[T]) -> None:
274        """
275        If a value is present, performs the given consumer with the value, otherwise does nothing
276
277        Args:
278            consumer: The consumer to be performed, if a value is present
279
280        Returns:
281            Nothing
282
283        Raises:
284            TypeError: if the consumer is None
285        """
286        require_non_none(consumer)
287        # pylint: disable=W0106
288        (await self.is_present()) and consumer(cast(T, self._value))
289
290    async def if_empty(self, empty_action: Runnable) -> None:
291        """
292        If a value is not present, performs the given runnable, otherwise does nothing
293
294        Args:
295            empty_action: The runnable to be performed, if a value is not present
296
297        Returns:
298            Nothing
299
300        Raises:
301            TypeError: if the runnable is None
302        """
303        require_non_none(empty_action)
304        # pylint: disable=W0106
305        (await self.is_empty()) and empty_action()
306
307    async def if_present_or_else(self, consumer: Consumer[T], empty_action: Runnable) -> None:
308        """
309        If a value is present, performs the given consumer with the value, otherwise performs the given empty_action
310
311        Args:
312            consumer: The consumer to be performed, if a value is present
313            empty_action: The runnable to be performed, if a value is not
314
315        Returns:
316            Nothing
317
318        Raises:
319            TypeError: if the consumer is None or if the runnable is None
320        """
321        require_non_none(consumer)
322        require_non_none(empty_action)
323        # pylint: disable=W0106
324        consumer(cast(T, self._value)) if await self.is_present() else empty_action()
325
326    def filter(self, predicate: Predicate[T]) -> AsyncJOptional[T]:
327        """
328        If a value is present, and the value matches the given predicate, returns an AsyncJOptional describing the
329        value, otherwise returns an empty AsyncJOptional
330
331        Args:
332            predicate: The predicate to apply to a value, if present
333
334        Returns:
335            AsyncJOptional[T]: An AsyncJOptional describing the value of this AsyncJOptional, if a value is present and
336            the value matches the given predicate, otherwise an empty AsyncJOptional
337
338        Raises:
339            TypeError: if the predicate is None
340        """
341        require_non_none(predicate)
342
343        async def _async_filter(current: AsyncJOptional[T]) -> Optional[T]:
344            return await current.or_else_get() if await current.is_empty() or predicate(await current.get()) else None
345
346        return AsyncJOptional.of_noneable(_async_filter(self))
347
348    def peek(self, consumer: Consumer[T]) -> AsyncJOptional[T]:
349        """
350        If a value is present, performs the given consumer with the value, otherwise does nothing and returns the
351        current equivalent of AsyncJOptional
352
353        Args:
354            consumer: The consumer to be performed, if a value is present
355
356        Returns:
357             AsyncJOptional[T]: The current equivalent of AsyncJOptional
358
359        Raises:
360            TypeError: if the consumer is None
361        """
362        require_non_none(consumer)
363
364        async def _async_peek(current: AsyncJOptional[T], cons: Consumer[T]) -> Optional[T]:
365            await current.if_present(cons)
366            return await current.or_else_get()
367
368        return AsyncJOptional.of_noneable(_async_peek(self, consumer))
369
370    def to_joptional(self) -> JOptional[Awaitable[Optional[T]]]:
371        """
372        Convert the current AsyncJOptional to a JOptional
373
374        Note: the returned JOptional will contain an Awaitable[Optional[T]] value
375
376        Returns:
377            JOptional[Awaitable[Optional[T]]]: The corresponding JOptional
378        """
379        return JOptional.of_noneable(cast(Awaitable[Optional[T]], self._get_value()))

A JOptional class for async value mapping

AsyncJOptional( value: Union[Awaitable[Optional[~T]], ~T, NoneType], create_key: object) 
77    def __init__(self, value: _VType[T], create_key: object):
78        """AsyncJOptional private constructor: Constructs an instance with the described value"""
79        assert create_key == self.__CREATE_KEY, \
80            "AsyncJOptional private constructor! Please use AsyncJOptional.of or AsyncJOptional.of_noneable"
81        self._value: _VType[T] = value

AsyncJOptional private constructor: Constructs an instance with the described value

@classmethod
def of( cls, value: Union[Awaitable[Optional[~T]], ~T, NoneType]) -> AsyncJOptional[~T]: 
38    @classmethod
39    # pylint: disable=E1136
40    def of(cls, value: _VType[T]) -> AsyncJOptional[T]:
41        """
42        Returns a JOptional describing the given non-None value or Awaitable value
43
44        Args:
45            value: The value (or Awaitable value) to describe, which must be non-None and not an Awaitable of None
46
47        Returns:
48            JOptional[T]: A JOptional of value T or Awaitable[T] type with the value present
49
50        Raises:
51            TypeError: Raises a TypeError on terminal operation if value is None or is an Awaitable of None
52        """
53        return AsyncJOptional(async_require_non_none(value, cls.__NO_VALUE_ERROR_MSG), cls.__CREATE_KEY)

Returns a JOptional describing the given non-None value or Awaitable value

Arguments:
  • value: The value (or Awaitable value) to describe, which must be non-None and not an Awaitable of None
Returns:

JOptional[T]: A JOptional of value T or Awaitable[T] type with the value present

Raises:
  • TypeError: Raises a TypeError on terminal operation if value is None or is an Awaitable of None
@classmethod
def of_noneable( cls, value: Union[Awaitable[Optional[~T]], ~T, NoneType]) -> AsyncJOptional[~T]: 
55    @classmethod
56    # pylint: disable=E1136
57    def of_noneable(cls, value: _VType[T]) -> AsyncJOptional[T]:
58        """
59        Returns a JOptional describing the given value or Awaitable value, if non-None, otherwise returns an empty
60        AsyncJOptional
61
62        Args:
63            value: The possibly-None or Awaitable of None value to describe
64
65        Returns:
66            JOptional[T]: A JOptional of value T  or Awaitable[T] type with the value present if the specified value is
67            non-None, otherwise an empty AsyncJOptional
68        """
69        return AsyncJOptional(value, cls.__CREATE_KEY)

Returns a JOptional describing the given value or Awaitable value, if non-None, otherwise returns an empty AsyncJOptional

Arguments:
  • value: The possibly-None or Awaitable of None value to describe
Returns:

JOptional[T]: A JOptional of value T or Awaitable[T] type with the value present if the specified value is non-None, otherwise an empty AsyncJOptional

@classmethod
def empty(cls) -> AsyncJOptional[~T]: 
71    @classmethod
72    def empty(cls) -> AsyncJOptional[T]:
73        """Return an empty instance of AsyncJOptional"""
74        return cls.__EMPTY

Return an empty instance of AsyncJOptional

async def is_present(self) -> bool: 
 94    async def is_present(self) -> bool:
 95        """
 96        If a value is present, returns true, otherwise false
 97
 98        Note: A value is present when not None and not an Awaitable of None
 99
100        Returns:
101            bool: true if a value is not None and not an Awaitable of None, otherwise false
102        """
103        return non_none(await self._get_value())

If a value is present, returns true, otherwise false

Note: A value is present when not None and not an Awaitable of None

Returns:

bool: true if a value is not None and not an Awaitable of None, otherwise false

async def is_empty(self) -> bool: 
105    async def is_empty(self) -> bool:
106        """
107        If a value is not present, returns true, otherwise false.
108
109        Note: A value is empty when None or an Awaitable of None
110
111        Returns:
112            bool: true if a value is None or an Awaitable of None, otherwise false
113        """
114        return is_none(await self._get_value())

If a value is not present, returns true, otherwise false.

Note: A value is empty when None or an Awaitable of None

Returns:

bool: true if a value is None or an Awaitable of None, otherwise false

def get(self) -> Awaitable[~T]: 
116    def get(self) -> Awaitable[T]:
117        """
118        If a value is present, returns the value, otherwise raises a ValueError
119
120        Returns:
121            value: The non-None value described by this AsyncJOptional
122
123        Raises:
124            ValueError: Raises a ValueError if the value is empty
125        """
126        return self.or_else_raise()

If a value is present, returns the value, otherwise raises a ValueError

Returns:

value: The non-None value described by this AsyncJOptional

Raises:
  • ValueError: Raises a ValueError if the value is empty
def or_else(self, other: Optional[~T] = None) -> Awaitable[Optional[~T]]: 
128    def or_else(self, other: Optional[T] = None) -> Awaitable[Optional[T]]:
129        """
130        If a value is present, returns the value, otherwise returns other
131
132        Args:
133            other: The value to be returned, if no value is present. May be None.
134
135        Returns:
136            value: The value, if present, otherwise other
137        """
138        return self.or_else_get(cast(Supplier[Optional[T]], partial(to_self, obj=other)))

If a value is present, returns the value, otherwise returns other

Arguments:
  • other: The value to be returned, if no value is present. May be None.
Returns:

value: The value, if present, otherwise other

async def or_else_get( self, supplier: Callable[[], Optional[~T]] = <function to_none>) -> Optional[~T]: 
140    async def or_else_get(self, supplier: Supplier[Optional[T]] = to_none) -> Optional[T]:
141        """
142        If a value is present, returns the value, otherwise returns the result produced by the supplying function
143
144        Args:
145            supplier: The supplying function that produces a value to be returned
146
147        Returns:
148            value: The value, if present, otherwise the result produced by the supplying function
149
150        Raises:
151            TypeError: if the supplying function is None
152        """
153        require_non_none(supplier)
154        return cast(Optional[T], self._value if await self.is_present() else supplier())

If a value is present, returns the value, otherwise returns the result produced by the supplying function

Arguments:
  • supplier: The supplying function that produces a value to be returned
Returns:

value: The value, if present, otherwise the result produced by the supplying function

Raises:
  • TypeError: if the supplying function is None
async def or_else_raise( self, supplier: Callable[[], Exception] = <staticmethod(<function AsyncJOptional.__or_else_raise_supplier_lambda>)>) -> ~T: 
165    async def or_else_raise(
166            self,
167            supplier: Supplier[Exception] = __or_else_raise_supplier_lambda
168    ) -> T:
169        """
170        If a value is present, returns the value, otherwise raises an exception produced by the exception supplying
171        function
172
173        Args:
174            supplier: The supplying function that produces an exception to be raised (default: ValueError)
175
176        Returns:
177            value: The value, if present
178
179        Raises:
180            Exception: If value is empty
181            TypeError: if the exception supplying function is None
182        """
183        require_non_none(supplier)
184        if await self.is_empty():
185            raise supplier()
186        return cast(T, self._value)

If a value is present, returns the value, otherwise raises an exception produced by the exception supplying function

Arguments:
  • supplier: The supplying function that produces an exception to be raised (default: ValueError)
Returns:

value: The value, if present

Raises:
  • Exception: If value is empty
  • TypeError: if the exception supplying function is None
def or_get( self, supplier: Callable[[], AsyncJOptional[~T]]) -> AsyncJOptional[~T]: 
188    def or_get(self, supplier: Supplier[AsyncJOptional[T]]) -> AsyncJOptional[T]:
189        """
190        This is the `or` equivalent in java Optional (Reminder: `or` is a python keyword)
191
192        If a value is present, returns an AsyncJOptional describing the value, otherwise returns an AsyncJOptional
193        produced by the supplying function
194
195        Args:
196            supplier: The supplying function that produces an AsyncJOptional to be returned
197
198        Returns:
199            AsyncJOptional[T]: Returns an AsyncJOptional describing the value of this AsyncJOptional, if a value is
200            present, otherwise an AsyncJOptional produced by the supplying function
201
202        Raises:
203            TypeError: if the supplying function is None
204        """
205        require_non_none(supplier)
206
207        async def _async_or_get(current: AsyncJOptional[T], suppl: Supplier[AsyncJOptional[T]]) -> Optional[T]:
208            return await current.get() if await current.is_present() else await require_non_none(suppl()).or_else_get()
209
210        return AsyncJOptional.of_noneable(_async_or_get(self, supplier))

This is the or equivalent in java Optional (Reminder: or is a python keyword)

If a value is present, returns an AsyncJOptional describing the value, otherwise returns an AsyncJOptional produced by the supplying function

Arguments:
  • supplier: The supplying function that produces an AsyncJOptional to be returned
Returns:

AsyncJOptional[T]: Returns an AsyncJOptional describing the value of this AsyncJOptional, if a value is present, otherwise an AsyncJOptional produced by the supplying function

Raises:
  • TypeError: if the supplying function is None
def map( self, mapper: Callable[[~T], Union[Awaitable[Optional[~R]], ~R, NoneType]]) -> AsyncJOptional[~R]: 
221    def map(self, mapper: Function[T, _VType[R]]) -> AsyncJOptional[R]:
222        """
223        If a value is present, returns an AsyncJOptional describing (as if by of_noneable) the result of applying the
224        given mapping function to the value, otherwise returns an empty AsyncJOptional
225
226        If the mapping function returns a None result then this method returns an empty AsyncJOptional.
227
228        Note: mapping function can be sync or Awaitable (like async coroutine)
229
230        Args:
231            mapper: The mapping function to apply to a value, if present
232
233        Returns:
234            AsyncJOptional[R]: An AsyncJOptional describing the result of applying a mapping function to the value of
235            this AsyncJOptional, if a value is present, otherwise an empty AsyncJOptional
236
237        Raises:
238            TypeError: If the mapping function is None
239        """
240        require_non_none(mapper)
241        return self.flat_map(partial(AsyncJOptional.__map_lambda, mapper=mapper))

If a value is present, returns an AsyncJOptional describing (as if by of_noneable) the result of applying the given mapping function to the value, otherwise returns an empty AsyncJOptional

If the mapping function returns a None result then this method returns an empty AsyncJOptional.

Note: mapping function can be sync or Awaitable (like async coroutine)

Arguments:
  • mapper: The mapping function to apply to a value, if present
Returns:

AsyncJOptional[R]: An AsyncJOptional describing the result of applying a mapping function to the value of this AsyncJOptional, if a value is present, otherwise an empty AsyncJOptional

Raises:
  • TypeError: If the mapping function is None
def flat_map( self, mapper: Callable[[~T], AsyncJOptional[~R]]) -> AsyncJOptional[~R]: 
243    def flat_map(self, mapper: Function[T, AsyncJOptional[R]]) -> AsyncJOptional[R]:
244        """
245        If a value is present, returns the result of applying the given AsyncJOptional-bearing mapping function to the
246        value, otherwise returns an empty AsyncJOptional
247
248        This method is similar to map(Function), but the mapping function is one whose result is already an
249        AsyncJOptional, and if invoked, flatMap does not wrap it within an additional AsyncJOptional
250
251        Args:
252            mapper: The mapping function to apply to a value, if present
253
254        Returns:
255            AsyncJOptional[R]: The result of applying a AsyncJOptional-bearing mapping function to the value of this
256            AsyncJOptional, if a value is present, otherwise an empty AsyncJOptional
257
258        Raises:
259            TypeError: if the mapping function is None
260        """
261        require_non_none(mapper)
262
263        async def _async_flat_map(
264                current: AsyncJOptional[T], flat_mapper: Function[T, AsyncJOptional[R]]
265        ) -> Optional[R]:
266            if await current.is_present():
267                new_async_joptional: AsyncJOptional[R] = require_non_none(flat_mapper(await current.get()))
268                return await new_async_joptional.or_else_get()
269            return None
270
271        return AsyncJOptional.of_noneable(_async_flat_map(self, mapper))

If a value is present, returns the result of applying the given AsyncJOptional-bearing mapping function to the value, otherwise returns an empty AsyncJOptional

This method is similar to map(Function), but the mapping function is one whose result is already an AsyncJOptional, and if invoked, flatMap does not wrap it within an additional AsyncJOptional

Arguments:
  • mapper: The mapping function to apply to a value, if present
Returns:

AsyncJOptional[R]: The result of applying a AsyncJOptional-bearing mapping function to the value of this AsyncJOptional, if a value is present, otherwise an empty AsyncJOptional

Raises:
  • TypeError: if the mapping function is None
async def if_present(self, consumer: Callable[[~T], NoneType]) -> None: 
273    async def if_present(self, consumer: Consumer[T]) -> None:
274        """
275        If a value is present, performs the given consumer with the value, otherwise does nothing
276
277        Args:
278            consumer: The consumer to be performed, if a value is present
279
280        Returns:
281            Nothing
282
283        Raises:
284            TypeError: if the consumer is None
285        """
286        require_non_none(consumer)
287        # pylint: disable=W0106
288        (await self.is_present()) and consumer(cast(T, self._value))

If a value is present, performs the given consumer with the value, otherwise does nothing

Arguments:
  • consumer: The consumer to be performed, if a value is present
Returns:

Nothing

Raises:
  • TypeError: if the consumer is None
async def if_empty(self, empty_action: Callable[[], NoneType]) -> None: 
290    async def if_empty(self, empty_action: Runnable) -> None:
291        """
292        If a value is not present, performs the given runnable, otherwise does nothing
293
294        Args:
295            empty_action: The runnable to be performed, if a value is not present
296
297        Returns:
298            Nothing
299
300        Raises:
301            TypeError: if the runnable is None
302        """
303        require_non_none(empty_action)
304        # pylint: disable=W0106
305        (await self.is_empty()) and empty_action()

If a value is not present, performs the given runnable, otherwise does nothing

Arguments:
  • empty_action: The runnable to be performed, if a value is not present
Returns:

Nothing

Raises:
  • TypeError: if the runnable is None
async def if_present_or_else( self, consumer: Callable[[~T], NoneType], empty_action: Callable[[], NoneType]) -> None: 
307    async def if_present_or_else(self, consumer: Consumer[T], empty_action: Runnable) -> None:
308        """
309        If a value is present, performs the given consumer with the value, otherwise performs the given empty_action
310
311        Args:
312            consumer: The consumer to be performed, if a value is present
313            empty_action: The runnable to be performed, if a value is not
314
315        Returns:
316            Nothing
317
318        Raises:
319            TypeError: if the consumer is None or if the runnable is None
320        """
321        require_non_none(consumer)
322        require_non_none(empty_action)
323        # pylint: disable=W0106
324        consumer(cast(T, self._value)) if await self.is_present() else empty_action()

If a value is present, performs the given consumer with the value, otherwise performs the given empty_action

Arguments:
  • consumer: The consumer to be performed, if a value is present
  • empty_action: The runnable to be performed, if a value is not
Returns:

Nothing

Raises:
  • TypeError: if the consumer is None or if the runnable is None
def filter( self, predicate: Callable[[~T], bool]) -> AsyncJOptional[~T]: 
326    def filter(self, predicate: Predicate[T]) -> AsyncJOptional[T]:
327        """
328        If a value is present, and the value matches the given predicate, returns an AsyncJOptional describing the
329        value, otherwise returns an empty AsyncJOptional
330
331        Args:
332            predicate: The predicate to apply to a value, if present
333
334        Returns:
335            AsyncJOptional[T]: An AsyncJOptional describing the value of this AsyncJOptional, if a value is present and
336            the value matches the given predicate, otherwise an empty AsyncJOptional
337
338        Raises:
339            TypeError: if the predicate is None
340        """
341        require_non_none(predicate)
342
343        async def _async_filter(current: AsyncJOptional[T]) -> Optional[T]:
344            return await current.or_else_get() if await current.is_empty() or predicate(await current.get()) else None
345
346        return AsyncJOptional.of_noneable(_async_filter(self))

If a value is present, and the value matches the given predicate, returns an AsyncJOptional describing the value, otherwise returns an empty AsyncJOptional

Arguments:
  • predicate: The predicate to apply to a value, if present
Returns:

AsyncJOptional[T]: An AsyncJOptional describing the value of this AsyncJOptional, if a value is present and the value matches the given predicate, otherwise an empty AsyncJOptional

Raises:
  • TypeError: if the predicate is None
def peek( self, consumer: Callable[[~T], NoneType]) -> AsyncJOptional[~T]: 
348    def peek(self, consumer: Consumer[T]) -> AsyncJOptional[T]:
349        """
350        If a value is present, performs the given consumer with the value, otherwise does nothing and returns the
351        current equivalent of AsyncJOptional
352
353        Args:
354            consumer: The consumer to be performed, if a value is present
355
356        Returns:
357             AsyncJOptional[T]: The current equivalent of AsyncJOptional
358
359        Raises:
360            TypeError: if the consumer is None
361        """
362        require_non_none(consumer)
363
364        async def _async_peek(current: AsyncJOptional[T], cons: Consumer[T]) -> Optional[T]:
365            await current.if_present(cons)
366            return await current.or_else_get()
367
368        return AsyncJOptional.of_noneable(_async_peek(self, consumer))

If a value is present, performs the given consumer with the value, otherwise does nothing and returns the current equivalent of AsyncJOptional

Arguments:
  • consumer: The consumer to be performed, if a value is present
Returns:

AsyncJOptional[T]: The current equivalent of AsyncJOptional

Raises:
  • TypeError: if the consumer is None
def to_joptional( self) -> dev4py.utils.joptional.JOptional[typing.Awaitable[typing.Optional[~T]]]: 
370    def to_joptional(self) -> JOptional[Awaitable[Optional[T]]]:
371        """
372        Convert the current AsyncJOptional to a JOptional
373
374        Note: the returned JOptional will contain an Awaitable[Optional[T]] value
375
376        Returns:
377            JOptional[Awaitable[Optional[T]]]: The corresponding JOptional
378        """
379        return JOptional.of_noneable(cast(Awaitable[Optional[T]], self._get_value()))

Convert the current AsyncJOptional to a JOptional

Note: the returned JOptional will contain an Awaitable[Optional[T]] value

Returns:

JOptional[Awaitable[Optional[T]]]: The corresponding JOptional