scaleway_core.utils package

Submodules

scaleway_core.utils.fetch_all_pages module

scaleway_core.utils.fetch_all_pages.fetch_all_pages(type: Type[T], key: str, fetcher: Callable[[...], T], args: Dict[str, Any], page: Optional[int] = None) List[Any]
Parameters:
  • key – The key to use to get the list of items from the response

  • fetcher – The function to call to fetch the response

Returns:

The list of items

async scaleway_core.utils.fetch_all_pages.fetch_all_pages_async(type: Type[T], key: str, fetcher: Callable[[...], Awaitable[T]], args: Dict[str, Any], page: Optional[int] = None) List[Any]
Parameters:
  • key – The key to use to get the list of items from the response

  • fetcher – The function to call to fetch the response

Returns:

The list of items

scaleway_core.utils.project_or_organization_id module

scaleway_core.utils.project_or_organization_id.project_or_organization_id(organization_id: Optional[str], project_id: Optional[str], default_project_id: Optional[str]) Dict[str, str]

scaleway_core.utils.random_name module

scaleway_core.utils.random_name.random_name(prefix: str = '', separator: str = '-') str

scaleway_core.utils.resolve_one_of module

class scaleway_core.utils.resolve_one_of.OneOfPossibility(*args, **kwds)

Bases: Generic[T]

default: Optional[T] = None
param: str
value: Optional[T]
scaleway_core.utils.resolve_one_of.resolve_one_of(possibilities: List[OneOfPossibility[Any]], is_required: bool = False) Dict[str, Any]

Resolves the ideal parameter and value amongst an optional list.

scaleway_core.utils.strenummeta module

class scaleway_core.utils.strenummeta.StrEnumMeta(cls, bases, classdict)

Bases: EnumMeta

scaleway_core.utils.validate_path_param module

scaleway_core.utils.validate_path_param.validate_path_param(name: str, value: Optional[Union[str, int, datetime]]) str

Returns the parameter if it’s valid as path parameter (string and not empty, or number), else throws an exception.

scaleway_core.utils.waiter module

class scaleway_core.utils.waiter.WaitForOptions(timeout: float = 300, min_delay: float = 1, max_delay: float = 30, stop: Optional[Callable[[T], U]] = None)

Bases: Generic[T, U]

The options to wait until a resource is ready.

max_delay: float = 30

The maximum delay before the next try in seconds.

Default:

30 seconds.

min_delay: float = 1

The minimum delay before the next try in seconds.

Default:

1 second.

stop: Optional[Callable[[T], U]] = None

The condition to stop trying.

Default:

Waits for non-transient value.

timeout: float = 300

Timeout in seconds.

Default:

300 seconds (5 minutes).

scaleway_core.utils.waiter.wait_for_resource(fetcher: Callable[[...], T], options: WaitForOptions[T, bool], args: Dict[str, Any]) T

Fetches resource several times until an expected condition is reached, timeouts, or throws an exception.

async scaleway_core.utils.waiter.wait_for_resource_async(fetcher: Callable[[...], Awaitable[T]], options: WaitForOptions[T, Union[bool, Awaitable[bool]]], args: Dict[str, Any]) T

Fetches resource several times until an expected condition is reached, timeouts, or throws an exception.

Module contents

class scaleway_core.utils.OneOfPossibility(*args, **kwds)

Bases: Generic[T]

default: Optional[T] = None
param: str
value: Optional[T]
class scaleway_core.utils.StrEnumMeta(cls, bases, classdict)

Bases: EnumMeta

class scaleway_core.utils.WaitForOptions(timeout: float = 300, min_delay: float = 1, max_delay: float = 30, stop: Optional[Callable[[T], U]] = None)

Bases: Generic[T, U]

The options to wait until a resource is ready.

max_delay: float = 30

The maximum delay before the next try in seconds.

Default:

30 seconds.

min_delay: float = 1

The minimum delay before the next try in seconds.

Default:

1 second.

stop: Optional[Callable[[T], U]] = None

The condition to stop trying.

Default:

Waits for non-transient value.

timeout: float = 300

Timeout in seconds.

Default:

300 seconds (5 minutes).

scaleway_core.utils.fetch_all_pages(type: Type[T], key: str, fetcher: Callable[[...], T], args: Dict[str, Any], page: Optional[int] = None) List[Any]
Parameters:
  • key – The key to use to get the list of items from the response

  • fetcher – The function to call to fetch the response

Returns:

The list of items

async scaleway_core.utils.fetch_all_pages_async(type: Type[T], key: str, fetcher: Callable[[...], Awaitable[T]], args: Dict[str, Any], page: Optional[int] = None) List[Any]
Parameters:
  • key – The key to use to get the list of items from the response

  • fetcher – The function to call to fetch the response

Returns:

The list of items

scaleway_core.utils.project_or_organization_id(organization_id: Optional[str], project_id: Optional[str], default_project_id: Optional[str]) Dict[str, str]
scaleway_core.utils.random_name(prefix: str = '', separator: str = '-') str
scaleway_core.utils.resolve_one_of(possibilities: List[OneOfPossibility[Any]], is_required: bool = False) Dict[str, Any]

Resolves the ideal parameter and value amongst an optional list.

scaleway_core.utils.validate_path_param(name: str, value: Optional[Union[str, int, datetime]]) str

Returns the parameter if it’s valid as path parameter (string and not empty, or number), else throws an exception.

scaleway_core.utils.wait_for_resource(fetcher: Callable[[...], T], options: WaitForOptions[T, bool], args: Dict[str, Any]) T

Fetches resource several times until an expected condition is reached, timeouts, or throws an exception.

async scaleway_core.utils.wait_for_resource_async(fetcher: Callable[[...], Awaitable[T]], options: WaitForOptions[T, Union[bool, Awaitable[bool]]], args: Dict[str, Any]) T

Fetches resource several times until an expected condition is reached, timeouts, or throws an exception.