gws.core.util

General utilities, available as gws.u.

Source code: gws.core.util

Module Contents

gws.core.util.action_url_path(name: str, **kwargs) → str

Build a server URL path for an action.

Example:

action_url_path('owsService', serviceUid='wms', projectUid='')  # '/_/owsService/serviceUid/wms'
Parameters:

  • name – Action command name.

  • **kwargs – Parameters, appended as /key/value path segments. Empty values are skipped.

Returns:

The URL path.

gws.core.util.app_lock(name='')

Return a reentrant lock, local to this process.

Locks with the same name are shared.

Parameters:

name – Lock name.

Returns:

A threading.RLock.

class gws.core.util.cached_property(fn)

Decorator for a cached property.

The value is computed on first access and stored as an instance attribute with the same name.

gws.core.util.chown_default(path, user=None, group=None)

Change the owner of a path, ignoring errors.

Parameters:

  • path – A path.

  • user – Owner, defaults to const.UID.

  • group – Group, defaults to const.GID.

gws.core.util.collect(pairs)

Group values by keys.

Parameters:

pairs – An iterable of (key, value) pairs. Pairs with a None key are skipped.

Returns:

A dict mapping each key to the list of its values.

gws.core.util.compact(x)

Remove all None values from a collection.

Parameters:

x – A dict, Data or an iterable.

Returns:

A new dict or Data object, or a list for other iterables.

gws.core.util.delete_app_global(name)

Delete a process-wide global value.

Parameters:

name – Value name.

gws.core.util.dirname(path)

Return the directory part of a path.

Parameters:

path – A path.

Returns:

The directory name.

gws.core.util.ensure_dir(dir_path: str, base_dir: str = None, mode: int = 493, user: int = None, group: int = None) → str

Check if a (possibly nested) directory exists and create it if it does not.

Parameters:

  • dir_path – Path to a directory. Must be absolute without base_dir and relative with it.

  • base_dir – Base directory.

  • mode – Directory creation mode.

  • user – Directory owner, defaults to const.UID.

  • group – Directory group, defaults to const.GID.

Returns:

The path to the directory.

Raises:

ValueError – If the path is absolute with base_dir, or relative without it.

gws.core.util.ensure_system_dirs()

Create all system directories listed in const.ALL_DIRS.

gws.core.util.ephemeral_cleanup(force=False)

Remove ephemeral files and empty directories older than two hours.

Throttled to once per two hours in each process, unless force is set.

Parameters:

force – Run even if the last run was recent.

Returns:

The detached find process, or None if throttled.

gws.core.util.ephemeral_dir(name: str) → str

Create a directory in the ephemeral directory, if it does not exist yet.

Parameters:

name – Directory name.

Returns:

The directory path.

gws.core.util.ephemeral_path(name: str) → str

Return a new unique path in the ephemeral directory.

The path is not created. Ephemeral paths are removed by ephemeral_cleanup after two hours.

Parameters:

name – Base name, appended to a unique prefix.

Returns:

The path.

gws.core.util.exit(code: int = 255)

Exit the application.

Parameters:

code – Exit code.

gws.core.util.first(it)

Return the first element of an iterable.

Parameters:

it – An iterable.

Returns:

The first element, or None if the iterable is empty.

gws.core.util.first_not_none(*args)

Return the first argument that is not None.

Parameters:

*args – Values.

Returns:

The first value that is not None, or None.

gws.core.util.format_map(fmt: str, x: dict | Data, default: str = '') → str

Format a string with values from a dict or a Data object.

Parameters:

  • fmt – Format string with {name} placeholders.

  • x – A dict or Data object with the values.

  • default – Replacement for missing or None values.

Returns:

The formatted string.

gws.core.util.get(x, key, default=None)

Get a nested value or attribute from a structure.

Parameters:

  • x – A dict, list, Data or any object.

  • key – A list or a dot-separated string of nested keys. List elements are addressed by numeric keys.

  • default – The default value.

Returns:

The value if it exists and the default otherwise.

gws.core.util.get_app_global(name, init_fn)

Get a process-wide global value, creating it if needed.

