Utils

pyatlan.utils

Classes

API(path: str, method: HTTPMethod, expected_status: int, endpoint: EndPoint, consumes: str = APPLICATION_JSON, produces: str = APPLICATION_JSON)

Source code in pyatlan/utils.py

def __init__(
    self,
    path: str,
    method: HTTPMethod,
    expected_status: int,
    endpoint: EndPoint,
    consumes: str = APPLICATION_JSON,
    produces: str = APPLICATION_JSON,
):
    self.path = path
    self.method = method
    self.expected_status = expected_status
    self.consumes = consumes
    self.produces = produces
    self.endpoint: EndPoint = endpoint

Functions

multipart_urljoin(base_path, *path_elems) staticmethod

Join a base path and multiple context path elements. Handle single leading and trailing / characters transparently.

Parameters:

Name	Type	Description	Default
base_path	string	the base path or url (ie. http://atlas/v2/)	required
*path_elems	string	multiple relative path elements (ie. /my/relative, /path)	()

Returns:

Name	Type	Description
string		the result of joining the base_path with the additional path elements

Source code in pyatlan/utils.py

@staticmethod
def multipart_urljoin(base_path, *path_elems):
    """Join a base path and multiple context path elements. Handle single
    leading and trailing `/` characters transparently.

    Args:
        base_path (string): the base path or url (ie. `http://atlas/v2/`)
        *path_elems (string): multiple relative path elements (ie. `/my/relative`, `/path`)

    Returns:
        string: the result of joining the base_path with the additional path elements
    """

    def urljoin_pair(left, right):
        return "/".join([left.rstrip("/"), right.strip("/")])

    return reduce(urljoin_pair, path_elems, base_path)

AuthorizationFilter

Bases: Filter

A Filter that will replace the authorization header with the text 'REDACTED'

ContextVarWrapper(contextvar: ContextVar, key_name: str)

Bases: Mapping

This class implements the Mapping protocol on a ContextVar. This allows 'extra' information needed for a LogAdaptor to be obtained from a ContextVar.

Create the ContextVarWrapper :param contextvar: the ContextVar that will provide the data :param key_name: the name that should be used to obtain a value from the contextvar

Source code in pyatlan/utils.py

def __init__(self, contextvar: ContextVar, key_name: str):
    """
    Create the ContextVarWrapper
    :param contextvar: the ContextVar that will provide the data
    :param key_name: the name that should be used to obtain a value from the contextvar
    """
    self.contextvar = contextvar
    self.key_name = key_name

JsonFormatter

Bases: Formatter

A custom JSON log formatter

Functions

format(record: logging.LogRecord) -> str

Format the log record

:param record: The log record to be formatted :returns: The formatted log message

Source code in pyatlan/utils.py

def format(self, record: logging.LogRecord) -> str:
    """
    Format the log record

    :param record: The log record to be formatted
    :returns: The formatted log message
    """
    log_record = {
        "asctime": self.formatTime(record, self.datefmt),
        "name": record.name,
        "levelname": record.levelname,
        "message": (
            record.msg if isinstance(record.msg, dict) else record.getMessage()
        ),
    }
    return json.dumps(log_record, ensure_ascii=False)

RequestIdAdapter(logger: logging.Logger, contextvar: ContextVar)

Bases: LoggerAdapter

This is a LoggerAdapter that can be used to provide a reqquestid from a ContextVar

Create the LoggerAdapter the will get the value to be used for 'requestid' from the given ContextVar :param logger: the Logger to wrap :param contextvar: the ContextVar from which to obtain the value to be used for 'requestid'

Source code in pyatlan/utils.py

def __init__(self, logger: logging.Logger, contextvar: ContextVar):
    """
    Create the LoggerAdapter the will get the value to be used for 'requestid'  from the given ContextVar
    :param logger: the Logger to wrap
    :param contextvar: the ContextVar from which to obtain the value to be used for 'requestid'

    """
    super().__init__(
        logger, ContextVarWrapper(contextvar=contextvar, key_name=REQUESTID)
    )

RequestIdFilter

Bases: Filter

A filter that will add requestid to the LogRecord if it is not already present. This is to support adding the requestid to the logging formatter

Functions

deep_get(dictionary, keys, default=None)

Returns dict key value using dict and it's dot_key string,

ie: key1.key2_nested.key3_nested, if found, otherwise returns default (None).

Source code in pyatlan/utils.py

def deep_get(dictionary, keys, default=None):
    """
    Returns dict key value using dict and it's dot_key string,

    ie: key1.key2_nested.key3_nested, if found, otherwise returns default (`None`).
    """
    return reduce(
        lambda d, key: d.get(key, default) if isinstance(d, dict) else default,
        keys.split("."),
        dictionary,
    )

get_parent_qualified_name(qualified_name: str) -> str

Returns qualified name of the parent asset :param qualified_name: qualified of the asset

Source code in pyatlan/utils.py

def get_parent_qualified_name(qualified_name: str) -> str:
    """
    Returns qualified name of the parent asset
    :param qualified_name: qualified of the asset
    """
    qn = qualified_name.split("/")
    return "/".join(qn[:-1])

get_python_version() -> str

Get the current Python version as a string.

Returns:

Name	Type	Description
str	str	Python version in format "major.minor.micro" or "unknown" if unavailable

Source code in pyatlan/utils.py

def get_python_version() -> str:
    """
    Get the current Python version as a string.

    Returns:
        str: Python version in format "major.minor.micro" or "unknown" if unavailable
    """
    try:
        return f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}"
    except Exception:
        return "unknown"

