openflexure_microscope_server.utilities

module documentation

(source)

Utility functions and classes.

Class	`VersionData`	A BaseModel containing the information about the server version.
Function	`coerce_thing_selector`	Return a valid selector for a mapping of things or None if empty.
Function	`load_patched_config`	Load a json configuration file, if it a patch, return the full config.
Function	`make_name_safe`	Replace unsafe characters in a filename or identifier with underscores.
Function	`make_path_safe`	Replace unsafe characters in a file path with underscores, preserving separators.
Function	`merge_patch`	Merge json data using the methods from IETF RFC 7396.
Function	`requires_lock`	Decorate a class method so that it requires the class lock to run.
Function	`resolve_path_from_dir`	Convert a relative or abs path specified in one dir to the working dir.
Function	`robust_version_strings`	Return a version string and information on its source.
Constant	`COMMIT_REGEX`	Undocumented
Constant	`LOGGER`	Undocumented
Constant	`P`	Undocumented
Constant	`REF_REGEX`	Undocumented
Constant	`REPO_DIR`	Undocumented
Type Variable	`LockableClass`	Undocumented
Type Variable	`T`	Undocumented
Type Alias	`JSONScalar`	Undocumented
Type Alias	`JSONType`	Undocumented
Function	`_get_hash_from_git_dir`	Get the commit hash from a .git directory directly.
Function	`_get_hash_from_git_ref`	Get the commit hash from a ref in the .git directory.
Function	`_get_version_from_toml`	Get the version string from a pyproject.toml file.
Function	`_is_lock_like`	Check if an object is a lock.
Constant	`_NAME_UNSAFE_PATTERN`	Undocumented
Constant	`_POSIX_UNSAFE_PATTERN`	Undocumented
Constant	`_WINDOWS_UNSAFE_PATTERN`	Undocumented

def coerce_thing_selector(thing_mapping: Mapping[str, lt.Thing], selected: str | None, default: str | None = None) -> str | None: (source) ¶

Return a valid selector for a mapping of things or None if empty.

Parameters
thing_mapping:`Mapping[str, lt.Thing]`	A mapping of str -> `Thing`
selected:`str \| None`	The key of the selected thing (or `None`)
default:`str \| None`	The default key to fallback to if selected is `None` or the key is missing

Returns
`str \| None`	`None` if the mapping is empty, else a key that is in the mapping. Order of preference is: selected, default, the first key.

def load_patched_config(config_path: str) -> dict: (source) ¶

Load a json configuration file, if it a patch, return the full config.

Load a json file. If it does not contains the "base_config_file" key then return the data as read.

If the configuration specifies a "base_config_file" but no "patch" then return the data from reading base_config_file as json.

If the configuration specifies a "base_config_file" and "patch" then apply the patch to the data in base_config_file and return the result. The patching method is merge_patch()

Parameters
config_path:`str`	The path of the configuration file.

Returns
`dict`	The contents of the configuration file after loading and patching the specified base configuration file if applicable.

def make_name_safe(unsafe_name_string: str) -> str: (source) ¶

Replace unsafe characters in a filename or identifier with underscores.

This excludes all path separators, ensuring the result is safe to use as a standalone filename or identifier component.

Parameters
unsafe_name_string:`str`	The original name string to sanitise.

Returns
`str`	A version of the input string safe to use as a file name or identifier.

def make_path_safe(unsafe_path_string: str) -> str: (source) ¶

Replace unsafe characters in a file path with underscores, preserving separators.

This can be used to check that user inputs have not been concatenated into an unsafe filepath.

This ensures compatibility across platforms by preserving only valid characters, including path separators (forward slash on POSIX, and forward slash/backslash/colon on Windows).

Parameters
unsafe_path_string:`str`	The original path string to sanitise.

Returns
`str`	A version of the input string safe to use as a file path.

@overload

def merge_patch(target: dict[str, JSONType], patch: dict[str, JSONType], enforce_dict: Literal[True]) -> dict[str, JSONType]:

@overload

def merge_patch(target: JSONType, patch: JSONType, enforce_dict: Literal[False]) -> JSONType:

@overload

def merge_patch(target: JSONType, patch: JSONType) -> JSONType:

(source) ¶

Merge json data using the methods from IETF RFC 7396.

Primarily this is designed to merge dictionaries, but IETF RFC 7396 provides defined methods for handling non-dictionary data loaded from JSON, so this has been provided in full.

Parameters
target:`JSONType`	The target object
patch:`JSONType`	The patch to be applied
enforce_dict:`bool`	Boolean, set True enfoces that the target and patch are both dictionaries.