Parameters:

  • name – Value name.

  • init_fn – Function without arguments that creates the value. It is called once, under an app lock.

Returns:

The value.

gws.core.util.get_cached_object(name: str, life_time: int, init_fn)

Get an object from the object cache, creating it if needed.

The object is pickled in const.OBJECT_CACHE_DIR and created under a server lock. A cached object older than life_time, or a falsy one, is created again. Load and store errors are logged and ignored.

Parameters:

  • name – Object name.

  • life_time – Life time in seconds.

  • init_fn – Function without arguments that creates the object.

Returns:

The object.

gws.core.util.get_ephemeral_content(name: str, init_fn) → bytes

Get bytes from the ephemeral content cache, creating them if needed.

The content is stored in the ephemeral directory and created under a server lock, so it is shared between processes until the ephemeral cleanup removes it.

Parameters:

  • name – Content name.

  • init_fn – Function without arguments that returns the content.

Returns:

The content.

gws.core.util.get_server_global(name: str, init_fn)

Get a server-wide global value, creating it if needed.

The value is kept in memory and pickled in const.GLOBALS_DIR, so other processes load it instead of creating it again. It is created under a server lock. Load and store errors are logged and ignored.

Parameters:

  • name – Value name.

  • init_fn – Function without arguments that creates the value.

Returns:

The value.

gws.core.util.has(x, key) → bool

Check if a nested value or attribute exists in a structure.

Parameters:

  • x – A dict, list, Data or any object.

  • key – A list or a dot-separated string of nested keys.

Returns:

True if the key exists.

gws.core.util.is_atom(x)

Check if the value is None or a scalar (number, bool, string or bytes).

Parameters:

x – A value.

Returns:

True if the value is an atom.

gws.core.util.is_bytes(x)

Check if the value is bytes or bytearray.

Parameters:

x – A value.

Returns:

True if the value is bytes or bytearray.

gws.core.util.is_data_object(x) → bool

Check if the argument is a Data object.

This is a placeholder, replaced by gws.is_data_object when gws is imported.

Parameters:

x – A value.

Returns:

True if the value is a Data object.

gws.core.util.is_dict(x)

Check if the value is a dict.

Parameters:

x – A value.

Returns:

True if the value is a dict.

gws.core.util.is_dir(path)

Check if the path is an existing directory.

Parameters:

path – Directory path.

Returns:

True if the directory exists.

gws.core.util.is_empty(x) → bool

Check if the value is empty.

A value is empty if it is None, has zero length, or is an object without attributes.

Parameters:

x – A value.

Returns:

True if the value is empty.

gws.core.util.is_file(path)

Check if the path is an existing file.

Parameters:

path – File path.

Returns:

True if the file exists.

gws.core.util.is_list(x)

Check if the value is a list or a tuple.

Parameters:

x – A value.

Returns:

True if the value is a list or a tuple.

gws.core.util.join_uid(parent_uid, object_uid)

Join a parent uid and an object uid with UID_DELIMITER.

If either uid is already joined, only its last part is used.

Parameters:

  • parent_uid – Parent uid.

  • object_uid – Object uid.

Returns:

The joined uid.

exception gws.core.util.LockBusyError

Bases: Exception

Raised when a server lock cannot be acquired within the timeout.

gws.core.util.merge(*args, **kwargs) → dict | Data

Create a new dict or Data object by merging values from dicts, Data objects and keyword args.

Later values override earlier ones, unless they are None.

Parameters:

  • *args – Dicts or Data objects. Empty values are skipped.

  • **kwargs – Keyword args, merged last.

Returns:

A new object of the same type as the first argument, or a dict if it is a dict or None.

gws.core.util.microtime() → int

Return the Unix time as an integer number of microseconds.

Returns:

Microseconds since the epoch.

gws.core.util.mstime() → int

Return the Unix time as an integer number of milliseconds.

Returns:

Milliseconds since the epoch.

gws.core.util.omit(x, *keys)

Return a copy of a dict or a Data object without the given keys.

Parameters:

  • x – A dict or Data.

  • *keys – Keys to remove.

Returns:

A new object of the same type, or an empty dict if x is of another type.

gws.core.util.parse_acl(acl)

Parse an ACL config into an ACL.

Parameters:

acl – An ACL config. Can be given as a string allow X, allow Y, deny Z, or as a list of dicts { role X type allow }, { role Y type deny }, or it can already be an ACL [1 X], [0 Y], or it can be None.

Returns:

Access list, empty for an empty value.

Raises:

ValueError – If the ACL config is invalid.

gws.core.util.pick(x, *keys)

Return a copy of a dict or a Data object with the given keys only.

Parameters:

  • x – A dict or Data.

  • *keys – Keys to keep.

Returns:

A new object of the same type, or an empty dict if x is of another type.

gws.core.util.pop(x, key, default=None)

Remove a key from a dict or a Data object and return its value.

Parameters:

  • x – A dict or Data.

  • key – The key.

  • default – Value to return if the key is missing or x is of another type.

Returns:

The value or the default.

gws.core.util.random_string(size: int) → str

Generate a random alphanumeric string.

Parameters:

size – String length.

Returns:

The string.

gws.core.util.read_file(path: str) → str

Read a UTF-8 text file.

Parameters:

path – File path.

Returns:

The file content.

Raises:

Exception – Errors from opening or reading the file are logged and raised again.

gws.core.util.read_file_b(path: str) → bytes

Read a binary file.

Parameters:

path – File path.

Returns:

The file content.

Raises:

Exception – Errors from opening or reading the file are logged and raised again.

gws.core.util.require(value: T | None, message: str = '') → T

Return the value if it is not None, otherwise raise an error.

Parameters:

  • value – A value.

  • message – Error message.

Returns:

The value.

Raises:

ValueError – If the value is None.

gws.core.util.serialize_to_path(obj, path)

Pickle an object to a file atomically.

Parameters:

  • obj – An object.

  • path – File path.

Returns:

The file path.

gws.core.util.server_lock(uid, timeout: float = 60)

Create an inter-process lock, to be used as a context manager.

The lock is acquired when the with block is entered.

Example:

with gws.u.server_lock('my_task', timeout=5):
    ...
Parameters:

  • uid – Lock identifier.

  • timeout – Seconds to wait for the lock. 0 means a single attempt.

Returns:

The lock object.

Raises:

LockBusyError – If the lock is not acquired within timeout.

gws.core.util.set_app_global(name, value)

Set a process-wide global value.

Parameters:

  • name – Value name.

  • value – The value.

Returns:

The value.

gws.core.util.sha256(x) → str

Compute the SHA-256 hash of a value.

Bytes, strings and numbers are hashed directly, other values are converted to JSON first, with sorted keys and Data objects as dicts.

Parameters:

x – A value.

Returns:

The hex digest.

gws.core.util.sleep(n: float)

Sleep for a number of seconds.

Parameters:

n – Seconds.

gws.core.util.split_uid(joined_uid: str) → tuple[str, str]

Split a joined uid at the first UID_DELIMITER.

Parameters:

joined_uid – Joined uid.

Returns:

A tuple of the parent uid and the object uid. If there is no delimiter, the object uid is empty.

gws.core.util.stime() → int

Return the Unix time as an integer number of seconds.

Returns:

Seconds since the epoch.

gws.core.util.strip(x)

Strip all strings and remove empty values from a collection.

Parameters:

x – A dict, Data or an iterable.

Returns:

A new dict or Data object, or a list for other iterables.

gws.core.util.T
gws.core.util.to_bytes(x, encoding='utf8') → bytes

Convert a value to bytes by converting it to a string and encoding it.

Parameters:

  • x – A value. Bytes are returned as is, None becomes empty bytes.

  • encoding – The encoding.

Returns:

Bytes.

gws.core.util.to_data_object(x)

Convert a value to a Data object.

This is a placeholder, replaced by gws.to_data_object when gws is imported.

Parameters:

x – A value.

Returns:

A Data object.

gws.core.util.to_dict(x) → dict

Convert a value to a dict.

Parameters:

x – A dict, None, a named tuple or an object.

Returns:

The dict itself, an empty dict for None, the fields of a named tuple, or the object’s vars.

Raises:

ValueError – If the value cannot be converted.

gws.core.util.to_float(x) → float

Convert a value to a float.

Parameters:

x – A value.

Returns:

The float value, or 0.0 if the conversion fails.

gws.core.util.to_int(x) → int

Convert a value to an int.

Parameters:

x – A value.

Returns:

The int value, or 0 if the conversion fails.

gws.core.util.to_json_value(x) → dict | list | str | int | float | bool | None

Recursively convert a value to a JSON serializable type.

Parameters:

x – A value. Dicts, lists and Data objects are converted recursively, other non-scalar values are converted with str.

Returns:

A JSON serializable value.

gws.core.util.to_lines(txt: str, comment: str = None) → list[str]

Convert a multiline string into a list of strings.

Parameters:

  • txt – A string.

  • comment – Comment marker. If given, everything from the marker to the end of the line is removed.

Returns:

A list of stripped, non-empty lines.

gws.core.util.to_list(x, delimiter: str = ',') → list

Convert a value to a list.

Parameters:

  • x – A value. A string (or bytes) is split by the delimiter, the parts are stripped and empty parts removed. A number or a bool becomes a one-element list, other iterables are converted to a list.

  • delimiter – The delimiter. If empty, a string becomes a one-element list.

Returns:

A list. Empty values and values that cannot be converted give an empty list.

gws.core.util.to_lower_dict(x) → dict

Convert a value to a dict with lower-case keys.

Parameters:

x – A value accepted by to_dict.

Returns:

A new dict.

gws.core.util.to_rounded_int(x) → int

Round a float and convert a value to an int.

Parameters:

x – A value.

Returns:

The int value, or 0 if the conversion fails.

gws.core.util.to_str(x, encodings: list[str] = None) → str

Convert a value to a string.

Bytes are decoded, None becomes an empty string, other values are converted with str.

Parameters:

  • x – A value.

  • encodings – A list of acceptable encodings. If the value is bytes, try each encoding and return the first result that decodes without errors. If none succeeds, decode as UTF-8 and ignore errors.

Returns:

A string.

gws.core.util.to_uid(x) → str

Convert a value to a uid.

The value is converted to a lower-case string, German umlauts are transliterated, and runs of other characters are replaced with underscores.

Parameters:

x – A value.

Returns:

A string of a-z, 0-9 and _, or an empty string for an empty value.

gws.core.util.to_upper_dict(x) → dict

Convert a value to a dict with upper-case keys.

Parameters:

x – A value accepted by to_dict.

Returns:

A new dict.

gws.core.util.UID_DELIMITER = '::'
gws.core.util.uniq(x)

Remove duplicate elements from a collection, keeping the order.

Parameters:

x – An iterable.

Returns:

A list of unique elements.

gws.core.util.unserialize_from_path(path)

Load a pickled object from a file.

Parameters:

path – File path.

Returns:

The object.

gws.core.util.utime() → float

Return the Unix time as a float number.

Returns:

Seconds since the epoch.

gws.core.util.write_debug_file(path: str, s: str | bytes)

Write a file to the debug directory <VAR_DIR>/debug.

Errors are logged and ignored.

Parameters:

  • path – File path, relative to the debug directory.

  • s – Content. A string is encoded as UTF-8.

gws.core.util.write_file(path: str, s: str, user: int = None, group: int = None)

Write a text file atomically, via a temporary file in the same directory.

Parameters:

  • path – File path.

  • s – Text content, encoded as UTF-8.

  • user – File owner, defaults to const.UID.

  • group – File group, defaults to const.GID.

Returns:

The file path.

Raises:

Exception – Write errors are logged and raised again.

gws.core.util.write_file_b(path: str, s: str | bytes, user: int = None, group: int = None)

Write a binary file atomically, via a temporary file in the same directory.

Parameters:

  • path – File path.

  • s – Content. A string is encoded as UTF-8.

  • user – File owner, defaults to const.UID.

  • group – File group, defaults to const.GID.

Returns:

The file path.

Raises:

Exception – Write errors are logged and raised again.