init_guid(func)

Decorator function that can be used on the Create method of an asset to initialize the guid.

Source code in pyatlan/utils.py

def init_guid(func):
    """Decorator function that can be used on the Create method of an asset to initialize the guid."""

    @wraps(func)
    def call(*args, **kwargs):
        ret_value = func(*args, **kwargs)
        if hasattr(ret_value, "guid"):
            ret_value.guid = str(
                -int(random.random() * 10000000000000000)  # noqa: S311
            )
        return ret_value

    return call

select_optional_set_fields(params: Dict[str, Any]) -> Dict

Filter the provided parameters to include only those fields that are not set to None.

Source code in pyatlan/utils.py

def select_optional_set_fields(params: Dict[str, Any]) -> Dict:
    """
    Filter the provided parameters to include
    only those fields that are not set to `None`.
    """
    return {key: value for key, value in params.items() if value is not None}

validate_type(name: str, _type, value)

Validate that the given value is of the specified type.

:param name: the name of the variable to be used in the error message :param _type: the type of the variable to be validated :param value: the value to be validated that it is of the specified type

Source code in pyatlan/utils.py

def validate_type(name: str, _type, value):
    """
    Validate that the given value is of the specified type.

    :param name: the name of the variable to be used in the error message
    :param _type: the type of the variable to be validated
    :param value: the value to be validated that it is of the specified type
    """
    if _type is int:
        if isinstance(value, _type) and not isinstance(value, bool):
            return
    elif isinstance(_type, tuple):
        if None in _type and value is None:
            return
        if any(isinstance(value, t) for t in _type if t is not None):
            return
    elif _type is None and value is None:
        return
    elif isinstance(value, _type):
        return

    # Construct the type name string, handling None explicitly
    if isinstance(_type, tuple):
        type_names = [t.__name__ if t is not None else "None" for t in _type]
        type_name = (
            ", ".join(type_names[:-1]) + f" or {type_names[-1]}"
            if len(type_names) > 1
            else type_names[0]
        )
    else:
        type_name = _type.__name__ if _type is not None else "None"

    raise ErrorCode.INVALID_PARAMETER_TYPE.exception_with_parameters(name, type_name)