"""
Configuration parsing and XML validation utilities for ClinVar Build.
This module provides utilities for parsing configuration files, and managing
logging output for long-running operations. It includes classes for handling
block-based configuration files and property management with controlled access.
"""
import os
import sys
import warnings
import logging
from pathlib import Path
from clinvar_build.constants import (
UtilsConfigData as ConfigNames,
_CONFIG_DIR,
)
from clinvar_build.errors import (
is_type,
)
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
[docs]
class ManagedProperty(object):
"""
A generic property factory defining setters and getters, with optional
type validation.
Parameters
----------
name : `str`
The name of the setters and getters
types: `Type`, default `NoneType`
Either a single type, or a tuple of types to test against.
Methods
-------
enable_setter()
Enables the setter for the property, allowing attribute assignment.
disable_setter()
Disables the setter for the property, making the property read-only.
set_with_setter(instance, value)
Enables the setter, sets the property value, and then disables
the setter, ensuring controlled updates.
Returns
-------
property
A property object with getter and setter.
"""
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
[docs]
def __init__(self, name: str, types: tuple[type] | type | None = None):
"""
Initialize the ManagedProperty.
"""
self.name = name
self.types = types
self._setter_enabled = True
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
# NOTE owner is part of the descriptor protocol for __get__, leave it
[docs]
def __get__(self, instance, owner):
"""Getter for the property."""
if instance is None:
return self
return instance.__dict__.get(self.name)
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
[docs]
def __set__(self, instance, value):
"""Setter for the property."""
owner = type(instance)
if not self._setter_enabled:
raise AttributeError(f"The property '{self.name}' on "
f"{owner.__name__} is read-only.")
if self.types and not isinstance(value, self.types):
raise ValueError(
f"Expected any of {self.types}, got {type(value)} "
f"for property '{self.name}'."
)
instance.__dict__[self.name] = value
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
[docs]
def enable_setter(self):
"""Enable the setter for the property."""
self._setter_enabled = True
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
[docs]
def disable_setter(self):
"""Disable the setter for the property."""
self._setter_enabled = False
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
[docs]
def set_with_setter(self, instance, value):
"""
Enable the setter, set the property value, and then disable the setter.
Parameters
----------
instance : `object`
The instance on which the property is being set.
value : `any`
The value to assign to the property.
"""
try:
self.enable_setter()
setattr(instance, self.name, value)
finally:
self.disable_setter()
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
[docs]
def check_environ(environ_variable:str=ConfigNames.config_dir,
fall_back:str | Path | None = _CONFIG_DIR) -> str:
"""
Retrieve an environment variable pointing to a directory path, with
optional fallback path.
Attempts to retrieve the specified environment variable. If the
variable is not set, the function will attempt to use the fallback
path if provided. This is useful for configuration management where
environment variables may not always be explicitly set.
Parameters
----------
environ_variable : `str`
The name of the environment variable to retrieve.
fall_back : `str`, `Path` or `None`
A fallback path to return if the environment variable is not set. If
None, an error will be raised when the environment variable is missing.
Returns
-------
str
A directory path.
Raises
------
KeyError
Raised when the environment variable is not set and
fall_back is None.
TypeError
Raised when environ_variable is not of type str or
fall_back is not of type str, Path, or None.
Notes
-----
The function will not check whether the path is available or whether
permissions allow for read or write access
Warnings
--------
UserWarning
Issued when the environment variable is not set and the
fallback path is used instead.
Examples
--------
>>> check_environ("MY_VAR")
'/path/to/default/config'
"""
# check input
is_type(environ_variable, str)
is_type(fall_back, (str, Path, type(None)))
# check if environ_variable
try:
res = os.environ[environ_variable]
except KeyError as e:
if fall_back is not None:
res = fall_back
# print warning
warnings.warn(
'The environmental variable `{}` is not set. Trying '
'to recover using the default configuration files in '
'{}.'.format(environ_variable, fall_back), Warning)
else:
raise e
# return
return res
# @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
# Custom logging handler for in-place progress updates
[docs]
class ProgressHandler(logging.StreamHandler):
"""
Custom handler that updates progress in place.
Uses ANSI escape codes to overwrite previous output instead of
printing new lines. Useful for progress updates during long-running
operations.
Parameters
----------
stream : `file-like object`, optional
Output stream. Defaults to sys.stdout.
Attributes
----------
_last_line_count : `int`
Number of lines in the previous message, used to calculate
how far to move the cursor up.
Examples
--------
>>> progress_logger = logging.getLogger('progress')
>>> handler = ProgressHandler()
>>> handler.setFormatter(logging.Formatter('%(asctime)s - %(message)s'))
>>> progress_logger.addHandler(handler)
>>> progress_logger.info("Processing: 100 records")
>>> progress_logger.info("Processing: 200 records") # Overwrites previous
"""
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
def __init__(self, stream=None):
if stream is None:
stream = sys.stdout
super().__init__(stream)
self._last_line_count = 0
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
[docs]
def emit(self, record: logging.LogRecord) -> None:
"""
Emit a log record with in-place updating.
Parameters
----------
record : `logging.LogRecord`
The log record to emit
"""
try:
msg = self.format(record)
lines = msg.split('\n')
# Move cursor up and clear previous lines
if self._last_line_count > 0:
self.stream.write(f"\033[{self._last_line_count}A")
for _ in range(self._last_line_count):
self.stream.write("\033[K\n")
self.stream.write(f"\033[{self._last_line_count}A")
# Write new message
self.stream.write(msg + '\n')
self.stream.flush()
# Remember line count for next update
self._last_line_count = len(lines)
except Exception:
self.handleError(record)
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
[docs]
def reset(self) -> None:
"""
Reset line count.
Call this when switching from in-place updates to normal logging
to prevent cursor position issues.
"""
self._last_line_count = 0