Returns
`JSONType`	Undocumented

def requires_lock(method: Callable[Concatenate[LockableClass, P], T]) -> Callable[Concatenate[LockableClass, P], T]: (source) ¶

Decorate a class method so that it requires the class lock to run.

The class should have a reentrant lock with the name self._lock.

def resolve_path_from_dir(path: str, directory: str) -> str: (source) ¶

Convert a relative or abs path specified in one dir to the working dir.

This also expands any environment variables.

Parameters
path:`str`	The specified path (could be absolute or relative)
directory:`str`	The directory from which the path was specified

Returns
`str`	The normalised path.

def robust_version_strings() -> VersionData: (source) ¶

Return a version string and information on its source.

Returning a python version string is not without problems. If the package is installed in an editable way, often METADATA is never updated. This provides 4 ways to provide a version string:

If both a Git directory and a pyproject.toml are available. Return the version

string from the toml file and the git hash from the .git directory as its source.
If a pyproject.toml is available but no Git directory then this is likely an

installation from source. Return the version string from the toml file and return the literal string "TOML" as its source.
If a Git directory is available but no pyproject.toml then something is very

weird - log an error. Then return "Undefined" as the version string, but still return the Git hash as the source.
Finally if neither a Git directory nor a pyproject.toml are available then the server has been installed from a distribution package (i.e. a wheel). In this case use the standard library function importlib.metadata.version for the version string (prepending a "v") and return "Dist" as its source.

Returns
`VersionData`	a VersionData instance with the version string and source.

COMMIT_REGEX = (source) ¶

Undocumented

Value

re.compile(r'[0-9a-f]{40}')

LOGGER = (source) ¶

Undocumented

Value

logging.getLogger(__name__)

P = (source) ¶

Undocumented

Value

ParamSpec('P')

REF_REGEX = (source) ¶

Undocumented

Value

re.compile(r'^ref:\s(.*)$')

REPO_DIR = (source) ¶

Undocumented

Value

os.path.dirname(os.path.dirname(os.path.dirname(__file__)))

LockableClass = (source) ¶

Undocumented

Value

TypeVar('LockableClass')

T = (source) ¶

Undocumented

Value

TypeVar('T')

JSONScalar: TypeAlias = (source) ¶

Undocumented

Value

str | int | float | bool | None

JSONType: TypeAlias = (source) ¶

Undocumented

Value

dict[str, JSONType] | list[JSONType] | JSONScalar

def _get_hash_from_git_dir(git_dir_path: str) -> str: (source) ¶

Get the commit hash from a .git directory directly.

This function doesn't rely on GIT being on the PATH or even installed. It doesn't require any packages outside the Python standard library. It directly inspects the .git directory to get the commit hash:

If the repository is on a detached HEAD then the hash will be in the file

.git/HEAD. A detached HEAD is could be from checking out a tag or checking out a commit directly.
In normal operation .git/HEAD points to a reference or "ref". This ref file

contains the hash.

Parameters
git_dir_path:`str`	The path to the .git directory.

Returns
`str`	The hash, or "Undefined" if there is any error, errors are logged not raised.

def _get_hash_from_git_ref(git_ref_path: str) -> str: (source) ¶

Get the commit hash from a ref in the .git directory.

For more detail see _get_hash_from_git_dir

Parameters
git_ref_path:`str`	The full path to the ref file in the .git directory.

Returns
`str`	The hash, or "Undefined" if there is any error, errors are logged not raised.

def _get_version_from_toml(toml_path: str) -> str: (source) ¶

Get the version string from a pyproject.toml file.

Parameters
toml_path:`str`	The path to the toml file.

Returns
`str`	The version string directly as reported in the toml file, or "Undefined" if there is any error, errors are logged not raised.

def _is_lock_like(obj: Any) -> bool: (source) ¶

Check if an object is a lock.

Cannot use isinstance(obj, threading.RLock), may be possible to use isinstance(obj, threading._RLock). But this uses a private method and may break. Instead making the check that both "acquire" and "release" exist.

_NAME_UNSAFE_PATTERN = (source) ¶

Undocumented

Value

re.compile(r'[^a-zA-Z0-9_\.-]')

_POSIX_UNSAFE_PATTERN = (source) ¶

Undocumented

Value

re.compile(r'[^a-zA-Z0-9_\.-/]')

_WINDOWS_UNSAFE_PATTERN = (source) ¶

Undocumented

Value

re.compile(r'[^a-zA-Z0-9_\.-:/\\]')