module documentation

Utility functions and classes.

Class ErrorCapturingThread Subclass of Thread that captures exceptions from the target function.
Class VersionData A BaseModel containing the information about the server version.
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 requires_lock Decorate a class method so that it requires the class lock to run.
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 T 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.
Function _wrap_and_catch_errors Run target function in a try-except block.
Constant _NAME_UNSAFE_PATTERN Undocumented
Constant _POSIX_UNSAFE_PATTERN Undocumented
Constant _WINDOWS_UNSAFE_PATTERN Undocumented
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:strThe original name string to sanitise.
Returns
strA 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:strThe original path string to sanitise.
Returns
strA version of the input string safe to use as a file path.
def requires_lock(method: Callable[P, T]) -> Callable[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 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:

  1. 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.
  2. 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.
  3. 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.
  4. 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
VersionDataa VersionData instance with the version string and source.
COMMIT_REGEX = (source)

Undocumented

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

Undocumented

Value
logging.getLogger(__name__)

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__)))

Undocumented

Value
TypeVar('T')
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:strThe path to the .git directory.
Returns
strThe 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:strThe full path to the ref file in the .git directory.
Returns
strThe 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:strThe path to the toml file.
Returns
strThe 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): (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.

def _wrap_and_catch_errors(target, error_buffer, *args, **kwargs): (source)

Run target function in a try-except block.

This function is designed only to be used by ErrorCapturingThread.

It will run a target function in a try-except block catching any exception. If an exception is caught it is added to error_buffer.

Parameters
targetThe target function to call
error_bufferAn empty list that is used for returning the exception from the thread. It is a list to ensure it is passed by reference.
*argsThe arguments for the target function
**kwargsThe Keyword arguments for the target function
_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_\.-:/\\]')